"""Testet die Absicherung der Duplikatpruefung und den Abbruch der Suchrunden. Laeuft ohne Netzzugriff und ohne Kosten, Datenbank, Modell und Suche sind durch Testdoubles ersetzt. Aufruf aus dem Projektstamm: venv/bin/python tests/test_qc_und_runden.py """ import asyncio import json import os import sys sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src")) import services.post_refresh_qc as qc # noqa: E402 import agents.eu_researcher as eur # noqa: E402 from agents.claude_client import ClaudeUsage # noqa: E402 ok = 0 fail = 0 def pruefe(name, bedingung, extra=""): global ok, fail if bedingung: ok += 1 print(" OK " + name) else: fail += 1 print(" FEHL " + name + " " + str(extra)) # --------------------------------------------------------------------------- # Testdouble fuer die Datenbank # --------------------------------------------------------------------------- class FakeCursor: def __init__(self, rows): self._rows = rows async def fetchall(self): return self._rows async def fetchone(self): return self._rows[0] if self._rows else None class FakeDB: """Liefert eine feste Faktenliste und merkt sich die Loeschauftraege.""" def __init__(self, fakten): self.fakten = fakten self.geloescht = [] async def execute(self, sql, params=()): if sql.strip().upper().startswith("DELETE"): self.geloescht.extend(params) return FakeCursor([]) return FakeCursor(list(self.fakten)) def fakt(fid, claim, status="established", quellen=3): return {"id": fid, "claim": claim, "status": status, "sources_count": quellen, "evidence": "e", "checked_at": "2026-08-01 10:00:00"} FAHRPLAN = "Die IMK beschloss einen gemeinsamen Bund-Laender-Fahrplan zum Aufbau der zivilen Verteidigungsfaehigkeit bis 2029" FAHRPLAN_ANDERS = "Die IMK beschloss einen Bund-Laender-Fahrplan fuer die zivile Verteidigung bis zum Jahr 2029" BUNDESWEHR = "Die IMK beschloss die institutionelle Einbindung der Bundeswehr in die Innenministerkonferenz" SOZIAL = "Die IMK beschloss Massnahmen zur haerteren Bekaempfung von organisiertem Sozialmissbrauch" MILLIARDEN = "Die IMK forderte zusaetzliche Milliardeninvestitionen des Bundes in den Zivilschutz" ABSCHIEBUNG = "Die Innenminister waren sich einig, dass der Bund bei Abschiebungen mehr leisten muss" GDP = "Die Gewerkschaft der Polizei bewertete die Beschluesse als grundsaetzlich richtig" SYRER = "In der Frage der Rueckfuehrung von Syrern besteht ein Konflikt zwischen SPD und CDU" IMK_DATUM = "Die 225. Innenministerkonferenz fand im Juni 2026 in Hamburg statt" BESCHLUESSE = "Die IMK fasste insgesamt 66 Beschluesse" geplante_cluster = [] async def fake_cluster(facts, incident_title): return geplante_cluster qc._haiku_find_duplicate_clusters = fake_cluster def lauf_qc(fakten, cluster): global geplante_cluster geplante_cluster = cluster db = FakeDB(fakten) entfernt = asyncio.run(qc.check_fact_duplicates(db, 49, "Innenministerkonferenz 2026")) return entfernt, db.geloescht print("\nA) Der Fall aus Lage 49, sieben verschiedene Fakten in einer Gruppe") zehn = [ fakt(469, FAHRPLAN), fakt(468, IMK_DATUM), fakt(470, BUNDESWEHR), fakt(471, SOZIAL), fakt(472, MILLIARDEN), fakt(473, ABSCHIEBUNG), fakt(474, SYRER, "disputed"), fakt(475, BESCHLUESSE), fakt(476, GDP), fakt(477, "Die IMK befasste sich mit der Umsetzung des europaeischen Asylsystems"), ] entfernt, geloescht = lauf_qc(zehn, [[468, 469, 470, 471, 472, 473, 475, 476]]) pruefe("A1 unplausible Gruppe wird komplett verworfen", entfernt == 0, entfernt) pruefe("A2 kein Fakt geloescht", geloescht == [], geloescht) print("\nB) Echte Dubletten werden weiterhin entfernt") entfernt, geloescht = lauf_qc( [fakt(1, FAHRPLAN, quellen=8), fakt(2, FAHRPLAN_ANDERS, quellen=2), fakt(3, BUNDESWEHR), fakt(4, SOZIAL), fakt(5, MILLIARDEN)], [[1, 2]]) pruefe("B1 echtes Duplikat entfernt", entfernt == 1, entfernt) pruefe("B2 der besser belegte Fakt bleibt", geloescht == [2], geloescht) print("\nC) Kleine Gruppe mit unaehnlichen Fakten wird nicht geloescht") entfernt, geloescht = lauf_qc( [fakt(1, FAHRPLAN), fakt(2, SOZIAL), fakt(3, BUNDESWEHR), fakt(4, MILLIARDEN), fakt(5, ABSCHIEBUNG), fakt(6, GDP)], [[1, 2]]) pruefe("C1 unaehnlicher Fakt bleibt erhalten", entfernt == 0, entfernt) print("\nD) Gemischte Gruppe, nur der wirklich aehnliche Fakt faellt weg") entfernt, geloescht = lauf_qc( [fakt(1, FAHRPLAN, quellen=9), fakt(2, FAHRPLAN_ANDERS, quellen=1), fakt(3, SOZIAL), fakt(4, BUNDESWEHR), fakt(5, MILLIARDEN), fakt(6, ABSCHIEBUNG), fakt(7, GDP), fakt(8, SYRER), fakt(9, BESCHLUESSE), fakt(10, IMK_DATUM)], [[1, 2, 3]]) pruefe("D1 genau ein Fakt entfernt", entfernt == 1, entfernt) pruefe("D2 der aehnliche wurde entfernt", geloescht == [2], geloescht) print("\nE) Grenzwerte") pruefe("E1 Anteilsgrenze gesetzt", 0 < qc.DEDUP_MAX_CLUSTER_ANTEIL <= 1, qc.DEDUP_MAX_CLUSTER_ANTEIL) pruefe("E2 Aehnlichkeitsgrenze gesetzt", 0 < qc.DEDUP_MIN_AEHNLICHKEIT < 1, qc.DEDUP_MIN_AEHNLICHKEIT) pruefe("E3 verschiedene Sachverhalte liegen unter der Grenze", qc._aehnlichkeit(FAHRPLAN, SOZIAL) < qc.DEDUP_MIN_AEHNLICHKEIT, round(qc._aehnlichkeit(FAHRPLAN, SOZIAL), 2)) pruefe("E4 echte Dublette liegt ueber der Grenze", qc._aehnlichkeit(FAHRPLAN, FAHRPLAN_ANDERS) >= qc.DEDUP_MIN_AEHNLICHKEIT, round(qc._aehnlichkeit(FAHRPLAN, FAHRPLAN_ANDERS), 2)) pruefe("E5 bei zwei Fakten bleibt die Gruppengrenze nutzbar", max(2, int(2 * qc.DEDUP_MAX_CLUSTER_ANTEIL)) == 2) # --------------------------------------------------------------------------- # Suchrunden # --------------------------------------------------------------------------- print("\nF) Die Suchphase endet, sobald die Treffergrenze erreicht ist") runden = {"n": 0} async def fake_bedrock_research(prompt, model=None, raw_text=False, timeout=None): runden["n"] += 1 # Jede Runde fordert vier Suchen an, das Modell will nie von selbst enden. antwort = {"action": "search", "reason": "weiter", "queries": [{"q": f"anfrage {runden['n']}-{i}", "market": "de-de", "full_content": False} for i in range(1, 5)]} return json.dumps(antwort), ClaudeUsage(input_tokens=100, output_tokens=50, cost_usd=0.05) async def fake_staan_research(q, market="de-de", extra_snippets=True, max_snippets=4, full_content=False, extra_exclude=None, timeout=None): # Jede Suche liefert 20 neue Treffer, nach zwei Runden ist die Grenze voll. basis = q.replace(" ", "") return [{ "title": "Treffer " + basis + str(i), "hostname": "example.org", "url": "https://example.org/" + basis + "/" + str(i), "snippet": "Auszug", "extra_snippets": [], "full_text": "", } for i in range(1, 21)] letzter_prompt = {"text": ""} async def fake_bedrock_final(prompt, model=None, raw_text=False, timeout=None): """Ab dem Abschlussauftrag wird eine Auswahl zurueckgegeben.""" letzter_prompt["text"] = prompt return "[]", ClaudeUsage(input_tokens=100, output_tokens=50, cost_usd=0.05) async def bedrock_weiche(prompt, model=None, raw_text=False, timeout=None): # Der Rundenauftrag traegt die Rundennummer im Kopf, der Abschlussauftrag # nicht. Daran laesst sich beides sicher unterscheiden. if "Es ist Runde " in prompt: return await fake_bedrock_research(prompt, model, raw_text, timeout) return await fake_bedrock_final(prompt, model, raw_text, timeout) eur.call_bedrock = bedrock_weiche eur.staan_search = fake_staan_research runden["n"] = 0 text, usage = asyncio.run(eur.run_eu_research( title="Innenministerkonferenz 2026", description="", incident_type="research", lang_instruction="", existing_context="", preferred_sources_block="", output_language="Deutsch", excluded_sources=[], research_language_iso="de", )) # Jede Suche liefert 20 Treffer, das Kontingent je Runde liegt bei 25. Die # Runde nimmt also zwei Suchen auf und ueberspringt den Rest. Entscheidend: # die Schleife laeuft weiter, damit die spaeteren Runden ihre Aufgabe # (Luecken schliessen, vertiefen) ueberhaupt ausfuehren koennen. Vorher war # nach Runde 2 Schluss, weil die Trefferzahl als harte Grenze wirkte. suchrunden = runden["n"] pruefe("F1 alle Runden laufen, die Vertiefung faellt nicht mehr aus", suchrunden == eur.EU_RESEARCH_MAX_ROUNDS_RESEARCH, suchrunden) pruefe("F2 je Runde greift das Kontingent statt eines Gesamtstopps", eur.EU_RESEARCH_MAX_NEW_PER_ROUND == 25, eur.EU_RESEARCH_MAX_NEW_PER_ROUND) pruefe("F3 Recherche liefert ein Ergebnis", text == "[]", text[:40]) pruefe("F4 Kosten entsprechen den gelaufenen Runden", usage.cost_usd > 0.05 * suchrunden, round(usage.cost_usd, 4)) # --------------------------------------------------------------------------- # G) Sättigung beendet die Suchphase weiterhin # --------------------------------------------------------------------------- print("\nG) Keine neuen Treffer, keine weiteren Runden") async def fake_staan_immer_gleich(q, market="de-de", extra_snippets=True, max_snippets=4, full_content=False, extra_exclude=None, timeout=None): """Liefert immer dieselben fuenf Treffer: ab Runde 2 gibt es nichts Neues.""" return [{ "title": "Immer derselbe " + str(i), "hostname": "example.org", "url": "https://example.org/immer/" + str(i), "snippet": "Auszug", "extra_snippets": [], "full_text": "", } for i in range(1, 6)] eur.staan_search = fake_staan_immer_gleich runden["n"] = 0 asyncio.run(eur.run_eu_research( title="Innenministerkonferenz 2026", description="", incident_type="research", lang_instruction="", existing_context="", preferred_sources_block="", output_language="Deutsch", excluded_sources=[], research_language_iso="de", )) pruefe("G1 Suchphase endet nach der ersten Runde ohne neue Treffer", runden["n"] == 2, runden["n"]) # --------------------------------------------------------------------------- # H) Verdrängung schafft Platz für spätere Runden # --------------------------------------------------------------------------- print("\nH) Platz fuer gezielte Treffer") korb = { "https://a.de/1": {"_runde": 1, "snippet": "", "full_text": ""}, "https://b.de/1": {"_runde": 1, "snippet": "Auszug", "full_text": ""}, "https://c.de/1": {"_runde": 2, "snippet": "", "full_text": ""}, } pruefe("H1 der schwaechste Treffer einer frueheren Runde weicht", eur._platz_schaffen(korb, 3) and "https://a.de/1" not in korb, list(korb)) nur_volltext = {"https://a.de/1": {"_runde": 1, "snippet": "x", "full_text": "langer Text"}} pruefe("H2 Treffer mit Volltext werden nicht verdraengt", not eur._platz_schaffen(nur_volltext, 3) and len(nur_volltext) == 1) gleiche_runde = {"https://a.de/1": {"_runde": 3, "snippet": "", "full_text": ""}} pruefe("H3 Treffer derselben Runde werden nicht verdraengt", not eur._platz_schaffen(gleiche_runde, 3) and len(gleiche_runde) == 1) pruefe("H4 leerer Korb meldet keinen Erfolg", not eur._platz_schaffen({}, 2)) # --------------------------------------------------------------------------- # I) Die dritte Adhoc-Runde hat einen eigenen Auftrag # --------------------------------------------------------------------------- print("\nI) Auftrag der Vertiefungsrunde") dritte = eur._PHASE_GUIDANCE_ADHOC.get(3, "") pruefe("I1 dritte Runde ist als Vertiefung beschrieben", "VERTIEFUNG" in dritte, dritte[:60]) for stichwort, name in ( ("fehlt", "I2 fragt nach der fehlenden Seite der Lage"), ("Institutionen", "I3 nennt Institutionen als typische Luecke"), ("rechtliche", "I4 nennt den rechtlichen Rahmen"), ("nicht noch einmal", "I5 verbietet die Wiederholung des Hauptereignisses"), ): pruefe(name, stichwort in dritte) print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen") sys.exit(1 if fail else 0)