774 lines
27 KiB
Python
774 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
EU-Utility Integration Test Report
|
|
Generated by Integration Tester Agent
|
|
|
|
This script performs comprehensive integration testing of:
|
|
1. All 12 core services
|
|
2. Cross-plugin communication via Event Bus and Plugin API
|
|
3. All 23+ plugins loading
|
|
4. Global hotkeys configuration
|
|
5. System tray integration
|
|
6. External API calls (Nexus API, HTTP Client)
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from typing import Dict, List, Any, Optional
|
|
import traceback
|
|
|
|
# Add project root to path
|
|
project_root = Path(__file__).parent.parent
|
|
if str(project_root) not in sys.path:
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
# Test Results Storage
|
|
class TestResult:
|
|
def __init__(self, name: str, category: str):
|
|
self.name = name
|
|
self.category = category
|
|
self.status = "PENDING"
|
|
self.message = ""
|
|
self.duration_ms = 0
|
|
self.error = None
|
|
|
|
class IntegrationTestReport:
|
|
def __init__(self):
|
|
self.results: List[TestResult] = []
|
|
self.start_time = datetime.now()
|
|
|
|
def add_result(self, result: TestResult):
|
|
self.results.append(result)
|
|
|
|
def get_summary(self) -> Dict[str, Any]:
|
|
total = len(self.results)
|
|
passed = len([r for r in self.results if r.status == "PASS"])
|
|
failed = len([r for r in self.results if r.status == "FAIL"])
|
|
skipped = len([r for r in self.results if r.status == "SKIP"])
|
|
pending = len([r for r in self.results if r.status == "PENDING"])
|
|
|
|
return {
|
|
"total": total,
|
|
"passed": passed,
|
|
"failed": failed,
|
|
"skipped": skipped,
|
|
"pending": pending,
|
|
"pass_rate": f"{(passed/max(total-skipped, 1))*100:.1f}%"
|
|
}
|
|
|
|
def print_report(self):
|
|
print("\n" + "=" * 80)
|
|
print("EU-UTILITY INTEGRATION TEST REPORT")
|
|
print("=" * 80)
|
|
print(f"Generated: {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print(f"Project: {project_root}")
|
|
print("=" * 80)
|
|
|
|
# Group by category
|
|
categories = {}
|
|
for result in self.results:
|
|
if result.category not in categories:
|
|
categories[result.category] = []
|
|
categories[result.category].append(result)
|
|
|
|
for category, results in sorted(categories.items()):
|
|
print(f"\n### {category.upper()} ###")
|
|
for r in results:
|
|
icon = "✅" if r.status == "PASS" else "❌" if r.status == "FAIL" else "⏭️" if r.status == "SKIP" else "⏳"
|
|
print(f" {icon} {r.name:.<50} {r.status}")
|
|
if r.message:
|
|
print(f" └─ {r.message}")
|
|
if r.error:
|
|
print(f" └─ Error: {str(r.error)[:100]}")
|
|
|
|
summary = self.get_summary()
|
|
print("\n" + "=" * 80)
|
|
print("SUMMARY")
|
|
print("=" * 80)
|
|
print(f" Total Tests: {summary['total']}")
|
|
print(f" Passed: {summary['passed']}")
|
|
print(f" Failed: {summary['failed']}")
|
|
print(f" Skipped: {summary['skipped']}")
|
|
print(f" Pass Rate: {summary['pass_rate']}")
|
|
print("=" * 80)
|
|
|
|
report = IntegrationTestReport()
|
|
|
|
# ============================================================================
|
|
# TEST 1: CORE SERVICES
|
|
# ============================================================================
|
|
|
|
def test_core_services():
|
|
"""Test all 12 core services."""
|
|
|
|
# 1. Plugin API
|
|
result = TestResult("Plugin API Initialization", "Core Services")
|
|
try:
|
|
from core.plugin_api import get_api, PluginAPI, APIType
|
|
api = get_api()
|
|
if api is not None and isinstance(api, PluginAPI):
|
|
result.status = "PASS"
|
|
result.message = "Plugin API singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Plugin API not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 2. Event Bus
|
|
result = TestResult("Event Bus Initialization", "Core Services")
|
|
try:
|
|
from core.event_bus import get_event_bus, EventBus
|
|
bus = get_event_bus()
|
|
if bus is not None and isinstance(bus, EventBus):
|
|
result.status = "PASS"
|
|
result.message = "Event Bus singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Event Bus not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 3. Data Store
|
|
result = TestResult("Data Store Initialization", "Core Services")
|
|
try:
|
|
from core.data_store import get_data_store, DataStore
|
|
store = get_data_store()
|
|
if store is not None and isinstance(store, DataStore):
|
|
result.status = "PASS"
|
|
result.message = "Data Store singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Data Store not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 4. HTTP Client
|
|
result = TestResult("HTTP Client Initialization", "Core Services")
|
|
try:
|
|
from core.http_client import get_http_client, HTTPClient
|
|
client = get_http_client()
|
|
if client is not None and isinstance(client, HTTPClient):
|
|
result.status = "PASS"
|
|
result.message = "HTTP Client singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "HTTP Client not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 5. Nexus API
|
|
result = TestResult("Nexus API Initialization", "Core Services")
|
|
try:
|
|
from core.nexus_api import get_nexus_api, NexusAPI
|
|
nexus = get_nexus_api()
|
|
if nexus is not None and isinstance(nexus, NexusAPI):
|
|
result.status = "PASS"
|
|
result.message = "Nexus API singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Nexus API not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 6. Task Manager
|
|
result = TestResult("Task Manager Initialization", "Core Services")
|
|
try:
|
|
from core.tasks import get_task_manager, TaskManager
|
|
tasks = get_task_manager()
|
|
if tasks is not None and isinstance(tasks, TaskManager):
|
|
result.status = "PASS"
|
|
result.message = "Task Manager singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Task Manager not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 7. Audio Manager
|
|
result = TestResult("Audio Manager Initialization", "Core Services")
|
|
try:
|
|
from core.audio import get_audio_manager, AudioManager
|
|
audio = get_audio_manager()
|
|
if audio is not None and isinstance(audio, AudioManager):
|
|
result.status = "PASS"
|
|
result.message = "Audio Manager singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Audio Manager not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 8. Clipboard Manager
|
|
result = TestResult("Clipboard Manager Initialization", "Core Services")
|
|
try:
|
|
from core.clipboard import get_clipboard_manager, ClipboardManager
|
|
clipboard = get_clipboard_manager()
|
|
if clipboard is not None and isinstance(clipboard, ClipboardManager):
|
|
result.status = "PASS"
|
|
result.message = "Clipboard Manager singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Clipboard Manager not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 9. Log Reader
|
|
result = TestResult("Log Reader Initialization", "Core Services")
|
|
try:
|
|
from core.log_reader import get_log_reader, LogReader
|
|
log_reader = get_log_reader()
|
|
if log_reader is not None and isinstance(log_reader, LogReader):
|
|
result.status = "PASS"
|
|
result.message = "Log Reader singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Log Reader not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 10. OCR Service
|
|
result = TestResult("OCR Service Initialization", "Core Services")
|
|
try:
|
|
from core.ocr_service import get_ocr_service, OCRService
|
|
ocr = get_ocr_service()
|
|
if ocr is not None and isinstance(ocr, OCRService):
|
|
result.status = "PASS"
|
|
result.message = "OCR Service singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "OCR Service not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 11. Screenshot Service
|
|
result = TestResult("Screenshot Service Initialization", "Core Services")
|
|
try:
|
|
from core.screenshot import get_screenshot_service, ScreenshotService
|
|
screenshot = get_screenshot_service()
|
|
if screenshot is not None and isinstance(screenshot, ScreenshotService):
|
|
result.status = "PASS"
|
|
result.message = "Screenshot Service singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Screenshot Service not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# 12. Window Manager
|
|
result = TestResult("Window Manager Initialization", "Core Services")
|
|
try:
|
|
from core.window_manager import get_window_manager, WindowManager
|
|
wm = get_window_manager()
|
|
if wm is not None and isinstance(wm, WindowManager):
|
|
result.status = "PASS"
|
|
result.message = "Window Manager singleton working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Window Manager not properly initialized"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# ============================================================================
|
|
# TEST 2: CROSS-PLUGIN COMMUNICATION
|
|
# ============================================================================
|
|
|
|
def test_cross_plugin_communication():
|
|
"""Test cross-plugin communication via Event Bus and Plugin API."""
|
|
|
|
# Test Event Bus - Publish/Subscribe
|
|
result = TestResult("Event Bus Publish/Subscribe", "Cross-Plugin Communication")
|
|
try:
|
|
from core.event_bus import get_event_bus, SkillGainEvent, LootEvent
|
|
bus = get_event_bus()
|
|
|
|
received_events = []
|
|
def callback(event):
|
|
received_events.append(event)
|
|
|
|
# Subscribe
|
|
sub_id = bus.subscribe(callback, event_types=[SkillGainEvent])
|
|
|
|
# Publish
|
|
test_event = SkillGainEvent(skill_name="Test", skill_value=10.0, gain_amount=0.1)
|
|
bus.publish_sync(test_event)
|
|
|
|
if len(received_events) > 0:
|
|
result.status = "PASS"
|
|
result.message = "Event publishing and subscription working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Event not received by subscriber"
|
|
|
|
bus.unsubscribe(sub_id)
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# Test Event Bus - Typed Events
|
|
result = TestResult("Event Bus Typed Events", "Cross-Plugin Communication")
|
|
try:
|
|
from core.event_bus import get_event_bus, LootEvent
|
|
bus = get_event_bus()
|
|
|
|
received = []
|
|
def typed_callback(event):
|
|
received.append(event)
|
|
|
|
sub_id = bus.subscribe_typed(LootEvent, typed_callback)
|
|
|
|
test_event = LootEvent(mob_name="TestMob", items=[{"name": "TestItem"}])
|
|
bus.publish_sync(test_event)
|
|
|
|
if len(received) > 0 and isinstance(received[0], LootEvent):
|
|
result.status = "PASS"
|
|
result.message = "Typed event subscription working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Typed event not received correctly"
|
|
|
|
bus.unsubscribe(sub_id)
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# Test Event Bus - Event Filtering
|
|
result = TestResult("Event Bus Filtering", "Cross-Plugin Communication")
|
|
try:
|
|
from core.event_bus import get_event_bus, DamageEvent
|
|
bus = get_event_bus()
|
|
|
|
received = []
|
|
def filter_callback(event):
|
|
received.append(event)
|
|
|
|
# Subscribe with filter (min_damage=100)
|
|
sub_id = bus.subscribe_typed(DamageEvent, filter_callback, min_damage=100)
|
|
|
|
# Publish low damage (should not be received)
|
|
low_damage = DamageEvent(damage_amount=50, is_outgoing=True)
|
|
bus.publish_sync(low_damage)
|
|
|
|
# Publish high damage (should be received)
|
|
high_damage = DamageEvent(damage_amount=150, is_outgoing=True)
|
|
bus.publish_sync(high_damage)
|
|
|
|
if len(received) == 1 and received[0].damage_amount == 150:
|
|
result.status = "PASS"
|
|
result.message = "Event filtering working correctly"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = f"Expected 1 high damage event, got {len(received)}"
|
|
|
|
bus.unsubscribe(sub_id)
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# Test Plugin API - Service Registration
|
|
result = TestResult("Plugin API Service Registration", "Cross-Plugin Communication")
|
|
try:
|
|
from core.plugin_api import get_api, APIType
|
|
api = get_api()
|
|
|
|
# Test registering a mock service
|
|
def mock_handler():
|
|
return "mock_result"
|
|
|
|
from dataclasses import dataclass
|
|
@dataclass
|
|
class MockEndpoint:
|
|
name: str
|
|
api_type: APIType
|
|
description: str
|
|
handler: callable
|
|
plugin_id: str
|
|
version: str = "1.0.0"
|
|
|
|
endpoint = MockEndpoint(
|
|
name="test_api",
|
|
api_type=APIType.UTILITY,
|
|
description="Test API",
|
|
handler=mock_handler,
|
|
plugin_id="test.plugin"
|
|
)
|
|
|
|
success = api.register_api(endpoint)
|
|
|
|
if success:
|
|
result.status = "PASS"
|
|
result.message = "API endpoint registration working"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Failed to register API endpoint"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# Test Plugin API - Data Sharing
|
|
result = TestResult("Plugin API Data Sharing", "Cross-Plugin Communication")
|
|
try:
|
|
from core.plugin_api import get_api
|
|
api = get_api()
|
|
|
|
# Set shared data
|
|
api.set_data("test_key", {"value": 123})
|
|
|
|
# Get shared data
|
|
data = api.get_data("test_key")
|
|
|
|
if data and data.get("value") == 123:
|
|
result.status = "PASS"
|
|
result.message = "Data sharing working correctly"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Data not retrieved correctly"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# ============================================================================
|
|
# TEST 3: PLUGIN LOADING
|
|
# ============================================================================
|
|
|
|
def test_plugin_loading():
|
|
"""Test that all plugins can be discovered and loaded."""
|
|
|
|
# Test Plugin Discovery
|
|
result = TestResult("Plugin Discovery", "Plugin Loading")
|
|
try:
|
|
from core.plugin_manager import PluginManager
|
|
|
|
# Create manager without overlay
|
|
manager = PluginManager(None)
|
|
discovered = manager.discover_plugins()
|
|
|
|
plugin_count = len(discovered)
|
|
|
|
if plugin_count >= 25:
|
|
result.status = "PASS"
|
|
result.message = f"Discovered {plugin_count} plugins"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = f"Only discovered {plugin_count} plugins, expected 25+"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# Test Individual Plugin Imports (25 plugins total)
|
|
plugin_dirs = [
|
|
"auction_tracker", "calculator", "chat_logger", "codex_tracker",
|
|
"crafting_calc", "dashboard", "dpp_calculator", "enhancer_calc",
|
|
"event_bus_example", "game_reader", "global_tracker", "inventory_manager",
|
|
"loot_tracker", "mining_helper", "mission_tracker", "nexus_search",
|
|
"plugin_store_ui", "price_alerts", "profession_scanner", "session_exporter",
|
|
"settings", "skill_scanner", "spotify_controller", "tp_runner", "universal_search"
|
|
]
|
|
|
|
for plugin_name in plugin_dirs:
|
|
result = TestResult(f"Plugin: {plugin_name}", "Plugin Loading")
|
|
try:
|
|
# Try to import the plugin module
|
|
module_name = f"plugins.{plugin_name}.plugin"
|
|
module = __import__(module_name, fromlist=[""])
|
|
|
|
# Check for plugin class
|
|
from plugins.base_plugin import BasePlugin
|
|
found_class = False
|
|
for attr_name in dir(module):
|
|
attr = getattr(module, attr_name)
|
|
if (isinstance(attr, type) and
|
|
issubclass(attr, BasePlugin) and
|
|
attr != BasePlugin):
|
|
found_class = True
|
|
result.status = "PASS"
|
|
result.message = f"Found plugin class: {attr_name}"
|
|
break
|
|
|
|
if not found_class:
|
|
result.status = "FAIL"
|
|
result.message = "No plugin class found"
|
|
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# Test Base Plugin
|
|
result = TestResult("Base Plugin Class", "Plugin Loading")
|
|
try:
|
|
from plugins.base_plugin import BasePlugin
|
|
if hasattr(BasePlugin, 'initialize') and hasattr(BasePlugin, 'get_ui'):
|
|
result.status = "PASS"
|
|
result.message = "BasePlugin has required methods"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "BasePlugin missing required methods"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# ============================================================================
|
|
# TEST 4: GLOBAL HOTKEYS
|
|
# ============================================================================
|
|
|
|
def test_global_hotkeys():
|
|
"""Test global hotkey configuration."""
|
|
|
|
result = TestResult("Hotkey Library Availability", "Global Hotkeys")
|
|
try:
|
|
import keyboard
|
|
result.status = "PASS"
|
|
result.message = "keyboard library available"
|
|
except ImportError:
|
|
result.status = "SKIP"
|
|
result.message = "keyboard library not installed (optional)"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
result = TestResult("Main Hotkey Configuration", "Global Hotkeys")
|
|
try:
|
|
# Check main.py for hotkey configuration
|
|
main_path = project_root / "core" / "main.py"
|
|
if main_path.exists():
|
|
content = main_path.read_text()
|
|
if "ctrl+shift+u" in content and "ctrl+shift+h" in content:
|
|
result.status = "PASS"
|
|
result.message = "Main hotkeys configured (Ctrl+Shift+U, Ctrl+Shift+H)"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Main hotkeys not found in configuration"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "main.py not found"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
result = TestResult("Plugin Hotkey Support", "Global Hotkeys")
|
|
try:
|
|
from plugins.base_plugin import BasePlugin
|
|
if hasattr(BasePlugin, 'hotkey') and hasattr(BasePlugin, 'on_hotkey'):
|
|
result.status = "PASS"
|
|
result.message = "BasePlugin supports hotkey configuration"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "BasePlugin missing hotkey support"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# ============================================================================
|
|
# TEST 5: SYSTEM TRAY INTEGRATION
|
|
# ============================================================================
|
|
|
|
def test_system_tray():
|
|
"""Test system tray integration."""
|
|
|
|
result = TestResult("System Tray Support (PyQt6)", "System Tray")
|
|
try:
|
|
from PyQt6.QtWidgets import QSystemTrayIcon, QMenu
|
|
result.status = "PASS"
|
|
result.message = "PyQt6 system tray classes available"
|
|
except ImportError:
|
|
result.status = "SKIP"
|
|
result.message = "PyQt6 not installed (GUI tests skipped)"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
result = TestResult("Floating Icon Implementation", "System Tray")
|
|
try:
|
|
floating_icon_path = project_root / "core" / "floating_icon.py"
|
|
if floating_icon_path.exists():
|
|
content = floating_icon_path.read_text()
|
|
if "QSystemTrayIcon" in content or "FloatingIcon" in content:
|
|
result.status = "PASS"
|
|
result.message = "Floating icon/system tray implementation found"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "System tray implementation incomplete"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "floating_icon.py not found"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# ============================================================================
|
|
# TEST 6: EXTERNAL API CALLS
|
|
# ============================================================================
|
|
|
|
def test_external_apis():
|
|
"""Test external API integrations."""
|
|
|
|
# Test Nexus API - Structure
|
|
result = TestResult("Nexus API Structure", "External APIs")
|
|
try:
|
|
from core.nexus_api import get_nexus_api, NexusAPI
|
|
nexus = get_nexus_api()
|
|
|
|
required_methods = ['search', 'get_item_details', 'get_market_data', 'is_available']
|
|
available = [m for m in required_methods if hasattr(nexus, m)]
|
|
|
|
if len(available) >= 3:
|
|
result.status = "PASS"
|
|
result.message = f"Nexus API has {len(available)}/{len(required_methods)} required methods"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = f"Nexus API missing methods: {set(required_methods) - set(available)}"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# Test HTTP Client
|
|
result = TestResult("HTTP Client Structure", "External APIs")
|
|
try:
|
|
from core.http_client import get_http_client, HTTPClient
|
|
http = get_http_client()
|
|
|
|
required_methods = ['get', 'post', 'get_stats', 'clear_cache']
|
|
available = [m for m in required_methods if hasattr(http, m)]
|
|
|
|
if len(available) >= 3:
|
|
result.status = "PASS"
|
|
result.message = f"HTTP Client has {len(available)}/{len(required_methods)} required methods"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = f"HTTP Client missing methods"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# Test HTTP Client - Cache
|
|
result = TestResult("HTTP Client Caching", "External APIs")
|
|
try:
|
|
from core.http_client import get_http_client
|
|
http = get_http_client()
|
|
|
|
if hasattr(http, 'cache') or hasattr(http, '_cache'):
|
|
result.status = "PASS"
|
|
result.message = "HTTP Client has caching capability"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "HTTP Client missing caching"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# Test Nexus API - Mock Call
|
|
result = TestResult("Nexus API Mock Search", "External APIs")
|
|
try:
|
|
from core.nexus_api import get_nexus_api
|
|
nexus = get_nexus_api()
|
|
|
|
# Just check that the method exists and is callable
|
|
if hasattr(nexus, 'search') and callable(getattr(nexus, 'search')):
|
|
result.status = "PASS"
|
|
result.message = "Nexus search method is callable"
|
|
else:
|
|
result.status = "FAIL"
|
|
result.message = "Nexus search method not available"
|
|
except Exception as e:
|
|
result.status = "FAIL"
|
|
result.error = e
|
|
result.message = str(e)
|
|
report.add_result(result)
|
|
|
|
# ============================================================================
|
|
# MAIN EXECUTION
|
|
# ============================================================================
|
|
|
|
if __name__ == "__main__":
|
|
print("Starting EU-Utility Integration Tests...")
|
|
print(f"Project Root: {project_root}")
|
|
print()
|
|
|
|
# Run all test suites
|
|
test_core_services()
|
|
test_cross_plugin_communication()
|
|
test_plugin_loading()
|
|
test_global_hotkeys()
|
|
test_system_tray()
|
|
test_external_apis()
|
|
|
|
# Print final report
|
|
report.print_report()
|
|
|
|
# Exit with appropriate code
|
|
summary = report.get_summary()
|
|
if summary["failed"] > 0:
|
|
sys.exit(1)
|
|
else:
|
|
sys.exit(0)
|