""" Integration Tests - Plugin Workflows ===================================== Tests for complete plugin workflows and interactions between components. """ import pytest import time from unittest.mock import Mock, patch, MagicMock @pytest.mark.integration class TestPluginLifecycle: """Test complete plugin lifecycle.""" def test_plugin_full_lifecycle(self, mock_overlay, temp_dir): """Test plugin from discovery to shutdown.""" from core.plugin_manager import PluginManager from plugins.base_plugin import BasePlugin # Create a test plugin class TestPlugin(BasePlugin): name = "Integration Test Plugin" version = "1.0.0" author = "Test" description = "Integration test" initialized = False shutdown_called = False def initialize(self): self.initialized = True self.log_info("Initialized") def shutdown(self): self.shutdown_called = True self.log_info("Shutdown") def get_ui(self): from PyQt6.QtWidgets import QWidget, QLabel, QVBoxLayout widget = QWidget() layout = QVBoxLayout(widget) layout.addWidget(QLabel("Test Plugin UI")) return widget # Create plugin manager pm = PluginManager(mock_overlay) pm.config["enabled"] = ["test.plugin"] pm.plugin_classes["test.plugin"] = TestPlugin # Load plugin result = pm.load_plugin(TestPlugin) assert result is True # Verify plugin is loaded plugin = pm.get_plugin("test.plugin") assert plugin is not None assert plugin.initialized is True # Verify UI can be retrieved ui = pm.get_plugin_ui("test.plugin") assert ui is not None # Shutdown pm.shutdown_all() assert plugin.shutdown_called is True def test_plugin_enable_disable_workflow(self, mock_overlay, temp_dir): """Test enabling and disabling plugins.""" from core.plugin_manager import PluginManager from plugins.base_plugin import BasePlugin class TogglePlugin(BasePlugin): name = "Toggle Test" pm = PluginManager(mock_overlay) pm.save_config = Mock() # Prevent file writes pm.plugin_classes["toggle.plugin"] = TogglePlugin # Initially disabled assert pm.is_plugin_enabled("toggle.plugin") is False # Enable plugin result = pm.enable_plugin("toggle.plugin") assert result is True assert pm.is_plugin_enabled("toggle.plugin") is True assert "toggle.plugin" in pm.config["enabled"] # Disable plugin result = pm.disable_plugin("toggle.plugin") assert result is True assert pm.is_plugin_enabled("toggle.plugin") is False assert "toggle.plugin" not in pm.config["enabled"] def test_plugin_settings_persistence(self, mock_overlay, temp_dir): """Test plugin settings are saved and loaded.""" from core.plugin_manager import PluginManager from plugins.base_plugin import BasePlugin class SettingsPlugin(BasePlugin): name = "Settings Test" def on_config_changed(self, key, value): self.config[key] = value pm = PluginManager(mock_overlay) # Create with custom settings plugin_config = {"setting1": "value1", "number": 42} pm.config["settings"]["settings.plugin"] = plugin_config pm.config["enabled"] = ["settings.plugin"] plugin = SettingsPlugin(mock_overlay, plugin_config) # Verify settings are available assert plugin.config["setting1"] == "value1" assert plugin.config["number"] == 42 @pytest.mark.integration class TestAPIWorkflows: """Test complete API workflows.""" def test_log_reading_and_parsing_workflow(self): """Test log reading and parsing workflow.""" from core.plugin_api import PluginAPI api = PluginAPI() # Mock log reader log_lines = [ "[2024-02-15 14:30:25] System: You gained 0.12 points in Rifle", "[2024-02-15 14:30:30] Loot: You received Shrapnel x 50", "[2024-02-15 14:30:35] Loot: You received Weapon Cells x 100", ] mock_reader = Mock(return_value=log_lines) api.register_log_service(mock_reader) # Read logs lines = api.read_log_lines(100) # Parse skill gains skill_gains = [] loot_events = [] for line in lines: if "gained" in line and "points" in line: # Parse skill gain parts = line.split() for i, part in enumerate(parts): if part == "gained": points = parts[i + 1] skill = parts[i + 3] skill_gains.append({"skill": skill, "points": float(points)}) elif "Loot:" in line: loot_events.append(line) assert len(skill_gains) == 1 assert skill_gains[0]["skill"] == "Rifle" assert len(loot_events) == 2 def test_window_detection_and_overlay_workflow(self): """Test window detection and overlay positioning workflow.""" from core.plugin_api import PluginAPI from core.window_manager import WindowInfo api = PluginAPI() # Mock window manager mock_wm = Mock() mock_wm.is_available.return_value = True window_info = WindowInfo( handle=12345, title="Entropia Universe", pid=67890, rect=(100, 100, 1100, 700), width=1000, height=600, is_visible=True, is_focused=True ) mock_wm.find_eu_window.return_value = window_info api.register_window_service(mock_wm) # Detect window eu_window = api.get_eu_window() assert eu_window is not None # Check if EU is focused is_focused = api.is_eu_focused() assert is_focused is True # Calculate overlay position (centered on EU window) center_x = eu_window['x'] + eu_window['width'] // 2 center_y = eu_window['y'] + eu_window['height'] // 2 assert center_x == 600 assert center_y == 400 def test_ocr_and_notification_workflow(self): """Test OCR recognition and notification workflow.""" from core.plugin_api import PluginAPI api = PluginAPI() # Mock OCR service ocr_text = """PED: 1500.00 Items: 45/200 TT Value: 2345.67""" mock_ocr = Mock(return_value=ocr_text) api.register_ocr_service(mock_ocr) # Mock notification service mock_notification = Mock() api.register_notification_service(mock_notification) # Perform OCR text = api.recognize_text(region=(100, 100, 200, 100)) # Parse PED value ped_value = None for line in text.split('\n'): if line.startswith('PED:'): ped_value = float(line.replace('PED:', '').strip()) assert ped_value == 1500.00 # Show notification with result api.show_notification( "Inventory Scan", f"Current PED: {ped_value:.2f}", duration=3000 ) mock_notification.show.assert_called_once() def test_nexus_search_and_data_storage_workflow(self): """Test Nexus search and data storage workflow.""" from core.plugin_api import PluginAPI from core.data_store import DataStore api = PluginAPI() # Mock Nexus API mock_nexus = Mock() search_results = [ {"Id": 123, "Name": "Test Item", "Value": 100.0, "Markup": 110.0} ] mock_nexus.search_items.return_value = search_results api.register_nexus_service(mock_nexus) # Create data store data_store = DataStore(":memory:") # In-memory for testing api.register_data_service(data_store) # Search for item items = api.search_items("test item", limit=5) # Store results api.set_data("last_search", items) api.set_data("search_time", time.time()) # Retrieve and verify stored_items = api.get_data("last_search") assert len(stored_items) == 1 assert stored_items[0]["Name"] == "Test Item" def test_event_subscription_and_publish_workflow(self, event_bus): """Test event subscription and publishing workflow.""" from core.plugin_api import PluginAPI api = PluginAPI() api.register_event_bus(event_bus) received_events = [] def on_loot(event): received_events.append(event.data) def on_skill_gain(event): received_events.append(event.data) # Subscribe to events api.subscribe("loot", on_loot) api.subscribe("skill_gain", on_skill_gain) # Publish events api.publish("loot", {"item": "Shrapnel", "amount": 50}) api.publish("skill_gain", {"skill": "Rifle", "points": 0.12}) # Verify events received assert len(received_events) == 2 assert received_events[0]["item"] == "Shrapnel" assert received_events[1]["skill"] == "Rifle" @pytest.mark.integration class TestUIIntegration: """Test UI integration workflows.""" def test_overlay_show_hide_workflow(self): """Test overlay show/hide workflow.""" pytest.importorskip("PyQt6") from PyQt6.QtWidgets import QApplication app = QApplication.instance() or QApplication([]) from core.overlay_window import OverlayWindow with patch.object(OverlayWindow, '_setup_window'): with patch.object(OverlayWindow, '_setup_ui'): with patch.object(OverlayWindow, '_setup_tray'): with patch.object(OverlayWindow, '_setup_shortcuts'): with patch.object(OverlayWindow, '_setup_animations'): with patch.object(OverlayWindow, 'hide'): window = OverlayWindow(None) # Initially hidden assert window.is_visible is False # Show overlay with patch.object(window, 'show'): with patch.object(window, 'raise_'): with patch.object(window, 'activateWindow'): window.show_overlay() assert window.is_visible is True # Hide overlay window.hide_overlay() assert window.is_visible is False def test_plugin_switching_workflow(self, mock_plugin_manager): """Test plugin switching workflow.""" pytest.importorskip("PyQt6") from PyQt6.QtWidgets import QApplication app = QApplication.instance() or QApplication([]) from core.overlay_window import OverlayWindow # Mock plugins mock_plugin1 = Mock() mock_plugin1.name = "Plugin 1" mock_ui1 = Mock() mock_plugin1.get_ui.return_value = mock_ui1 mock_plugin2 = Mock() mock_plugin2.name = "Plugin 2" mock_ui2 = Mock() mock_plugin2.get_ui.return_value = mock_ui2 mock_plugin_manager.get_all_plugins.return_value = { "plugin1": mock_plugin1, "plugin2": mock_plugin2 } with patch.object(OverlayWindow, '_setup_window'): with patch.object(OverlayWindow, '_setup_tray'): with patch.object(OverlayWindow, '_setup_shortcuts'): with patch.object(OverlayWindow, '_setup_animations'): with patch.object(OverlayWindow, 'hide'): window = OverlayWindow(mock_plugin_manager) # Mock the plugin loading window.sidebar_buttons = [Mock(), Mock()] window.plugin_stack = Mock() # Switch to plugin 2 window._on_plugin_selected(1) assert window.current_plugin_index == 1 def test_dashboard_widget_workflow(self): """Test dashboard widget workflow.""" pytest.importorskip("PyQt6") from PyQt6.QtWidgets import QApplication app = QApplication.instance() or QApplication([]) from core.dashboard import Dashboard, PEDTrackerWidget with patch.object(Dashboard, '_setup_ui'): with patch.object(Dashboard, '_add_default_widgets'): dashboard = Dashboard() # Create and add widget widget = PEDTrackerWidget() dashboard.add_widget(widget) assert widget in dashboard.widgets # Update widget data widget.update_data({"ped": 1500.00, "change": 50.00}) # Remove widget dashboard.remove_widget(widget) assert widget not in dashboard.widgets @pytest.mark.integration class TestSettingsWorkflow: """Test settings-related workflows.""" def test_settings_save_load_workflow(self, temp_dir): """Test settings save and load workflow.""" from core.settings import Settings config_path = temp_dir / "config" / "settings.json" config_path.parent.mkdir(parents=True) # Create settings and modify settings1 = Settings(str(config_path)) settings1.set("hotkeys.toggle", "ctrl+shift+u") settings1.set("theme.mode", "dark") settings1.set("overlay.opacity", 0.9) settings1.save() # Load in new instance settings2 = Settings(str(config_path)) assert settings2.get("hotkeys.toggle") == "ctrl+shift+u" assert settings2.get("theme.mode") == "dark" assert settings2.get("overlay.opacity") == 0.9 def test_plugin_settings_isolation(self, temp_dir): """Test plugin settings are isolated.""" from core.plugin_manager import PluginManager from core.settings import Settings config_path = temp_dir / "config" / "settings.json" config_path.parent.mkdir(parents=True) settings = Settings(str(config_path)) # Set plugin-specific settings settings.set("plugins.calculator.precision", 2) settings.set("plugins.tracker.auto_save", True) settings.set("plugins.scanner.region", (100, 100, 200, 200)) # Verify isolation assert settings.get("plugins.calculator.precision") == 2 assert settings.get("plugins.tracker.auto_save") is True assert settings.get("plugins.scanner.region") == [100, 100, 200, 200] @pytest.mark.integration class TestErrorHandlingWorkflows: """Test error handling in workflows.""" def test_plugin_load_error_handling(self, mock_overlay): """Test plugin load error handling.""" from core.plugin_manager import PluginManager from plugins.base_plugin import BasePlugin class BrokenPlugin(BasePlugin): name = "Broken Plugin" def initialize(self): raise Exception("Initialization failed") pm = PluginManager(mock_overlay) # Should not raise exception, just return False result = pm.load_plugin(BrokenPlugin) assert result is False def test_api_service_unavailable_handling(self): """Test API handling when services unavailable.""" from core.plugin_api import PluginAPI, ServiceNotAvailableError api = PluginAPI() # OCR not available assert api.ocr_available() is False with pytest.raises(ServiceNotAvailableError): api.recognize_text(region=(0, 0, 100, 100)) # HTTP not available result = api.http_get("https://example.com") assert result['success'] is False assert 'error' in result def test_graceful_degradation_on_missing_services(self): """Test graceful degradation when services are missing.""" from core.plugin_api import PluginAPI api = PluginAPI() # All these should return sensible defaults when services unavailable assert api.read_log_lines(100) == [] assert api.get_eu_window() is None assert api.is_eu_focused() is False assert api.play_sound("test.wav") is False assert api.copy_to_clipboard("test") is False assert api.paste_from_clipboard() == "" assert api.get_data("key", "default") == "default"