#!/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)