Keine Neuheiten erkannt.
" ) SEITE_MIT_TREFFERN = """02.03.2026 - kurz.
kurz
") is True def test_struktur_verdaechtig_wenige_treffer_nur_bei_linkreicher_seite(): viele_links = "".join('x' for _ in range(40)) assert struktur_verdaechtig(2, viele_links) is True assert struktur_verdaechtig(2, "x") is False assert struktur_verdaechtig(5, viele_links) is False def test_unsichere_treffer_fehlender_verlag_oder_generischer_titel(): assert treffer_ist_unsicher(QuellenTreffer(titel="X", quellen_url="u")) is True assert treffer_ist_unsicher(TREFFER_UNSICHER()) is True # kein Verlag assert treffer_ist_unsicher(TREFFER_SICHER()) is False assert titel_ist_verdaechtig("Mehr") is True assert titel_ist_verdaechtig("ab") is True assert titel_ist_verdaechtig("Fadenfroh") is False def test_extrahiere_json_toleriert_code_fences(): text = "Antwort:\n```json\n{\"folge_urls\": []}\n```\nEnde." assert extrahiere_ki_json(text) == {"folge_urls": []} assert extrahiere_ki_json("völlig kaputt") is None # ---------------- Konfiguration & Fallback ---------------- def test_konfiguration_standard_aus(): k = lade_ki_konfiguration({}) assert not k.aktiv and not k.vollstaendig assert k.konfidenz_min == 0.7 assert k.timeout >= 1.0 def test_service_ohne_konfiguration_baut_keinen_ki_client(session_factory, monkeypatch): """Fallback aktiv=0: kein KI-Client, exakt der klassische Crawler.""" monkeypatch.setenv("SPIELE_NEUHEITEN_KI_AKTIV", "0") class Streng: def __call__(self): # pragma: no cover — darf nie aufgerufen werden raise AssertionError("KI-Client ohne Aktivierung gebaut") def schliessen(self): pass quelle = StatischeQuelle("spielbox", [TREFFER_SICHER()]) service = WebQuellenSyncService( session_factory, adapter_klassen=(quelle,), client_fabrik=lambda v: _ohne_netz_client(), ki_fabrik=Streng(), ) zeilen = service.laufe() assert zeilen[0].neu == 1 and zeilen[0].fehlermeldung is None with session_factory() as db: assert db.query(Neuheit).count() == 1 def test_service_mit_aktiv_ohne_key_laeuft_klassisch(session_factory, monkeypatch): """Fallback: AKTIV=1, aber Key fehlt → vollständige Konfiguration fehlt.""" monkeypatch.setenv("SPIELE_NEUHEITEN_KI_AKTIV", "1") monkeypatch.setenv("SPIELE_NEUHEITEN_KI_BASIS_URL", "https://llm.example.test/v1") monkeypatch.delenv("SPIELE_NEUHEITEN_KI_API_KEY", raising=False) monkeypatch.setenv("SPIELE_NEUHEITEN_KI_MODELL", "test-modell") class Streng: def __call__(self): # pragma: no cover raise AssertionError("KI-Client ohne Key gebaut") def schliessen(self): pass quelle = StatischeQuelle("spielbox", [TREFFER_SICHER()]) service = WebQuellenSyncService( session_factory, adapter_klassen=(quelle,), client_fabrik=lambda v: _ohne_netz_client(), ki_fabrik=Streng(), ) zeilen = service.laufe() assert zeilen[0].neu == 1 def test_timeout_genau_ein_retry_dann_klassischer_befund(): """Timeout: insgesamt zwei Versuche, danach klassischer Treffer unverändert.""" transport = LlmTransport([httpx.ReadTimeout("zu lang"), httpx.ReadTimeout("nochmal")]) hilfe = ki_hilfe(transport) treffer_liste, eingriffe = hilfe.treffer_nachbessern([TREFFER_UNSICHER()]) assert transport.aufrufe == 2 # erster Versuch + genau ein Retry assert eingriffe == [] original = treffer_liste[0] assert original.verlag is None # klassischer Befund bleibt assert original.quellen_url == "https://www.spielbox.de/x/1" def test_kaputtes_json_kein_retry_kein_eingriff(): transport = LlmTransport(["das ist {kaputtes JSON ohne Sinn"]) hilfe = ki_hilfe(transport) treffer_liste, eingriffe = hilfe.treffer_nachbessern([TREFFER_UNSICHER()]) assert transport.aufrufe == 1 # Parsing-Fehler werden nicht wiederholt assert eingriffe == [] assert treffer_liste[0].verlag is None def test_serverfehler_nach_retry_faellt_zurueck(): transport = LlmTransport([503, 503]) hilfe = ki_hilfe(transport) _, eingriffe = hilfe.treffer_nachbessern([TREFFER_UNSICHER()]) folge_urls = hilfe.folge_urls_bei_bedarf(0, SEITE_LEER_VIELE_LINKS, BASIS_URL, set()) assert transport.aufrufe == 4 # je Aufrufart zwei Versuche assert eingriffe == [] and folge_urls == [] # ---------------- Konfidenz-Schwelle ---------------- def test_feldkorrektur_unter_konfidenz_min_wird_verworfen(): schwach = dict(GUTE_FELDANTWORT, konfidenz=0.5) transport = LlmTransport([schwach]) hilfe = ki_hilfe(transport, konfidenz_min=0.7) treffer_liste, eingriffe = hilfe.treffer_nachbessern([TREFFER_UNSICHER()]) assert transport.aufrufe == 1 assert eingriffe == [] assert treffer_liste[0].titel == "Pegasus Spiele: Next Station Berlin" assert treffer_liste[0].verlag is None def test_url_vorschlaege_unter_konfidenz_min_werden_verworfen(): vorschlaege = { "folge_urls": ["/neuheiten-neue-spiele?start=6"], "konfidenz": 0.3, } transport = LlmTransport([vorschlaege]) hilfe = ki_hilfe(transport) urls = hilfe.folge_urls_bei_bedarf(0, SEITE_LEER_VIELE_LINKS, BASIS_URL, set()) assert urls == [] # ---------------- Struktur-Erkennung ---------------- def test_struktur_erkennung_nur_gleiche_domain_urls_werden_verfolgt(): """Fremde Domains und Nicht-HTTP-Schemata werden hart herausgefiltert.""" gemischt = { "folge_urls": [ "/neuheiten-neue-spiele?start=6", # gleiche Domain ✓ "https://www.spielbox.de/neuheiten?page=2", # gleiche Domain ✓ "https://boese.example.org/neuheiten", # fremde Domain ✗ "ftp://www.spielbox.de/liste", # Schema ✗ "/neuheiten-neue-spiele?start=6#oben", # Duplikat ✗ "javascript:zeigeListe()", # Schema ✗ ], "konfidenz": 0.9, } crawler_client, transport, _ = _crawler_client({BASIS_URL: SEITE_LEER_VIELE_LINKS}) transport.antworten["https://www.spielbox.de/neuheiten-neue-spiele?start=6"] = _antwort(SEITE_MIT_TREFFERN) transport.antworten["https://www.spielbox.de/neuheiten?page=2"] = _antwort(SEITE_MIT_TREFFERN) adapter = SpielboxQuelle(crawler_client) adapter.ki_hilfe = ki_hilfe(LlmTransport([gemischt])) ergebnis = adapter.sammle() gefolgt = [u for u in transport.anfragen if u != BASIS_URL] assert "https://www.spielbox.de/neuheiten-neue-spiele?start=6" in gefolgt assert all(u.startswith("https://www.spielbox.de") for u in gefolgt) assert not any("boese" in u or u.startswith("ftp") for u in gefolgt) assert len(ergebnis.treffer) == 2 # aus den beiden nachgeschobenen Seiten assert ergebnis.ki_folge_urls == gefolgt def test_struktur_erkennung_maximal_20_folge_urls(): """Rate-Limit-Schutz: auch bei 30 Vorschlägen werden höchstens 20 verfolgt.""" vorschlaege = { "folge_urls": [f"/liste?seite={i}" for i in range(30)], "konfidenz": 0.9, } crawler_client, transport, _ = _crawler_client({BASIS_URL: SEITE_LEER_VIELE_LINKS}) for i in range(30): transport.antworten[f"https://www.spielbox.de/liste?seite={i}"] = _antwort(SEITE_MIT_TREFFERN) adapter = SpielboxQuelle(crawler_client) adapter.ki_hilfe = ki_hilfe(LlmTransport([vorschlaege])) adapter.sammle() gefolgt = [u for u in transport.anfragen if u != BASIS_URL] assert len(gefolgt) <= MAX_FOLGE_URLS assert len(gefolgt) == 20 assert "https://www.spielbox.de/liste?seite=25" not in gefolgt def test_struktur_erkennung_greift_nicht_bei_normalen_ergebnissen(): """Drei sichere Treffer → keine Verdachts-Heuristik, kein LLM-Aufruf.""" seite = SEITE_MIT_TREFFERN * 3 crawler_client, transport, _ = _crawler_client({BASIS_URL: seite}) adapter = SpielboxQuelle(crawler_client) adapter.ki_hilfe = ki_hilfe(LlmTransport([])) # würde bei Aufruf zählen ergebnis = adapter.sammle() assert len(ergebnis.treffer) == 3 assert len(transport.anfragen) == 1 # nur die Startseite # ---------------- Feld-Korrektur ---------------- def test_feldkorrektur_nur_bei_unsicheren_treffern_url_bleibt_fix(): """Sichere Treffer kosten kein Token; korrigiert wird nie die URL.""" transport = LlmTransport([GUTE_FELDANTWORT]) hilfe = ki_hilfe(transport) sicher = TREFFER_SICHER() unsicher = TREFFER_UNSICHER() treffer_liste, eingriffe = hilfe.treffer_nachbessern([sicher, unsicher]) assert transport.aufrufe == 1 # nur der unsichere Treffer ging ans LLM assert treffer_liste[0] is sicher # unangetastet korrigiert = treffer_liste[1] assert korrigiert.titel == "Next Station Berlin" assert korrigiert.verlag == "Pegasus Spiele" assert korrigiert.quellen_url == unsicher.quellen_url # URL bleibt fix assert len(eingriffe) == 1 protokoll = eingriffe[0] assert protokoll["modell"] == "test-modell" assert protokoll["konfidenz"] == 0.9 assert protokoll["korrigiert"] == { "titel": "Next Station Berlin", "verlag": "Pegasus Spiele", } assert protokoll["vorher"]["titel"] == "Pegasus Spiele: Next Station Berlin" def test_feldkorrektur_ohne_aenderung_erzeugt_keinen_eingriff(): identisch = { "titel": "Pegasus Spiele: Next Station Berlin", "verlag": None, "autor": None, "konfidenz": 0.9, "begruendung": "passt", } transport = LlmTransport([identisch]) hilfe = ki_hilfe(transport) treffer_liste, eingriffe = hilfe.treffer_nachbessern([TREFFER_UNSICHER()]) assert eingriffe == [] assert treffer_liste[0].quellen_url == "https://www.spielbox.de/x/1" # ---------------- Sync-Integration + Audit ---------------- class StatischeQuelle: """Adapter-Ersatz mit fester Trefferliste (Service-Tests ohne Netz).""" def __init__(self, name: str = "spielbox", treffer: list[QuellenTreffer] | None = None): self.name = name self.anzeigename = name.title() self.start_urls = (f"https://{name}.example.org/liste",) self._treffer = treffer if treffer is not None else [] def __call__(self, client): return self def sammle(self) -> SammelErgebnis: return SammelErgebnis(treffer=list(self._treffer), seiten=1) def _ohne_netz_client() -> WebQuellenClient: return WebQuellenClient(transport=None, schlaf=lambda s: None, uhr=FakeUhr()) class FakeAudit: def __init__(self): self.eintraege: list[tuple] = [] def log_sync(self, actor, action, objekt_typ, objekt_id, details, **kwargs): self.eintraege.append((actor, action, objekt_typ, objekt_id, details)) def test_sync_auditert_ki_eingriffe_mit_modell_und_konfidenz(): """End-to-End: Crawler-Seite mit unsicherem Treffer → KI-Korrektur → Audit. „Ringträger“ ohne Doppelpunkt: der Parser findet keinen Verlag → Treffer ist unsicher → das LLM darf titel/verlag korrigieren. """ seite = ( '15.08.2026 - Text.