# -*- coding: utf-8 -*- """ Tests fuer den SyncState der EPG-Synchronisation. Testet Persistenz, Fetch-Logik, Enrichment-Warteschlange und Fehlertoleranzen. """ from __future__ import annotations import json from datetime import datetime, timedelta, timezone from pathlib import Path import pytest from plugins.tv_program.sync.sync_state import SyncState # ============================================================================ # SyncState - JSON I/O # ============================================================================ class TestSyncStateLadenSpeichern: """Tests fuer die JSON-Persistenz des Sync-States.""" def test_load_fehlende_datei_standardwerte(self, tmp_path: Path) -> None: """Fehlende Datei setzt Standardwerte, kein Fehler.""" state = SyncState(tmp_path) state.load() assert state.last_epg_fetch == "" assert state.enrichment_queue == [] assert state.enriched_titles == {} def test_save_und_load_roundtrip(self, tmp_path: Path) -> None: """Gespeicherte Daten koennen wieder geladen werden.""" state = SyncState(tmp_path) state.last_epg_fetch = "2025-01-15T10:00:00+00:00" state.enrichment_queue = ["tatort", "tagesschau"] state.enriched_titles = {"krimi": "tmdb"} state.next_sync_after = "2025-01-15T16:00:00+00:00" state.save() # Neuen State aus gleicher Datei laden state2 = SyncState(tmp_path) state2.load() assert state2.last_epg_fetch == "2025-01-15T10:00:00+00:00" assert state2.enrichment_queue == ["tatort", "tagesschau"] assert state2.enriched_titles == {"krimi": "tmdb"} assert state2.next_sync_after == "2025-01-15T16:00:00+00:00" def test_load_korrupte_datei_toleriert(self, tmp_path: Path) -> None: """Korrupte JSON-Datei wird toleriert, Standardwerte werden gesetzt.""" datei = tmp_path / SyncState.FILE_NAME datei.write_text("das ist kein JSON!!!", encoding="utf-8") state = SyncState(tmp_path) state.load() # Darf keinen Fehler werfen assert state.last_epg_fetch == "" assert state.enrichment_queue == [] def test_save_erzeugt_datei(self, tmp_path: Path) -> None: """Nach dem Speichern existiert die State-Datei.""" state = SyncState(tmp_path) state.save() assert (tmp_path / SyncState.FILE_NAME).exists() def test_save_atomisch_ueber_tmp_datei(self, tmp_path: Path) -> None: """Save schreibt zuerst eine .tmp-Datei und benennt dann um.""" state = SyncState(tmp_path) state.last_epg_fetch = "2025-01-01T00:00:00+00:00" state.save() # Die finale Datei muss existieren assert (tmp_path / SyncState.FILE_NAME).exists() # Die tmp-Datei darf nicht mehr existieren (wurde umbenannt) assert not (tmp_path / "sync_state.tmp").exists() def test_save_erzeugt_gueltiges_json(self, tmp_path: Path) -> None: """Die gespeicherte Datei enthaelt gueltiges JSON.""" state = SyncState(tmp_path) state.enrichment_queue = ["test-id"] state.save() datei = tmp_path / SyncState.FILE_NAME inhalt = json.loads(datei.read_text(encoding="utf-8")) assert isinstance(inhalt, dict) assert inhalt["enrichment_queue"] == ["test-id"] # ============================================================================ # SyncState - needs_epg_fetch() # ============================================================================ class TestSyncStateNeedsEpgFetch: """Tests fuer die EPG-Fetch-Pruefung.""" def test_nie_synchronisiert_braucht_fetch(self, tmp_path: Path) -> None: """Ohne vorherigen Fetch wird True zurueckgegeben.""" state = SyncState(tmp_path) assert state.needs_epg_fetch() is True def test_kuerzlich_synchronisiert_kein_fetch(self, tmp_path: Path) -> None: """Nach kuerzlichem Fetch mit Zukunfts-Sync wird False zurueckgegeben.""" state = SyncState(tmp_path) jetzt = datetime.now(tz=timezone.utc) state.last_epg_fetch = jetzt.isoformat() state.next_sync_after = (jetzt + timedelta(hours=6)).isoformat() assert state.needs_epg_fetch() is False def test_next_sync_after_vergangen_braucht_fetch(self, tmp_path: Path) -> None: """Wenn next_sync_after in der Vergangenheit liegt, wird True zurueckgegeben.""" state = SyncState(tmp_path) vergangen = datetime.now(tz=timezone.utc) - timedelta(hours=1) state.last_epg_fetch = (vergangen - timedelta(hours=6)).isoformat() state.next_sync_after = vergangen.isoformat() assert state.needs_epg_fetch() is True def test_last_fetch_gesetzt_ohne_next_sync_braucht_fetch(self, tmp_path: Path) -> None: """Mit last_epg_fetch aber ohne next_sync_after wird True zurueckgegeben.""" state = SyncState(tmp_path) state.last_epg_fetch = datetime.now(tz=timezone.utc).isoformat() state.next_sync_after = "" assert state.needs_epg_fetch() is True def test_ungueltiger_zeitstempel_braucht_fetch(self, tmp_path: Path) -> None: """Ungueltiger Zeitstempel fuehrt sicherheitshalber zu True.""" state = SyncState(tmp_path) state.last_epg_fetch = "gestern" state.next_sync_after = "ungueltig" assert state.needs_epg_fetch() is True # ============================================================================ # SyncState - mark_epg_fetched() # ============================================================================ class TestSyncStateMarkEpgFetched: """Tests fuer das Markieren eines abgeschlossenen EPG-Fetch.""" def test_mark_setzt_zeitstempel(self, tmp_path: Path) -> None: """mark_epg_fetched setzt last_epg_fetch auf aktuelle Zeit.""" state = SyncState(tmp_path) vorher = datetime.now(tz=timezone.utc) state.mark_epg_fetched(retention_days=14, min_interval_hours=6) nachher = datetime.now(tz=timezone.utc) fetch_time = datetime.fromisoformat(state.last_epg_fetch) assert vorher <= fetch_time <= nachher def test_mark_setzt_next_sync_after(self, tmp_path: Path) -> None: """mark_epg_fetched berechnet next_sync_after korrekt.""" state = SyncState(tmp_path) state.mark_epg_fetched(retention_days=14, min_interval_hours=6) fetch_time = datetime.fromisoformat(state.last_epg_fetch) next_sync = datetime.fromisoformat(state.next_sync_after) differenz = next_sync - fetch_time # Sollte ungefaehr 6 Stunden sein assert timedelta(hours=5, minutes=59) <= differenz <= timedelta(hours=6, minutes=1) def test_mark_speichert_automatisch(self, tmp_path: Path) -> None: """mark_epg_fetched ruft save() auf.""" state = SyncState(tmp_path) state.mark_epg_fetched(retention_days=14, min_interval_hours=6) # Datei muss existieren (save wurde aufgerufen) assert (tmp_path / SyncState.FILE_NAME).exists() def test_mark_danach_kein_fetch_noetig(self, tmp_path: Path) -> None: """Nach mark_epg_fetched ist needs_epg_fetch() False.""" state = SyncState(tmp_path) state.mark_epg_fetched(retention_days=14, min_interval_hours=6) assert state.needs_epg_fetch() is False # ============================================================================ # SyncState - needs_enrichment() # ============================================================================ class TestSyncStateNeedsEnrichment: """Tests fuer die Enrichment-Warteschlangen-Pruefung.""" def test_leere_queue_kein_enrichment(self, tmp_path: Path) -> None: """Leere Warteschlange gibt False zurueck.""" state = SyncState(tmp_path) assert state.needs_enrichment() is False def test_nicht_leere_queue_braucht_enrichment(self, tmp_path: Path) -> None: """Nicht-leere Warteschlange gibt True zurueck.""" state = SyncState(tmp_path) state.enrichment_queue = ["tatort"] assert state.needs_enrichment() is True # ============================================================================ # SyncState - add_to_enrichment_queue() # ============================================================================ class TestSyncStateAddToEnrichmentQueue: """Tests fuer das Hinzufuegen zur Enrichment-Warteschlange.""" def test_neue_ids_werden_hinzugefuegt(self, tmp_path: Path) -> None: """Neue Serien-IDs werden zur Queue hinzugefuegt.""" state = SyncState(tmp_path) state.add_to_enrichment_queue(["tatort", "tagesschau"]) assert "tatort" in state.enrichment_queue assert "tagesschau" in state.enrichment_queue def test_bereits_angereicherte_werden_uebersprungen(self, tmp_path: Path) -> None: """Bereits angereicherte IDs werden nicht zur Queue hinzugefuegt.""" state = SyncState(tmp_path) state.enriched_titles = {"tatort": "tmdb"} state.add_to_enrichment_queue(["tatort", "tagesschau"]) assert "tatort" not in state.enrichment_queue assert "tagesschau" in state.enrichment_queue def test_duplikate_in_queue_werden_verhindert(self, tmp_path: Path) -> None: """Bereits in der Queue vorhandene IDs werden nicht dupliziert.""" state = SyncState(tmp_path) state.enrichment_queue = ["tatort"] state.add_to_enrichment_queue(["tatort", "tagesschau"]) assert state.enrichment_queue.count("tatort") == 1 assert "tagesschau" in state.enrichment_queue def test_hinzufuegen_speichert_bei_aenderung(self, tmp_path: Path) -> None: """Bei neuen Eintraegen wird automatisch gespeichert.""" state = SyncState(tmp_path) state.add_to_enrichment_queue(["tatort"]) assert (tmp_path / SyncState.FILE_NAME).exists() def test_hinzufuegen_leere_liste_kein_save(self, tmp_path: Path) -> None: """Leere Liste fuehrt nicht zum Speichern.""" state = SyncState(tmp_path) state.add_to_enrichment_queue([]) assert not (tmp_path / SyncState.FILE_NAME).exists() # ============================================================================ # SyncState - mark_title_enriched() # ============================================================================ class TestSyncStateMarkTitleEnriched: """Tests fuer das Markieren einzelner Titel als angereichert.""" def test_entfernt_aus_queue(self, tmp_path: Path) -> None: """Die markierte ID wird aus der Queue entfernt.""" state = SyncState(tmp_path) state.enrichment_queue = ["tatort", "tagesschau"] state.mark_title_enriched("tatort", "tmdb") assert "tatort" not in state.enrichment_queue assert "tagesschau" in state.enrichment_queue def test_fuegt_zu_enriched_hinzu(self, tmp_path: Path) -> None: """Die markierte ID wird zu enriched_titles hinzugefuegt.""" state = SyncState(tmp_path) state.mark_title_enriched("tatort", "tmdb") assert state.enriched_titles["tatort"] == "tmdb" def test_mark_speichert_automatisch(self, tmp_path: Path) -> None: """mark_title_enriched ruft save() auf.""" state = SyncState(tmp_path) state.mark_title_enriched("tatort", "omdb") assert (tmp_path / SyncState.FILE_NAME).exists() def test_mark_id_nicht_in_queue_kein_fehler(self, tmp_path: Path) -> None: """Markieren einer ID, die nicht in der Queue ist, verursacht keinen Fehler.""" state = SyncState(tmp_path) state.mark_title_enriched("unbekannt", "tmdb") assert state.enriched_titles["unbekannt"] == "tmdb" # ============================================================================ # SyncState - mark_enrichment_done() # ============================================================================ class TestSyncStateMarkEnrichmentDone: """Tests fuer das Abschliessen einer Enrichment-Runde.""" def test_setzt_zeitstempel(self, tmp_path: Path) -> None: """mark_enrichment_done setzt last_enrichment auf aktuelle Zeit.""" state = SyncState(tmp_path) state.mark_enrichment_done() assert state.last_enrichment != "" # Muss ein gueltiger ISO-Zeitstempel sein datetime.fromisoformat(state.last_enrichment) def test_speichert_automatisch(self, tmp_path: Path) -> None: """mark_enrichment_done ruft save() auf.""" state = SyncState(tmp_path) state.mark_enrichment_done() assert (tmp_path / SyncState.FILE_NAME).exists() def test_queue_bleibt_unveraendert(self, tmp_path: Path) -> None: """mark_enrichment_done veraendert die Queue nicht direkt.""" state = SyncState(tmp_path) state.enrichment_queue = ["tatort"] state.mark_enrichment_done() # mark_enrichment_done leert die Queue NICHT (nur Zeitstempel) # Die Queue wird separat ueber mark_title_enriched geleert assert state.last_enrichment != "" # ============================================================================ # SyncState - Verzeichnis-Erstellung # ============================================================================ class TestSyncStateVerzeichnis: """Tests fuer die automatische Verzeichnis-Erstellung.""" def test_erstellt_cache_verzeichnis(self, tmp_path: Path) -> None: """Das Cache-Verzeichnis wird automatisch erstellt.""" subdir = tmp_path / "tief" / "verschachtelt" state = SyncState(subdir) assert subdir.exists()