# -*- coding: utf-8 -*- """ Tests fuer den ActivityTracker Service. Testet Persistenz, Cleanup, Event-Listener und Query-Methoden. """ import asyncio import json import time from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import AsyncMock, MagicMock import pytest from trixy_core.activity.models import ( SatelliteConnectionEntry, ConfigConnectionEntry, TrainingRunEntry, IntentLogEntry, ErrorLogEntry, ConversationEntry, ) from trixy_core.activity.tracker import ActivityTracker, _now_iso, _parse_iso # ============================================================================ # Fixtures # ============================================================================ @pytest.fixture def mock_app(): """Mock-Application mit EventManager und ServiceContainer.""" app = MagicMock() app.events = MagicMock() app.events.register = MagicMock() app.services = MagicMock() app.services.get_service = MagicMock(return_value=None) return app @pytest.fixture def tracker(mock_app, tmp_path): """ActivityTracker mit temporaerem Verzeichnis.""" t = ActivityTracker(mock_app, retention_days=7, data_directory=str(tmp_path / "activity")) return t @pytest.fixture async def started_tracker(tracker): """Gestarteter ActivityTracker (Tasks werden nach Test aufgeraeumt).""" await tracker.start() yield tracker await tracker.stop() # ============================================================================ # Hilfsfunktionen # ============================================================================ def _make_satellite_connected_event(satellite_id="sat-1", alias="Wohnzimmer", room="wohnzimmer", ip="192.168.1.10"): """Erstellt ein Mock-SatelliteConnected Event.""" event = MagicMock() event.satellite_id = satellite_id event.alias = alias event.room = room event.ip_address = ip return event def _make_satellite_disconnected_event(satellite_id="sat-1", reason="timeout"): event = MagicMock() event.satellite_id = satellite_id event.reason = reason return event def _make_intent_handled_event(intent="test_intent", satellite_id="sat-1", room_id="wohnzimmer", text="test anfrage", success=True): event = MagicMock() event.intent = intent event.satellite_id = satellite_id event.room_id = room_id event.original_text = text event.success = success return event def _make_conversation_started_event(conversation_id="conv-1", satellite_id="sat-1"): event = MagicMock() event.conversation_id = conversation_id event.satellite_id = satellite_id return event def _make_conversation_ended_event(conversation_id="conv-1", satellite_id="sat-1", reason="completed"): event = MagicMock() event.conversation_id = conversation_id event.satellite_id = satellite_id event.reason = reason return event # ============================================================================ # Tests: Modelle # ============================================================================ class TestModels: """Tests fuer die Datenklassen.""" def test_satellite_connection_entry_defaults(self): entry = SatelliteConnectionEntry() assert entry.satellite_id == "" assert entry.alias == "" assert entry.room == "" assert entry.connected_at == "" assert entry.disconnected_at == "" assert entry.ip_address == "" def test_config_connection_entry_defaults(self): entry = ConfigConnectionEntry() assert entry.peer_address == "" assert entry.connected_at == "" assert entry.disconnected_at == "" def test_training_run_entry_defaults(self): entry = TrainingRunEntry() assert entry.trainer_id == "" assert entry.duration_seconds == 0.0 assert entry.success is False def test_intent_log_entry_defaults(self): entry = IntentLogEntry() assert entry.intent == "" assert entry.plugin == "" assert entry.success is True def test_error_log_entry_defaults(self): entry = ErrorLogEntry() assert entry.source == "" assert entry.message == "" def test_conversation_entry_defaults(self): entry = ConversationEntry() assert entry.session_id == "" assert entry.turns == [] # ============================================================================ # Tests: Hilfsfunktionen # ============================================================================ class TestHelpers: """Tests fuer _now_iso und _parse_iso.""" def test_now_iso_format(self): result = _now_iso() # Muss als ISO parsbar sein dt = datetime.fromisoformat(result) assert dt.tzinfo is not None # Muss UTC enthalten def test_parse_iso_valid(self): iso = "2026-03-15T10:30:00+00:00" dt = _parse_iso(iso) assert dt is not None assert dt.year == 2026 assert dt.month == 3 def test_parse_iso_empty(self): assert _parse_iso("") is None def test_parse_iso_invalid(self): assert _parse_iso("not-a-date") is None # ============================================================================ # Tests: Lifecycle # ============================================================================ class TestLifecycle: """Tests fuer Start/Stop.""" @pytest.mark.asyncio async def test_start_creates_directory(self, tracker, tmp_path): """Start erstellt data/activity/ Verzeichnis.""" await tracker.start() assert (tmp_path / "activity").exists() await tracker.stop() @pytest.mark.asyncio async def test_start_registers_event_listeners(self, tracker): """Start registriert alle Event-Listener.""" await tracker.start() calls = tracker._application.events.register.call_args_list event_names = [c[0][0] for c in calls] assert "satellite_connected" in event_names assert "satellite_disconnected" in event_names assert "intent_handled" in event_names assert "conversation_started" in event_names assert "conversation_ended" in event_names await tracker.stop() @pytest.mark.asyncio async def test_stop_cancels_tasks(self, tracker): """Stop beendet Flush- und Cleanup-Tasks.""" await tracker.start() assert tracker._flush_task is not None assert tracker._cleanup_task is not None await tracker.stop() assert tracker._flush_task is None assert tracker._cleanup_task is None # ============================================================================ # Tests: Event-Listener # ============================================================================ class TestEventListeners: """Tests fuer Event-Handler.""" @pytest.mark.asyncio async def test_satellite_connected(self, started_tracker): """satellite_connected erstellt neuen Eintrag.""" event = _make_satellite_connected_event() await started_tracker._on_satellite_connected("satellite_connected", event) assert len(started_tracker._satellite_connections) == 1 entry = started_tracker._satellite_connections[0] assert entry["satellite_id"] == "sat-1" assert entry["alias"] == "Wohnzimmer" assert entry["room"] == "wohnzimmer" assert entry["ip_address"] == "192.168.1.10" assert entry["connected_at"] != "" assert entry["disconnected_at"] == "" @pytest.mark.asyncio async def test_satellite_disconnected(self, started_tracker): """satellite_disconnected setzt disconnected_at.""" event_conn = _make_satellite_connected_event() await started_tracker._on_satellite_connected("satellite_connected", event_conn) event_disc = _make_satellite_disconnected_event() await started_tracker._on_satellite_disconnected("satellite_disconnected", event_disc) entry = started_tracker._satellite_connections[0] assert entry["disconnected_at"] != "" @pytest.mark.asyncio async def test_satellite_disconnected_no_match(self, started_tracker): """satellite_disconnected bei unbekanntem Satellite aendert nichts.""" event = _make_satellite_disconnected_event(satellite_id="unknown") await started_tracker._on_satellite_disconnected("satellite_disconnected", event) # Kein Fehler, keine Aenderung @pytest.mark.asyncio async def test_intent_handled(self, started_tracker): """intent_handled erstellt neuen Intent-Log-Eintrag.""" event = _make_intent_handled_event() await started_tracker._on_intent_handled("intent_handled", event) assert len(started_tracker._intent_log) == 1 entry = started_tracker._intent_log[0] assert entry["intent"] == "test_intent" assert entry["satellite_id"] == "sat-1" assert entry["room_id"] == "wohnzimmer" assert entry["original_text"] == "test anfrage" assert entry["success"] is True @pytest.mark.asyncio async def test_conversation_started(self, started_tracker): """conversation_started erstellt neuen Conversation-Eintrag.""" event = _make_conversation_started_event() await started_tracker._on_conversation_started("conversation_started", event) assert len(started_tracker._conversations) == 1 entry = started_tracker._conversations[0] assert entry["session_id"] == "conv-1" assert entry["satellite_id"] == "sat-1" assert entry["started_at"] != "" assert entry["completed_at"] == "" @pytest.mark.asyncio async def test_conversation_ended(self, started_tracker): """conversation_ended setzt completed_at.""" event_start = _make_conversation_started_event() await started_tracker._on_conversation_started("conversation_started", event_start) event_end = _make_conversation_ended_event() await started_tracker._on_conversation_ended("conversation_ended", event_end) entry = started_tracker._conversations[0] assert entry["completed_at"] != "" # ============================================================================ # Tests: Oeffentliche Logging-Methoden # ============================================================================ class TestLoggingMethods: """Tests fuer externe Logging-Methoden.""" @pytest.mark.asyncio async def test_log_config_connection(self, started_tracker): """log_config_connection erstellt neuen Eintrag.""" started_tracker.log_config_connection("192.168.1.100:54321") assert len(started_tracker._config_connections) == 1 entry = started_tracker._config_connections[0] assert entry["peer_address"] == "192.168.1.100:54321" assert entry["connected_at"] != "" @pytest.mark.asyncio async def test_log_config_disconnection(self, started_tracker): """log_config_disconnection setzt disconnected_at.""" started_tracker.log_config_connection("192.168.1.100:54321") started_tracker.log_config_disconnection("192.168.1.100:54321") entry = started_tracker._config_connections[0] assert entry["disconnected_at"] != "" @pytest.mark.asyncio async def test_log_training_start(self, started_tracker): """log_training_start erstellt neuen Training-Eintrag.""" started_tracker.log_training_start("wakeword", "WakewordModel") assert len(started_tracker._training_runs) == 1 entry = started_tracker._training_runs[0] assert entry["trainer_id"] == "wakeword" assert entry["model_name"] == "WakewordModel" @pytest.mark.asyncio async def test_log_training_complete(self, started_tracker): """log_training_complete setzt completed_at und Ergebnis.""" started_tracker.log_training_start("wakeword", "WakewordModel") started_tracker.log_training_complete("wakeword", success=True) entry = started_tracker._training_runs[0] assert entry["completed_at"] != "" assert entry["success"] is True assert entry["duration_seconds"] >= 0 @pytest.mark.asyncio async def test_log_training_complete_failed(self, started_tracker): """log_training_complete mit Fehler.""" started_tracker.log_training_start("voice", "VoiceModel") started_tracker.log_training_complete("voice", success=False, error="GPU Fehler") entry = started_tracker._training_runs[0] assert entry["success"] is False assert entry["error"] == "GPU Fehler" @pytest.mark.asyncio async def test_log_error(self, started_tracker): """log_error erstellt neuen Fehler-Eintrag.""" started_tracker.log_error("NetworkService", "Timeout", satellite_id="sat-1") assert len(started_tracker._error_log) == 1 entry = started_tracker._error_log[0] assert entry["source"] == "NetworkService" assert entry["message"] == "Timeout" assert entry["satellite_id"] == "sat-1" # ============================================================================ # Tests: Query-Methoden # ============================================================================ class TestQueryMethods: """Tests fuer Query-Methoden.""" @pytest.mark.asyncio async def test_get_last_intents(self, started_tracker): """get_last_intents liefert die letzten N Eintraege.""" for i in range(10): event = _make_intent_handled_event(intent=f"intent_{i}") await started_tracker._on_intent_handled("intent_handled", event) result = started_tracker.get_last_intents(3) assert len(result) == 3 assert result[0]["intent"] == "intent_7" assert result[2]["intent"] == "intent_9" @pytest.mark.asyncio async def test_get_last_intents_empty(self, started_tracker): """get_last_intents bei leerer Liste.""" result = started_tracker.get_last_intents(5) assert result == [] @pytest.mark.asyncio async def test_get_last_intents_by_room(self, started_tracker): """get_last_intents_by_room filtert nach Raum.""" for room in ["wohnzimmer", "kueche", "wohnzimmer"]: event = _make_intent_handled_event(room_id=room) await started_tracker._on_intent_handled("intent_handled", event) result = started_tracker.get_last_intents_by_room("wohnzimmer", 10) assert len(result) == 2 result = started_tracker.get_last_intents_by_room("kueche", 10) assert len(result) == 1 @pytest.mark.asyncio async def test_get_last_conversations(self, started_tracker): """get_last_conversations liefert die letzten N Conversations.""" for i in range(5): event = _make_conversation_started_event(conversation_id=f"conv-{i}") await started_tracker._on_conversation_started("conversation_started", event) result = started_tracker.get_last_conversations(2) assert len(result) == 2 assert result[0]["session_id"] == "conv-3" @pytest.mark.asyncio async def test_get_last_errors(self, started_tracker): """get_last_errors liefert die letzten N Fehler.""" for i in range(5): started_tracker.log_error("src", f"msg_{i}") result = started_tracker.get_last_errors(3) assert len(result) == 3 assert result[0]["message"] == "msg_2" @pytest.mark.asyncio async def test_get_error_count_since(self, started_tracker): """get_error_count_since zaehlt aktuelle Fehler.""" started_tracker.log_error("src", "recent") assert started_tracker.get_error_count_since(1) == 1 assert started_tracker.get_error_count_since(24) == 1 @pytest.mark.asyncio async def test_get_last_training(self, started_tracker): """get_last_training liefert den letzten Trainings-Lauf.""" started_tracker.log_training_start("wakeword", "WW-Model") started_tracker.log_training_start("voice", "V-Model") result = started_tracker.get_last_training() assert result is not None assert result["trainer_id"] == "voice" result = started_tracker.get_last_training("wakeword") assert result is not None assert result["trainer_id"] == "wakeword" result = started_tracker.get_last_training("unknown") assert result is None @pytest.mark.asyncio async def test_get_last_config_connection(self, started_tracker): """get_last_config_connection liefert den letzten Eintrag.""" assert started_tracker.get_last_config_connection() is None started_tracker.log_config_connection("10.0.0.1:1234") result = started_tracker.get_last_config_connection() assert result is not None assert result["peer_address"] == "10.0.0.1:1234" @pytest.mark.asyncio async def test_get_satellite_connection_history(self, started_tracker): """get_satellite_connection_history filtert nach Alias (case-insensitive).""" event1 = _make_satellite_connected_event(satellite_id="s1", alias="Wohnzimmer") event2 = _make_satellite_connected_event(satellite_id="s2", alias="Kueche") event3 = _make_satellite_connected_event(satellite_id="s1", alias="Wohnzimmer") await started_tracker._on_satellite_connected("satellite_connected", event1) await started_tracker._on_satellite_connected("satellite_connected", event2) await started_tracker._on_satellite_connected("satellite_connected", event3) result = started_tracker.get_satellite_connection_history("wohnzimmer") assert len(result) == 2 result = started_tracker.get_satellite_connection_history("KUECHE") assert len(result) == 1 @pytest.mark.asyncio async def test_get_intent_stats(self, started_tracker): """get_intent_stats zaehlt Intents der letzten N Tage.""" for _ in range(3): event = _make_intent_handled_event(intent="wetter") await started_tracker._on_intent_handled("intent_handled", event) for _ in range(2): event = _make_intent_handled_event(intent="timer") await started_tracker._on_intent_handled("intent_handled", event) stats = started_tracker.get_intent_stats(7) assert stats["wetter"] == 3 assert stats["timer"] == 2 @pytest.mark.asyncio async def test_clear_all(self, started_tracker): """clear_all loescht alle Daten.""" started_tracker.log_error("src", "msg") started_tracker.log_config_connection("peer") event = _make_intent_handled_event() await started_tracker._on_intent_handled("intent_handled", event) started_tracker.clear_all() assert started_tracker._error_log == [] assert started_tracker._config_connections == [] assert started_tracker._intent_log == [] assert started_tracker._satellite_connections == [] assert started_tracker._training_runs == [] assert started_tracker._conversations == [] # ============================================================================ # Tests: Persistenz # ============================================================================ class TestPersistence: """Tests fuer JSON-Persistenz.""" @pytest.mark.asyncio async def test_flush_creates_json_files(self, tracker, tmp_path): """Flush erstellt JSON-Dateien.""" await tracker.start() tracker.log_error("src", "msg") tracker._flush_all() error_file = tmp_path / "activity" / "error_log.json" assert error_file.exists() data = json.loads(error_file.read_text()) assert len(data) == 1 assert data[0]["source"] == "src" await tracker.stop() @pytest.mark.asyncio async def test_load_on_start(self, mock_app, tmp_path): """Start laedt bestehende JSON-Dateien.""" # Datei manuell erstellen activity_dir = tmp_path / "activity" activity_dir.mkdir() error_file = activity_dir / "error_log.json" error_file.write_text(json.dumps([ {"timestamp": _now_iso(), "source": "test", "message": "saved", "satellite_id": ""} ])) tracker = ActivityTracker(mock_app, data_directory=str(activity_dir)) await tracker.start() assert len(tracker._error_log) == 1 assert tracker._error_log[0]["source"] == "test" await tracker.stop() @pytest.mark.asyncio async def test_flush_only_when_dirty(self, tracker, tmp_path): """Flush schreibt nur wenn Daten geaendert wurden.""" await tracker.start() tracker._dirty = False tracker._flush_all() # Datei sollte nicht existieren (kein Flush ohne dirty) error_file = tmp_path / "activity" / "error_log.json" assert not error_file.exists() await tracker.stop() @pytest.mark.asyncio async def test_corrupted_json_handled(self, mock_app, tmp_path): """Korrupte JSON-Dateien werden graceful behandelt.""" activity_dir = tmp_path / "activity" activity_dir.mkdir() (activity_dir / "error_log.json").write_text("NOT VALID JSON{{{") tracker = ActivityTracker(mock_app, data_directory=str(activity_dir)) await tracker.start() assert tracker._error_log == [] await tracker.stop() # ============================================================================ # Tests: Retention / Cleanup # ============================================================================ class TestRetention: """Tests fuer Retention-Cleanup.""" @pytest.mark.asyncio async def test_cleanup_removes_old_entries(self, mock_app, tmp_path): """Cleanup entfernt Eintraege aelter als retention_days.""" activity_dir = tmp_path / "activity" activity_dir.mkdir() old_ts = (datetime.now(timezone.utc) - timedelta(days=10)).isoformat() recent_ts = _now_iso() error_file = activity_dir / "error_log.json" error_file.write_text(json.dumps([ {"timestamp": old_ts, "source": "old", "message": "alt", "satellite_id": ""}, {"timestamp": recent_ts, "source": "new", "message": "neu", "satellite_id": ""}, ])) tracker = ActivityTracker(mock_app, retention_days=7, data_directory=str(activity_dir)) await tracker.start() # Alter Eintrag sollte entfernt worden sein assert len(tracker._error_log) == 1 assert tracker._error_log[0]["source"] == "new" await tracker.stop() @pytest.mark.asyncio async def test_trim_list_max_entries(self, started_tracker): """trim_list begrenzt auf MAX_ENTRIES.""" # Direkt die interne Liste fuellen started_tracker._intent_log = [ {"timestamp": _now_iso(), "intent": f"i_{i}", "plugin": "", "satellite_id": "", "room_id": "", "original_text": "", "success": True} for i in range(10005) ] started_tracker._trim_list(started_tracker._intent_log) assert len(started_tracker._intent_log) == 10000 # Die aeltesten 5 sollten entfernt worden sein assert started_tracker._intent_log[0]["intent"] == "i_5"