# -*- coding: utf-8 -*- """ Plugin Test-Harness. Bietet eine isolierte Testumgebung für Plugin-Tests. """ from __future__ import annotations import asyncio from dataclasses import dataclass, field from pathlib import Path from typing import Any, TYPE_CHECKING if TYPE_CHECKING: from trixy_core.plugins.plugin_manager import PluginManager from trixy_core.plugins.base import TrixyPlugin @dataclass class TestContext: """ Kontext für einen Plugin-Test. Enthält alle Mock-Objekte und Hilfsmethoden für Tests. Attributes: plugin: Das zu testende Plugin events: Liste der emittierten Events logs: Liste der Log-Nachrichten config: Test-Konfiguration state: Beliebiger Test-Zustand """ plugin: Any = None events: list[dict[str, Any]] = field(default_factory=list) logs: list[dict[str, Any]] = field(default_factory=list) config: dict[str, Any] = field(default_factory=dict) state: dict[str, Any] = field(default_factory=dict) def clear_events(self) -> None: """Löscht alle aufgezeichneten Events.""" self.events.clear() def clear_logs(self) -> None: """Löscht alle aufgezeichneten Logs.""" self.logs.clear() def clear_all(self) -> None: """Löscht alle aufgezeichneten Daten.""" self.clear_events() self.clear_logs() self.state.clear() def get_events_by_name(self, name: str) -> list[dict[str, Any]]: """Gibt Events mit einem bestimmten Namen zurück.""" return [e for e in self.events if e.get("name") == name] def has_event(self, name: str) -> bool: """Prüft, ob ein Event mit diesem Namen existiert.""" return any(e.get("name") == name for e in self.events) def get_last_event(self) -> dict[str, Any] | None: """Gibt das letzte Event zurück.""" return self.events[-1] if self.events else None def get_logs_by_level(self, level: str) -> list[dict[str, Any]]: """Gibt Logs mit einem bestimmten Level zurück.""" return [log for log in self.logs if log.get("level") == level] def set_state(self, key: str, value: Any) -> None: """Setzt einen Zustandswert.""" self.state[key] = value def get_state(self, key: str, default: Any = None) -> Any: """Gibt einen Zustandswert zurück.""" return self.state.get(key, default) class PluginTestHarness: """ Test-Harness für Plugin-Tests. Bietet eine isolierte Umgebung zum Testen von Plugins ohne die vollständige Anwendung zu starten. Example: # Test-Harness erstellen harness = PluginTestHarness() # Plugin laden await harness.load_plugin(MyPlugin) # Plugin testen context = harness.context await harness.emit_event("test_event", {"data": "test"}) # Ergebnisse prüfen assert context.has_event("response_event") # Aufräumen await harness.cleanup() """ def __init__( self, config: dict[str, Any] | None = None, plugin_dir: Path | None = None, ) -> None: """ Initialisiert das Test-Harness. Args: config: Optionale Basis-Konfiguration plugin_dir: Optionales Plugin-Verzeichnis """ self._config = config or {} self._plugin_dir = plugin_dir self._context = TestContext(config=self._config.copy()) self._plugin: Any = None self._mock_app: Any = None self._mock_event_manager: Any = None self._event_handlers: dict[str, list[Any]] = {} self._running = False @property def context(self) -> TestContext: """Der Test-Kontext.""" return self._context @property def plugin(self) -> Any: """Das geladene Plugin.""" return self._plugin @property def is_running(self) -> bool: """Läuft das Harness?""" return self._running async def load_plugin( self, plugin_class: type, config: dict[str, Any] | None = None, ) -> Any: """ Lädt ein Plugin in das Test-Harness. Args: plugin_class: Die Plugin-Klasse config: Plugin-spezifische Konfiguration Returns: Die Plugin-Instanz """ from trixy_core.plugins.testing.mocks import ( MockApplication, MockEventManager, ) # Mock-Objekte erstellen self._mock_event_manager = MockEventManager() self._mock_event_manager.set_emit_callback(self._on_event_emitted) self._mock_app = MockApplication( event_manager=self._mock_event_manager, ) # Plugin-Konfiguration vorbereiten plugin_config = {**self._config, **(config or {})} # Plugin instanziieren self._plugin = plugin_class( application=self._mock_app, config=plugin_config, ) self._context.plugin = self._plugin self._context.config = plugin_config return self._plugin async def start_plugin(self) -> bool: """ Startet das geladene Plugin. Returns: True bei Erfolg """ if self._plugin is None: return False if self._running: return True try: # Plugin-Lifecycle aufrufen if hasattr(self._plugin, "on_pre_start"): await self._call_method(self._plugin.on_pre_start) if hasattr(self._plugin, "on_start"): await self._call_method(self._plugin.on_start) if hasattr(self._plugin, "on_post_start"): await self._call_method(self._plugin.on_post_start) self._running = True return True except Exception as e: self._log("error", f"Plugin-Start fehlgeschlagen: {e}") return False async def stop_plugin(self) -> bool: """ Stoppt das geladene Plugin. Returns: True bei Erfolg """ if self._plugin is None: return False if not self._running: return True try: # Plugin-Lifecycle aufrufen if hasattr(self._plugin, "on_pre_stop"): await self._call_method(self._plugin.on_pre_stop) if hasattr(self._plugin, "on_stop"): await self._call_method(self._plugin.on_stop) if hasattr(self._plugin, "on_post_stop"): await self._call_method(self._plugin.on_post_stop) self._running = False return True except Exception as e: self._log("error", f"Plugin-Stop fehlgeschlagen: {e}") return False async def emit_event( self, name: str, data: dict[str, Any] | None = None, ) -> list[Any]: """ Emittiert ein Event an das Plugin. Args: name: Event-Name data: Event-Daten Returns: Liste der Handler-Ergebnisse """ event_data = data or {} results = [] # Event aufzeichnen self._context.events.append({ "name": name, "data": event_data, "source": "test", }) # Handler aufrufen if self._plugin and hasattr(self._plugin, "_event_handlers"): handlers = getattr(self._plugin, "_event_handlers", {}) if name in handlers: for handler in handlers[name]: try: result = await self._call_method( handler, name, event_data ) results.append(result) except Exception as e: self._log("error", f"Handler-Fehler: {e}") return results async def call_method( self, method_name: str, *args: Any, **kwargs: Any, ) -> Any: """ Ruft eine Methode des Plugins auf. Args: method_name: Name der Methode *args: Argumente **kwargs: Keyword-Argumente Returns: Rückgabewert der Methode """ if self._plugin is None: raise RuntimeError("Kein Plugin geladen") method = getattr(self._plugin, method_name, None) if method is None: raise AttributeError(f"Methode nicht gefunden: {method_name}") return await self._call_method(method, *args, **kwargs) def set_config(self, key: str, value: Any) -> None: """Setzt einen Konfigurationswert.""" self._context.config[key] = value if self._plugin and hasattr(self._plugin, "config"): self._plugin.config[key] = value def get_config(self, key: str, default: Any = None) -> Any: """Gibt einen Konfigurationswert zurück.""" return self._context.config.get(key, default) async def cleanup(self) -> None: """Räumt das Test-Harness auf.""" if self._running: await self.stop_plugin() self._plugin = None self._mock_app = None self._mock_event_manager = None self._event_handlers.clear() self._context.clear_all() def _on_event_emitted( self, name: str, data: dict[str, Any], ) -> None: """Callback für emittierte Events.""" self._context.events.append({ "name": name, "data": data, "source": "plugin", }) def _log(self, level: str, message: str) -> None: """Zeichnet eine Log-Nachricht auf.""" self._context.logs.append({ "level": level, "message": message, }) async def _call_method( self, method: Any, *args: Any, **kwargs: Any, ) -> Any: """Ruft eine Methode auf (sync oder async).""" result = method(*args, **kwargs) if asyncio.iscoroutine(result): return await result return result def register_event_handler( self, event_name: str, handler: Any, ) -> None: """ Registriert einen Event-Handler für Tests. Args: event_name: Event-Name handler: Handler-Funktion """ if event_name not in self._event_handlers: self._event_handlers[event_name] = [] self._event_handlers[event_name].append(handler) def unregister_event_handler( self, event_name: str, handler: Any, ) -> bool: """ Entfernt einen Event-Handler. Args: event_name: Event-Name handler: Handler-Funktion Returns: True wenn entfernt """ if event_name in self._event_handlers: try: self._event_handlers[event_name].remove(handler) return True except ValueError: pass return False async def wait_for_event( self, event_name: str, timeout: float = 5.0, ) -> dict[str, Any] | None: """ Wartet auf ein Event. Args: event_name: Event-Name timeout: Timeout in Sekunden Returns: Event-Daten oder None bei Timeout """ start_count = len(self._context.get_events_by_name(event_name)) elapsed = 0.0 interval = 0.05 while elapsed < timeout: current = self._context.get_events_by_name(event_name) if len(current) > start_count: return current[-1] await asyncio.sleep(interval) elapsed += interval return None def assert_event_emitted( self, event_name: str, count: int | None = None, ) -> None: """ Prüft, ob ein Event emittiert wurde. Args: event_name: Event-Name count: Erwartete Anzahl (None = mindestens 1) Raises: AssertionError: Wenn Bedingung nicht erfüllt """ events = self._context.get_events_by_name(event_name) if count is None: assert len(events) > 0, f"Event '{event_name}' wurde nicht emittiert" else: assert len(events) == count, ( f"Event '{event_name}' wurde {len(events)} mal emittiert, " f"erwartet: {count}" ) def assert_no_errors(self) -> None: """ Prüft, dass keine Fehler aufgezeichnet wurden. Raises: AssertionError: Wenn Fehler vorhanden """ errors = self._context.get_logs_by_level("error") assert len(errors) == 0, f"Fehler aufgetreten: {errors}"