Trixy7/tests/test_conversation.py

# -*- coding: utf-8 -*-
"""
Tests für Conversation Module.
"""

import pytest
import asyncio
from datetime import datetime, timedelta
from unittest.mock import MagicMock, AsyncMock, patch

from trixy_core.conversation.session import (
    ConversationSession,
    SessionConfig,
    SessionState,
    SessionTurn,
    TurnType,
)
from trixy_core.conversation.context import (
    ConversationContext,
    ContextVariable,
    ContextScope,
)


class TestSessionConfig:
    """Tests für SessionConfig."""

    def test_default_config(self):
        """Testet Standard-Konfiguration."""
        config = SessionConfig()
        assert config.input_timeout_seconds == 60.0
        assert config.follow_up_timeout_seconds == 60.0
        assert config.max_turns == 20
        assert config.max_follow_ups == 5


class TestSessionState:
    """Tests für SessionState."""

    def test_states(self):
        """Testet Session-Zustände."""
        assert SessionState.CREATED.value == "created"
        assert SessionState.AWAITING_INPUT.value == "awaiting"
        assert SessionState.PROCESSING.value == "processing"
        assert SessionState.FOLLOW_UP.value == "follow_up"
        assert SessionState.COMPLETED.value == "completed"


class TestTurnType:
    """Tests für TurnType."""

    def test_types(self):
        """Testet Turn-Typen."""
        assert TurnType.USER_SPEECH.value == "user_speech"
        assert TurnType.USER_TEXT.value == "user_text"
        assert TurnType.ASSISTANT_SPEECH.value == "assistant_speech"
        assert TurnType.FOLLOW_UP.value == "follow_up"


class TestSessionTurn:
    """Tests für SessionTurn."""

    def test_turn_creation(self):
        """Testet Turn-Erstellung."""
        turn = SessionTurn(
            turn_id="t1",
            turn_type=TurnType.USER_SPEECH,
            timestamp=datetime.now(),
            content="Hallo Trixy",
            confidence=0.95,
        )
        assert turn.content == "Hallo Trixy"
        assert turn.confidence == 0.95

    def test_turn_to_dict(self):
        """Testet Dictionary-Konvertierung."""
        turn = SessionTurn(
            turn_id="t1",
            turn_type=TurnType.ASSISTANT_SPEECH,
            timestamp=datetime.now(),
            content="Hallo!",
            intent="greeting",
        )
        d = turn.to_dict()
        assert d["turn_id"] == "t1"
        assert d["turn_type"] == "assistant_speech"
        assert d["intent"] == "greeting"


class TestConversationSession:
    """Tests für ConversationSession."""

    @pytest.fixture
    def config(self):
        """Erstellt Test-Konfiguration."""
        return SessionConfig(
            input_timeout_seconds=10.0,
            max_turns=5,
            max_follow_ups=2,
        )

    @pytest.fixture
    def session(self, config):
        """Erstellt Session."""
        return ConversationSession(
            session_id="test-123",
            satellite_id="sat-001",
            wakeword_type="custom",
            config=config,
        )

    def test_session_creation(self, session):
        """Testet Session-Erstellung."""
        assert session.session_id == "test-123"
        assert session.satellite_id == "sat-001"
        assert session.state == SessionState.CREATED
        assert not session.is_completed

    def test_session_start(self, session):
        """Testet Session-Start."""
        session.start()
        assert session.state == SessionState.AWAITING_INPUT
        assert session.started_at is not None

    def test_add_user_turn(self, session):
        """Testet User-Turn hinzufügen."""
        turn = session.add_user_turn(
            content="Wie ist das Wetter?",
            intent="weather_query",
        )

        assert session.turn_count == 1
        assert turn.content == "Wie ist das Wetter?"
        assert session.state == SessionState.PROCESSING

    def test_add_assistant_turn(self, session):
        """Testet Assistenten-Turn hinzufügen."""
        turn = session.add_assistant_turn(
            content="Das Wetter ist schön.",
        )

        assert session.turn_count == 1
        assert session.state == SessionState.RESPONDING

    def test_add_follow_up(self, session):
        """Testet Rückfrage hinzufügen."""
        session.add_user_turn("Mach das Licht an")

        turn = session.add_assistant_turn(
            content="Welches Licht meinst du?",
            is_follow_up=True,
        )

        assert turn.requires_response is True
        assert session.follow_up_count == 1
        assert session.has_pending_follow_up
        assert session.state == SessionState.FOLLOW_UP

    def test_answer_follow_up(self, session):
        """Testet Antwort auf Rückfrage."""
        session.add_user_turn("Mach das Licht an")
        follow_up = session.add_assistant_turn(
            content="Welches Licht?",
            is_follow_up=True,
        )

        answer = session.add_user_turn("Das Wohnzimmerlicht")

        assert follow_up.response_turn_id == answer.turn_id
        assert not session.has_pending_follow_up

    def test_complete_session(self, session):
        """Testet Session-Abschluss."""
        session.start()
        session.complete("Test abgeschlossen")

        assert session.is_completed
        assert session.state == SessionState.COMPLETED
        assert session.completed_at is not None

    def test_timeout_session(self, session):
        """Testet Session-Timeout."""
        session.timeout()

        assert session.is_completed
        assert session.state == SessionState.TIMEOUT

    def test_cancel_session(self, session):
        """Testet Session-Abbruch."""
        session.cancel("Abgebrochen")

        assert session.is_completed
        assert session.state == SessionState.CANCELLED

    def test_user_turns(self, session):
        """Testet User-Turns-Filter."""
        session.add_user_turn("User 1")
        session.add_assistant_turn("Assistent 1")
        session.add_user_turn("User 2")

        user_turns = session.user_turns
        assert len(user_turns) == 2

    def test_assistant_turns(self, session):
        """Testet Assistenten-Turns-Filter."""
        session.add_user_turn("User")
        session.add_assistant_turn("Antwort")
        session.add_assistant_turn("Rückfrage", is_follow_up=True)

        assistant_turns = session.assistant_turns
        assert len(assistant_turns) == 2

    def test_max_turns_limit(self, session):
        """Testet Turn-Limit."""
        for i in range(5):
            session.add_user_turn(f"Turn {i}")

        with pytest.raises(RuntimeError):
            session.add_user_turn("Zu viel")

    def test_context(self, session):
        """Testet Kontext."""
        session.set_context("room", "wohnzimmer")
        assert session.get_context("room") == "wohnzimmer"
        assert session.get_context("missing", "default") == "default"

    def test_to_dict(self, session):
        """Testet Dictionary-Konvertierung."""
        session.add_user_turn("Test")
        d = session.to_dict()

        assert d["session_id"] == "test-123"
        assert d["turn_count"] == 1
        assert "turns" in d

    def test_transcript(self, session):
        """Testet Transkript."""
        session.add_user_turn("Hallo")
        session.add_assistant_turn("Hallo!")
        session.add_assistant_turn("Wie kann ich helfen?", is_follow_up=True)

        transcript = session.get_transcript()
        assert "User: Hallo" in transcript
        assert "Assistent: Hallo!" in transcript
        assert "Rückfrage" in transcript

    def test_duration(self, session):
        """Testet Dauer-Berechnung."""
        assert session.duration_seconds >= 0

        session.complete()
        assert session.duration_seconds >= 0

    def test_is_active(self, session):
        """Testet is_active."""
        assert session.is_active

        session.complete()
        assert not session.is_active


class TestContextScope:
    """Tests für ContextScope."""

    def test_scopes(self):
        """Testet Kontext-Scopes."""
        assert ContextScope.TURN.value == "turn"
        assert ContextScope.SESSION.value == "session"
        assert ContextScope.USER.value == "user"
        assert ContextScope.GLOBAL.value == "global"


class TestContextVariable:
    """Tests für ContextVariable."""

    def test_variable_creation(self):
        """Testet Variable-Erstellung."""
        var = ContextVariable(
            name="test",
            value=42,
            scope=ContextScope.SESSION,
        )
        assert var.name == "test"
        assert var.value == 42

    def test_update(self):
        """Testet Update."""
        var = ContextVariable(name="test", value=1)
        old_time = var.updated_at

        var.update(2)
        assert var.value == 2
        assert var.updated_at >= old_time


class TestConversationContext:
    """Tests für ConversationContext."""

    @pytest.fixture
    def context(self):
        """Erstellt Context."""
        return ConversationContext()

    def test_set_get(self, context):
        """Testet Set/Get."""
        context.set("test", "value")
        assert context.get("test") == "value"

    def test_get_default(self, context):
        """Testet Default-Wert."""
        assert context.get("missing", "default") == "default"

    def test_has(self, context):
        """Testet has."""
        context.set("exists", True)
        assert context.has("exists")
        assert not context.has("missing")

    def test_remove(self, context):
        """Testet remove."""
        context.set("test", "value")
        assert context.remove("test")
        assert not context.has("test")

    def test_turn_scope(self, context):
        """Testet Turn-Scope."""
        context.set("turn_var", "value", ContextScope.TURN)
        assert context.get("turn_var") == "value"

        context.clear_turn()
        assert not context.has("turn_var")

    def test_turn_overrides_session(self, context):
        """Testet Turn überschreibt Session."""
        context.set("var", "session", ContextScope.SESSION)
        context.set("var", "turn", ContextScope.TURN)

        assert context.get("var") == "turn"

    def test_clear_scope(self, context):
        """Testet Scope-Löschen."""
        context.set("a", 1, ContextScope.SESSION)
        context.set("b", 2, ContextScope.USER)

        context.clear_scope(ContextScope.SESSION)

        assert not context.has("a")
        assert context.has("b")

    def test_get_all(self, context):
        """Testet get_all."""
        context.set("a", 1, ContextScope.SESSION)
        context.set("b", 2, ContextScope.TURN)

        all_vars = context.get_all()
        assert all_vars["a"] == 1
        assert all_vars["b"] == 2

    def test_merge(self, context):
        """Testet merge."""
        other = ConversationContext()
        other.set("other_var", "value")

        context.merge(other)
        assert context.get("other_var") == "value"

    def test_intent_helpers(self, context):
        """Testet Intent-Helper."""
        context.set_intent("weather_query", 0.95)
        assert context.get_intent() == "weather_query"

    def test_entity_helpers(self, context):
        """Testet Entity-Helper."""
        context.set_entities({"location": "Berlin"})
        entities = context.get_entities()
        assert entities["location"] == "Berlin"

    def test_slot_helpers(self, context):
        """Testet Slot-Helper."""
        context.set_slot("room", "wohnzimmer")
        context.set_slot("device", "licht")

        assert context.get_slot("room") == "wohnzimmer"
        assert context.slots == {"room": "wohnzimmer", "device": "licht"}

        context.clear_slots()
        assert context.slots == {}

    def test_to_dict(self, context):
        """Testet Dictionary-Konvertierung."""
        context.set("test", "value")
        d = context.to_dict()

        assert "variables" in d
        assert "turn_variables" in d