Trixy7/tests/test_nlp.py

# -*- coding: utf-8 -*-
"""
Tests für das NLP-System.

Testet IntentRegistry, @intent Decorator, NLPProvider Interface und Event-Flow.
"""

import asyncio
import pytest
from dataclasses import dataclass
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch

from trixy_core.nlp import (
    IntentDefinition,
    IntentRegistry,
    IntentSlot,
    IntentReceivedData,
    IntentResult,
    NLPConfig,
    NLPContext,
    NLPProvider,
    NLPResult,
    NLPState,
    intent,
    discover_intent_handlers,
    get_intent_metadata,
    is_intent_handler,
)


# ============================================================================
# IntentRegistry Tests
# ============================================================================

class TestIntentRegistry:
    """Tests für die IntentRegistry."""

    @pytest.fixture(autouse=True)
    def setup_registry(self):
        """Setzt die Registry vor jedem Test zurück."""
        registry = IntentRegistry.get_instance()
        registry.clear()
        yield registry
        registry.clear()

    def test_singleton_pattern(self):
        """Registry sollte Singleton sein."""
        registry1 = IntentRegistry.get_instance()
        registry2 = IntentRegistry.get_instance()
        assert registry1 is registry2

    def test_register_intent(self, setup_registry):
        """Intent sollte registriert werden können."""
        registry = setup_registry

        def handler():
            pass

        intent_def = IntentDefinition(
            name="test_intent",
            handler=handler,
            plugin_name="test_plugin",
            description="Test Intent",
        )

        result = registry.register(intent_def)

        assert result is True
        assert "test_intent" in registry
        assert registry.count() == 1

    def test_register_duplicate_intent_same_plugin(self, setup_registry):
        """Doppelter Intent vom gleichen Plugin sollte ersetzt werden."""
        registry = setup_registry

        def handler1():
            pass

        def handler2():
            pass

        intent1 = IntentDefinition(
            name="test_intent",
            handler=handler1,
            plugin_name="test_plugin",
        )
        intent2 = IntentDefinition(
            name="test_intent",
            handler=handler2,
            plugin_name="test_plugin",
        )

        registry.register(intent1)
        result = registry.register(intent2)

        assert result is True
        assert registry.get_handler("test_intent") is handler2

    def test_register_duplicate_intent_higher_priority(self, setup_registry):
        """Intent mit höherer Priorität sollte überschreiben."""
        registry = setup_registry

        def handler1():
            pass

        def handler2():
            pass

        intent1 = IntentDefinition(
            name="test_intent",
            handler=handler1,
            plugin_name="plugin1",
            priority=0,
        )
        intent2 = IntentDefinition(
            name="test_intent",
            handler=handler2,
            plugin_name="plugin2",
            priority=10,
        )

        registry.register(intent1)
        result = registry.register(intent2)

        assert result is True
        assert registry.get_handler("test_intent") is handler2

    def test_register_duplicate_intent_lower_priority(self, setup_registry):
        """Intent mit niedrigerer Priorität sollte abgelehnt werden."""
        registry = setup_registry

        def handler1():
            pass

        def handler2():
            pass

        intent1 = IntentDefinition(
            name="test_intent",
            handler=handler1,
            plugin_name="plugin1",
            priority=10,
        )
        intent2 = IntentDefinition(
            name="test_intent",
            handler=handler2,
            plugin_name="plugin2",
            priority=0,
        )

        registry.register(intent1)
        result = registry.register(intent2)

        assert result is False
        assert registry.get_handler("test_intent") is handler1

    def test_unregister_intent(self, setup_registry):
        """Intent sollte entfernt werden können."""
        registry = setup_registry

        def handler():
            pass

        intent_def = IntentDefinition(
            name="test_intent",
            handler=handler,
        )

        registry.register(intent_def)
        result = registry.unregister("test_intent")

        assert result is True
        assert "test_intent" not in registry
        assert registry.count() == 0

    def test_unregister_nonexistent(self, setup_registry):
        """Entfernen eines nicht existierenden Intents sollte False zurückgeben."""
        registry = setup_registry
        result = registry.unregister("nonexistent")
        assert result is False

    def test_unregister_plugin(self, setup_registry):
        """Alle Intents eines Plugins sollten entfernt werden können."""
        registry = setup_registry

        def handler():
            pass

        for i in range(3):
            registry.register(IntentDefinition(
                name=f"intent_{i}",
                handler=handler,
                plugin_name="test_plugin",
            ))

        registry.register(IntentDefinition(
            name="other_intent",
            handler=handler,
            plugin_name="other_plugin",
        ))

        count = registry.unregister_plugin("test_plugin")

        assert count == 3
        assert registry.count() == 1
        assert "other_intent" in registry

    def test_get_handler(self, setup_registry):
        """Handler sollte zurückgegeben werden."""
        registry = setup_registry

        def handler():
            return "result"

        registry.register(IntentDefinition(
            name="test_intent",
            handler=handler,
        ))

        result = registry.get_handler("test_intent")
        assert result is handler

    def test_get_handler_disabled(self, setup_registry):
        """Handler für deaktivierten Intent sollte None sein."""
        registry = setup_registry

        def handler():
            pass

        registry.register(IntentDefinition(
            name="test_intent",
            handler=handler,
            enabled=False,
        ))

        result = registry.get_handler("test_intent")
        assert result is None

    def test_get_all_as_dict(self, setup_registry):
        """Intents sollten als Dictionary-Liste zurückgegeben werden."""
        registry = setup_registry

        def handler():
            pass

        registry.register(IntentDefinition(
            name="intent1",
            handler=handler,
            plugin_name="plugin1",
            description="First intent",
            examples=["example1", "example2"],
            slots={
                "param": IntentSlot(name="param", slot_type=str),
            },
        ))

        result = registry.get_all_as_dict()

        assert len(result) == 1
        assert result[0]["name"] == "intent1"
        assert result[0]["description"] == "First intent"
        assert result[0]["examples"] == ["example1", "example2"]
        assert "param" in result[0]["slots"]

    def test_enable_disable(self, setup_registry):
        """Intents sollten aktiviert/deaktiviert werden können."""
        registry = setup_registry

        def handler():
            pass

        registry.register(IntentDefinition(
            name="test_intent",
            handler=handler,
        ))

        assert registry.get_handler("test_intent") is handler

        registry.disable("test_intent")
        assert registry.get_handler("test_intent") is None

        registry.enable("test_intent")
        assert registry.get_handler("test_intent") is handler


# ============================================================================
# IntentSlot Tests
# ============================================================================

class TestIntentSlot:
    """Tests für IntentSlot."""

    def test_basic_slot(self):
        """Slot sollte mit Standardwerten erstellt werden."""
        slot = IntentSlot(name="device")

        assert slot.name == "device"
        assert slot.slot_type is str
        assert slot.required is False
        assert slot.description == ""

    def test_slot_with_all_params(self):
        """Slot sollte alle Parameter akzeptieren."""
        slot = IntentSlot(
            name="temperature",
            slot_type=float,
            required=True,
            description="Zieltemperatur",
            examples=["20", "22.5"],
        )

        assert slot.name == "temperature"
        assert slot.slot_type is float
        assert slot.required is True
        assert slot.description == "Zieltemperatur"
        assert slot.examples == ["20", "22.5"]

    def test_slot_to_dict(self):
        """Slot sollte zu Dictionary konvertiert werden."""
        slot = IntentSlot(
            name="room",
            slot_type=str,
            required=True,
            description="Raumname",
        )

        result = slot.to_dict()

        assert result["name"] == "room"
        assert result["type"] == "str"
        assert result["required"] is True
        assert result["description"] == "Raumname"


# ============================================================================
# @intent Decorator Tests
# ============================================================================

class TestIntentDecorator:
    """Tests für den @intent Decorator."""

    def test_decorator_basic(self):
        """Decorator sollte Metadaten hinzufügen."""

        @intent("test_intent")
        async def handler(data):
            pass

        assert is_intent_handler(handler)
        metadata = get_intent_metadata(handler)
        assert metadata is not None
        assert metadata["name"] == "test_intent"

    def test_decorator_with_slots(self):
        """Decorator sollte Slots normalisieren."""

        @intent(
            "turn_on",
            slots={"device": str, "room": str},
        )
        async def handler(data):
            pass

        metadata = get_intent_metadata(handler)
        assert "device" in metadata["slots"]
        assert "room" in metadata["slots"]
        assert isinstance(metadata["slots"]["device"], IntentSlot)

    def test_decorator_with_intent_slot(self):
        """Decorator sollte IntentSlot direkt akzeptieren."""

        @intent(
            "set_temp",
            slots={"temp": IntentSlot("temp", float, required=True)},
        )
        async def handler(data):
            pass

        metadata = get_intent_metadata(handler)
        assert metadata["slots"]["temp"].slot_type is float
        assert metadata["slots"]["temp"].required is True

    def test_decorator_with_examples(self):
        """Decorator sollte Beispiele speichern."""

        @intent(
            "greeting",
            examples=["Hallo", "Guten Tag", "Hi"],
        )
        async def handler(data):
            pass

        metadata = get_intent_metadata(handler)
        assert len(metadata["examples"]) == 3
        assert "Hallo" in metadata["examples"]

    def test_decorator_preserves_function(self):
        """Decorator sollte Funktion nicht verändern."""

        @intent("test")
        async def handler(data):
            return "result"

        # Async-Funktion bleibt async
        assert asyncio.iscoroutinefunction(handler)

    def test_discover_handlers(self):
        """discover_intent_handlers sollte alle Handler finden."""

        class TestPlugin:
            @intent("intent1")
            async def handler1(self, data):
                pass

            @intent("intent2", slots={"param": str})
            async def handler2(self, data):
                pass

            def normal_method(self):
                pass

        plugin = TestPlugin()
        handlers = discover_intent_handlers(plugin)

        assert len(handlers) == 2
        names = [h[0] for h in handlers]
        assert "intent1" in names
        assert "intent2" in names


# ============================================================================
# Handler Data Classes Tests
# ============================================================================

class TestIntentReceivedData:
    """Tests für IntentReceivedData."""

    def test_get_slot(self):
        """Slot-Werte sollten abgerufen werden können."""
        data = IntentReceivedData(
            intent="test",
            slots={"device": "Licht", "room": "Wohnzimmer"},
        )

        assert data.get_slot("device") == "Licht"
        assert data.get_slot("room") == "Wohnzimmer"
        assert data.get_slot("unknown") is None
        assert data.get_slot("unknown", "default") == "default"

    def test_has_slot(self):
        """has_slot sollte korrekt prüfen."""
        data = IntentReceivedData(
            intent="test",
            slots={"device": "Licht", "empty": None},
        )

        assert data.has_slot("device") is True
        assert data.has_slot("empty") is False
        assert data.has_slot("unknown") is False

    def test_get_required_slots(self):
        """Erforderliche Slots sollten geprüft werden."""
        data = IntentReceivedData(
            intent="test",
            slots={"device": "Licht"},
        )

        all_present, missing = data.get_required_slots("device")
        assert all_present is True
        assert missing == []

        all_present, missing = data.get_required_slots("device", "room")
        assert all_present is False
        assert "room" in missing


class TestIntentResult:
    """Tests für IntentResult."""

    def test_success_with_response(self):
        """Erfolgreiche Antwort sollte erstellt werden."""
        result = IntentResult.success_with_response(
            "Licht wurde eingeschaltet.",
            device="Licht",
        )

        assert result.success is True
        assert result.response_text == "Licht wurde eingeschaltet."
        assert result.data["device"] == "Licht"

    def test_failure(self):
        """Fehler-Ergebnis sollte erstellt werden."""
        result = IntentResult.failure(
            "Gerät nicht gefunden",
            "Das Gerät konnte nicht gefunden werden.",
        )

        assert result.success is False
        assert result.error == "Gerät nicht gefunden"
        assert result.response_text == "Das Gerät konnte nicht gefunden werden."

    def test_silent_success(self):
        """Stilles Erfolgs-Ergebnis sollte erstellt werden."""
        result = IntentResult.silent_success(action="completed")

        assert result.success is True
        assert result.suppress_tts is True
        assert result.has_response() is False


# ============================================================================
# NLP Provider Tests
# ============================================================================

class TestNLPResult:
    """Tests für NLPResult."""

    def test_failure_result(self):
        """Fehler-Ergebnis sollte erstellt werden."""
        result = NLPResult.failure("Model error")

        assert result.success is False
        assert result.error == "Model error"

    def test_has_response(self):
        """has_response sollte korrekt prüfen."""
        result_with = NLPResult(intent="test", response_text="Antwort")
        result_without = NLPResult(intent="test")

        assert result_with.has_response() is True
        assert result_without.has_response() is False


class TestNLPContext:
    """Tests für NLPContext."""

    def test_basic_context(self):
        """Kontext sollte erstellt werden."""
        context = NLPContext(
            text="Licht an",
            satellite_id="sat1",
            room_id="wohnzimmer",
        )

        assert context.text == "Licht an"
        assert context.satellite_id == "sat1"
        assert context.room_id == "wohnzimmer"
        assert context.language == "de"

    def test_get_conversation_history_empty(self):
        """Leerer Verlauf ohne Session."""
        context = NLPContext(text="test")
        history = context.get_conversation_history()
        assert history == []


class TestNLPConfig:
    """Tests für NLPConfig."""

    def test_default_config(self):
        """Standardkonfiguration sollte gesetzt sein."""
        config = NLPConfig()

        assert config.backend == "llama_cpp"
        assert config.temperature == 0.1
        assert config.num_threads == 4
        assert config.use_gpu is False

    def test_custom_config(self):
        """Benutzerdefinierte Konfiguration sollte akzeptiert werden."""
        config = NLPConfig(
            backend="ollama",
            model_name="llama3",
            use_gpu=True,
            extra={"host": "localhost"},
        )

        assert config.backend == "ollama"
        assert config.model_name == "llama3"
        assert config.use_gpu is True
        assert config.extra["host"] == "localhost"


# ============================================================================
# Mock NLP Provider Tests
# ============================================================================

class MockNLPProvider(NLPProvider):
    """Mock-Provider für Tests."""

    def __init__(self):
        super().__init__()
        self._initialized = False

    async def initialize(self, config: NLPConfig) -> bool:
        self._config = config
        self._state = NLPState.READY
        self._initialized = True
        return True

    async def process(self, context: NLPContext) -> NLPResult:
        if not self._initialized:
            return NLPResult.failure("Not initialized")

        # Einfaches Mocking basierend auf Text
        if "licht" in context.text.lower():
            return NLPResult(
                intent="turn_on_device",
                confidence=0.9,
                slots={"device": "Licht"},
                response_text="Licht wird eingeschaltet.",
            )

        return NLPResult(
            intent="unknown",
            confidence=0.3,
        )

    async def shutdown(self) -> None:
        self._state = NLPState.SHUTDOWN
        self._initialized = False


class TestNLPProviderInterface:
    """Tests für das NLP Provider Interface."""

    @pytest.fixture
    def provider(self):
        """Erstellt einen Mock-Provider."""
        return MockNLPProvider()

    @pytest.mark.asyncio
    async def test_provider_lifecycle(self, provider):
        """Provider-Lebenszyklus sollte funktionieren."""
        assert provider.state == NLPState.UNINITIALIZED

        config = NLPConfig(model_name="test")
        await provider.initialize(config)
        assert provider.state == NLPState.READY
        assert provider.is_ready is True

        await provider.shutdown()
        assert provider.state == NLPState.SHUTDOWN

    @pytest.mark.asyncio
    async def test_provider_process(self, provider):
        """Provider sollte Text verarbeiten."""
        await provider.initialize(NLPConfig())

        context = NLPContext(text="Schalte das Licht ein")
        result = await provider.process(context)

        assert result.success is True
        assert result.intent == "turn_on_device"
        assert result.confidence >= 0.9
        assert "Licht" in result.slots.values()

    @pytest.mark.asyncio
    async def test_provider_unknown_intent(self, provider):
        """Unbekannter Intent sollte erkannt werden."""
        await provider.initialize(NLPConfig())

        context = NLPContext(text="xyz abc")
        result = await provider.process(context)

        assert result.intent == "unknown"
        assert result.confidence < 0.5


# ============================================================================
# Integration Tests
# ============================================================================

class TestIntegration:
    """Integrationstests für das NLP-System."""

    @pytest.fixture(autouse=True)
    def setup(self):
        """Setup und Teardown."""
        registry = IntentRegistry.get_instance()
        registry.clear()
        yield
        registry.clear()

    @pytest.mark.asyncio
    async def test_full_flow(self):
        """Kompletter Flow von Intent-Registrierung bis Handler-Aufruf."""
        registry = IntentRegistry.get_instance()

        # Handler definieren
        handler_called = False
        handler_data = None

        @intent(
            "turn_on_device",
            slots={"device": str},
            examples=["Licht an"],
        )
        async def handle_turn_on(data: IntentReceivedData) -> IntentResult:
            nonlocal handler_called, handler_data
            handler_called = True
            handler_data = data
            return IntentResult.success_with_response(
                f"{data.get_slot('device')} wurde eingeschaltet."
            )

        # Plugin-Objekt simulieren
        class TestPlugin:
            pass

        plugin = TestPlugin()
        plugin.handle_turn_on = handle_turn_on

        # Handler entdecken und registrieren
        handlers = discover_intent_handlers(plugin)
        for name, handler, metadata in handlers:
            from trixy_core.nlp.intent_registry import IntentDefinition
            registry.register(IntentDefinition(
                name=name,
                handler=handler,
                slots=metadata.get("slots", {}),
                examples=metadata.get("examples", []),
            ))

        # Intent-Daten erstellen
        data = IntentReceivedData(
            intent="turn_on_device",
            slots={"device": "Licht"},
            original_text="Schalte das Licht ein",
        )

        # Handler aufrufen
        handler = registry.get_handler("turn_on_device")
        assert handler is not None

        result = await handler(data)

        assert handler_called is True
        assert handler_data.intent == "turn_on_device"
        assert result.success is True
        assert "Licht" in result.response_text

    @pytest.mark.asyncio
    async def test_provider_with_registry(self):
        """Provider sollte Intents aus Registry verwenden."""
        registry = IntentRegistry.get_instance()

        # Intents registrieren
        def dummy_handler():
            pass

        registry.register(IntentDefinition(
            name="turn_on_device",
            handler=dummy_handler,
            description="Schaltet ein Gerät ein",
            examples=["Licht an", "Lampe einschalten"],
            slots={"device": IntentSlot("device", str)},
        ))

        # Provider mit Intents initialisieren
        provider = MockNLPProvider()
        await provider.initialize(NLPConfig())

        # Kontext mit verfügbaren Intents
        intents = registry.get_all_as_dict()
        context = NLPContext(
            text="Schalte das Licht ein",
            available_intents=intents,
        )

        result = await provider.process(context)

        assert result.intent == "turn_on_device"