Trixy7/trixy_core/plugins/testing/harness.py

# -*- 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}"