Trixy7/tests/test_events.py

# -*- coding: utf-8 -*-
"""
Unit-Tests für trixy_core.events.
"""

import pytest
from unittest.mock import MagicMock, AsyncMock
from datetime import datetime

from trixy_core.events.enums import EventPriority
from trixy_core.events.event_data.base import EventData
from trixy_core.events.event_data.basic import (
    SatelliteConnected, SatelliteDisconnected, MusicSocketConnected,
    WakewordReceived, RawAudioInputReceived,
    SpeechRecognized, IntentReceived,
    ConversationStarted, ConversationEnded,
    PluginLoaded, PluginUnloaded, SystemShutdown
)
from trixy_core.events.decorators import TrixyEvent, event_handler, get_event_handlers
from trixy_core.events.eventmanager import EventManager


class TestEventPriority:
    """Tests für EventPriority."""

    def test_priority_ordering(self):
        """Testet die Prioritätsreihenfolge."""
        assert EventPriority.SYSTEM < EventPriority.HIGHEST
        assert EventPriority.HIGHEST < EventPriority.HIGH
        assert EventPriority.HIGH < EventPriority.NORMAL
        assert EventPriority.NORMAL < EventPriority.LOW
        assert EventPriority.LOW < EventPriority.LOWEST
        assert EventPriority.LOWEST < EventPriority.MONITOR

    def test_priority_values(self):
        """Testet die Prioritätswerte."""
        assert EventPriority.SYSTEM == 0
        assert EventPriority.NORMAL == 30
        assert EventPriority.MONITOR == 100


class TestEventData:
    """Tests für EventData-Basisklasse."""

    def test_event_data_creation(self):
        """Testet die Event-Data-Erstellung."""
        event = EventData()
        assert event.event_id is not None
        assert isinstance(event.timestamp, datetime)
        assert event.source == ""
        assert event.cancelled is False

    def test_event_data_cancel(self):
        """Testet das Abbrechen eines Events."""
        event = EventData()
        assert event.is_cancelled() is False
        event.cancel()
        assert event.is_cancelled() is True

    def test_event_name_snake_case(self):
        """Testet die Event-Name-Konvertierung."""
        assert EventData.event_name() == "event_data"

    def test_custom_source(self):
        """Testet benutzerdefinierte Quelle."""
        event = EventData(source="TestSource")
        assert event.source == "TestSource"

    def test_metadata(self):
        """Testet Metadaten."""
        event = EventData(metadata={"key": "value"})
        assert event.metadata == {"key": "value"}


class TestBasicEventData:
    """Tests für grundlegende Event-Datenklassen."""

    def test_satellite_connected(self):
        """Testet SatelliteConnected."""
        event = SatelliteConnected(
            satellite_id="sat-1",
            room="wohnzimmer",
            alias="Echo 1",
            mac_address="AA:BB:CC:DD:EE:FF",
            ip_address="192.168.1.100"
        )
        assert event.satellite_id == "sat-1"
        assert event.room == "wohnzimmer"
        assert event.alias == "Echo 1"
        assert SatelliteConnected.event_name() == "satellite_connected"

    def test_satellite_disconnected(self):
        """Testet SatelliteDisconnected."""
        event = SatelliteDisconnected(
            satellite_id="sat-1",
            reason="Connection lost"
        )
        assert event.satellite_id == "sat-1"
        assert event.reason == "Connection lost"

    def test_music_socket_connected(self):
        """Testet MusicSocketConnected."""
        event = MusicSocketConnected(
            satellite_id="sat-1",
            alias="Echo 1",
            room="wohnzimmer"
        )
        assert event.satellite_id == "sat-1"
        assert event.alias == "Echo 1"
        assert event.room == "wohnzimmer"
        assert MusicSocketConnected.event_name() == "music_socket_connected"

    def test_wakeword_received(self):
        """Testet WakewordReceived."""
        event = WakewordReceived(
            satellite_id="sat-1",
            wakeword_model="custom",
            confidence=0.95,
            audio_level=0.7
        )
        assert event.confidence == 0.95
        assert event.audio_level == 0.7
        assert WakewordReceived.event_name() == "wakeword_received"

    def test_raw_audio_input_received(self):
        """Testet RawAudioInputReceived."""
        audio = b"\x00\x00" * 100
        event = RawAudioInputReceived(
            audio_data=audio,
            sample_rate=16000,
            duration_seconds=1.0
        )
        assert event.audio_data == audio
        assert event.sample_rate == 16000

    def test_speech_recognized(self):
        """Testet SpeechRecognized."""
        event = SpeechRecognized(
            text="Hallo Trixy",
            confidence=0.98,
            language="de"
        )
        assert event.text == "Hallo Trixy"
        assert event.language == "de"

    def test_intent_received(self):
        """Testet IntentReceived."""
        event = IntentReceived(
            intent="turn_on_light",
            confidence=0.9,
            slots={"room": "wohnzimmer"}
        )
        assert event.intent == "turn_on_light"
        assert event.slots["room"] == "wohnzimmer"

    def test_conversation_started(self):
        """Testet ConversationStarted."""
        event = ConversationStarted(
            satellite_id="sat-1",
            conversation_id="conv-123"
        )
        assert event.conversation_id == "conv-123"

    def test_conversation_ended(self):
        """Testet ConversationEnded."""
        event = ConversationEnded(
            conversation_id="conv-123",
            reason="completed"
        )
        assert event.reason == "completed"

    def test_plugin_loaded(self):
        """Testet PluginLoaded."""
        event = PluginLoaded(
            plugin_name="TestPlugin",
            plugin_version="1.0.0"
        )
        assert event.plugin_name == "TestPlugin"

    def test_system_shutdown(self):
        """Testet SystemShutdown."""
        event = SystemShutdown(
            reason="User request",
            force=True
        )
        assert event.reason == "User request"
        assert event.force is True


class TestEventDecorators:
    """Tests für Event-Dekoratoren."""

    def test_trixy_event_decorator(self):
        """Testet den TrixyEvent-Dekorator."""
        class TestHandler:
            @TrixyEvent(["wakeword_received", "satellite_connected"])
            def on_events(self, event_name, event_data):
                pass

        handlers = get_event_handlers(TestHandler())
        assert len(handlers) == 2
        event_names = [h["event_name"] for h in handlers]
        assert "wakeword_received" in event_names
        assert "satellite_connected" in event_names

    def test_trixy_event_with_priority(self):
        """Testet TrixyEvent mit Priorität."""
        class TestHandler:
            @TrixyEvent(["test_event"], priority=EventPriority.HIGH)
            def on_event(self, event_name, event_data):
                pass

        handlers = get_event_handlers(TestHandler())
        assert len(handlers) == 1
        assert handlers[0]["priority"] == EventPriority.HIGH

    def test_event_handler_decorator(self):
        """Testet den event_handler-Dekorator."""
        class TestHandler:
            @event_handler(WakewordReceived)
            async def on_wakeword(self, event):
                pass

        handlers = get_event_handlers(TestHandler())
        assert len(handlers) == 1
        assert handlers[0]["event_name"] == "wakeword_received"
        assert handlers[0]["event_class"] == WakewordReceived


class TestEventManager:
    """Tests für EventManager."""

    @pytest.fixture
    def mock_application(self):
        """Erstellt eine Mock-Anwendung."""
        return MagicMock()

    @pytest.fixture
    def manager(self, mock_application):
        """Erstellt einen EventManager."""
        return EventManager(mock_application)

    def test_manager_initialization(self, manager):
        """Testet die Manager-Initialisierung."""
        assert manager._handlers == {}
        assert manager._global_handlers == []

    def test_register_handler(self, manager):
        """Testet die Handler-Registrierung."""
        def handler(event_name, event_data):
            pass

        manager.register("test_event", handler)
        assert manager.has_handlers("test_event")

    def test_unregister_handler(self, manager):
        """Testet die Handler-Deregistrierung."""
        def handler(event_name, event_data):
            pass

        manager.register("test_event", handler)
        result = manager.unregister("test_event", handler)

        assert result is True
        assert not manager.has_handlers("test_event")

    def test_register_global_handler(self, manager):
        """Testet die globale Handler-Registrierung."""
        def handler(event_name, event_data):
            pass

        manager.register_global(handler)
        assert len(manager._global_handlers) == 1

    @pytest.mark.asyncio
    async def test_trigger_event(self, manager):
        """Testet das Auslösen eines Events."""
        called = []

        def handler(event_name, event_data):
            called.append((event_name, event_data))

        manager.register("test_event", handler)
        event_data = EventData()

        result = await manager.trigger("test_event", event_data)

        assert len(called) == 1
        assert called[0][0] == "test_event"
        assert called[0][1] == event_data

    @pytest.mark.asyncio
    async def test_trigger_async_handler(self, manager):
        """Testet das Auslösen mit async Handler."""
        called = []

        async def async_handler(event_name, event_data):
            called.append(event_name)

        manager.register("async_event", async_handler)
        await manager.trigger("async_event", EventData())

        assert "async_event" in called

    @pytest.mark.asyncio
    async def test_event_cancellation(self, manager):
        """Testet die Event-Abbruchfunktion."""
        order = []

        def first_handler(event_name, event_data):
            order.append("first")
            event_data.cancel()

        def second_handler(event_name, event_data):
            order.append("second")

        manager.register("cancel_test", first_handler, EventPriority.HIGH)
        manager.register("cancel_test", second_handler, EventPriority.NORMAL)

        event = EventData()
        await manager.trigger("cancel_test", event)

        assert "first" in order
        assert "second" not in order  # Wurde abgebrochen

    @pytest.mark.asyncio
    async def test_monitor_receives_cancelled_events(self, manager):
        """Testet, dass MONITOR-Handler abgebrochene Events erhalten."""
        received = []

        def canceller(event_name, event_data):
            event_data.cancel()

        def monitor(event_name, event_data):
            received.append(event_name)

        manager.register("monitor_test", canceller, EventPriority.HIGH)
        manager.register("monitor_test", monitor, EventPriority.MONITOR)

        await manager.trigger("monitor_test", EventData())

        assert "monitor_test" in received

    @pytest.mark.asyncio
    async def test_priority_ordering(self, manager):
        """Testet die Prioritätsreihenfolge."""
        order = []

        def low_handler(event_name, event_data):
            order.append("low")

        def high_handler(event_name, event_data):
            order.append("high")

        manager.register("priority_test", low_handler, EventPriority.LOW)
        manager.register("priority_test", high_handler, EventPriority.HIGH)

        await manager.trigger("priority_test", EventData())

        assert order == ["high", "low"]

    def test_get_registered_events(self, manager):
        """Testet das Abrufen registrierter Events."""
        manager.register("event1", lambda *args: None)
        manager.register("event2", lambda *args: None)

        events = manager.get_registered_events()

        assert "event1" in events
        assert "event2" in events

    def test_clear_all(self, manager):
        """Testet das Löschen aller Handler."""
        manager.register("event1", lambda *args: None)
        manager.register_global(lambda *args: None)

        manager.clear_all()

        assert manager._handlers == {}
        assert manager._global_handlers == []

    def test_register_object(self, manager):
        """Testet die Registrierung eines Objekts."""
        class Handler:
            @TrixyEvent(["test_event"])
            def on_test(self, event_name, event_data):
                pass

        handler = Handler()
        manager.register_object(handler)

        assert manager.has_handlers("test_event")