AegisSight-Monitor/tests/test_eu_belegsuche.py

"""Testet die gezielte Belegsuche der EU-Fassung.

Laeuft ohne Netzzugriff und ohne Kosten, das planende Modell und die
europaeische Suche sind durch Testdoubles ersetzt.

Aufruf aus dem Projektstamm:
    venv/bin/python tests/test_eu_belegsuche.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 agents.eu_researcher as eur  # noqa: E402
from agents.claude_client import ClaudeUsage  # noqa: E402
from config import (  # noqa: E402
    EU_VERIFY_FULLTEXT_CHARS, EU_VERIFY_FULLTEXT_QUERIES, EU_VERIFY_HITS_PER_QUERY,
    EU_VERIFY_MAX_ITEMS, STAAN_COST_PER_QUERY_USD,
)

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))


zustand = {"plan_prompt": "", "suchen": []}
plan_antwort = {"queries": []}


async def fake_bedrock(prompt, model=None, raw_text=False, timeout=None):
    zustand["plan_prompt"] = prompt
    return json.dumps(plan_antwort), ClaudeUsage(input_tokens=10, output_tokens=5, cost_usd=0.002)


treffer_je_anfrage = {}


async def fake_staan(q, market="de-de", extra_snippets=True, max_snippets=3,
                     full_content=False, extra_exclude=None, timeout=None):
    zustand["suchen"].append({"q": q, "market": market, "full_content": full_content})
    return treffer_je_anfrage.get(q, [])


def treffer(n, praefix="t", host="example.org"):
    return [{
        "title": praefix + " Titel " + str(i),
        "hostname": host,
        "url": "https://" + host + "/" + praefix + str(i),
        "snippet": "Auszug " + str(i),
        "extra_snippets": ["Zusatz " + str(i)],
    } for i in range(1, n + 1)]


eur.call_bedrock = fake_bedrock
eur.staan_search = fake_staan


def neu():
    zustand["suchen"].clear()
    treffer_je_anfrage.clear()


print("\nA) Planung ist punktbezogen und fuehrt die Anfragen aus")
neu()
plan_antwort["queries"] = [
    {"q": "Ceuta Opferzahl offiziell", "market": "de-de", "for_items": [1]},
    {"q": "Ceuta border deaths toll", "market": "en-us", "for_items": [1, 2]},
    {"q": "Sanchez Ceuta declaracion", "market": "es-es", "for_items": [2]},
]
treffer_je_anfrage["Ceuta Opferzahl offiziell"] = treffer(2, "a")
treffer_je_anfrage["Ceuta border deaths toll"] = treffer(2, "b")
treffer_je_anfrage["Sanchez Ceuta declaracion"] = treffer(2, "c")
_r = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["57 Tote gemeldet", "Sanchez sprach von Angriff"],
    output_language="Deutsch", max_queries=7,
))
block, usage, ausgefuehrt = _r.block, _r.usage, _r.queries
pruefe("A1 alle geplanten Anfragen ausgefuehrt", len(zustand["suchen"]) == 3, zustand["suchen"])
pruefe("A2 ausgefuehrte Anfragen zurueckgemeldet", len(ausgefuehrt) == 3, ausgefuehrt)
pruefe("A3 jeder Punkt steht im Planungsauftrag",
       "57 Tote gemeldet" in zustand["plan_prompt"] and "Sanchez sprach von Angriff" in zustand["plan_prompt"])
pruefe("A4 Planungsauftrag verlangt Punktzuordnung", "for_items" in zustand["plan_prompt"])
pruefe("A5 Obergrenze steht im Auftrag", "7" in zustand["plan_prompt"])
pruefe("A6 alle Treffer nummeriert im Block", block.count("[S") == 6, block.count("[S"))
pruefe("A7 Auszuege uebernommen", block.count("Auszug") >= 6, block.count("Auszug"))
pruefe("A8 unbekannter Markt faellt auf die Vorgabe zurueck",
       [s["market"] for s in zustand["suchen"]] == ["de-de", "en-us", "de-de"],
       [s["market"] for s in zustand["suchen"]])
erwartet = round(0.002 + 3 * STAAN_COST_PER_QUERY_USD, 6)
pruefe("A9 Suchkosten eingerechnet", round(usage.cost_usd, 6) == erwartet, usage.cost_usd)

print("\nB) Dieselbe Quelle steht nur einmal im Belegblock")
neu()
plan_antwort["queries"] = [
    {"q": "anfrage eins", "market": "de-de"},
    {"q": "anfrage zwei", "market": "de-de"},
]
treffer_je_anfrage["anfrage eins"] = treffer(3, "x")
treffer_je_anfrage["anfrage zwei"] = treffer(3, "x")  # identische URLs
_rb = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt"], max_queries=7,
))
block_b = _rb.block
pruefe("B1 Dubletten entfernt", block_b.count("[S") == 3, block_b.count("[S"))
pruefe("B2 Nummerierung bleibt luecklos",
       all(("[S" + str(i) + "]") in block_b for i in (1, 2, 3)))

print("\nC) Nachrunde meidet bereits gestellte Anfragen")
neu()
plan_antwort["queries"] = [
    {"q": "alte anfrage", "market": "de-de"},
    {"q": "neue anfrage", "market": "de-de"},
]
treffer_je_anfrage["neue anfrage"] = treffer(2, "n")
_rc = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt"], max_queries=5,
    avoid_queries=["alte anfrage"], label="Nachhak-Suche",
))
block_c, ausgefuehrt_c = _rc.block, _rc.queries
pruefe("C1 bereits gestellte Anfrage wird verworfen",
       [s["q"] for s in zustand["suchen"]] == ["neue anfrage"], zustand["suchen"])
pruefe("C2 Sperrliste steht im Planungsauftrag", "alte anfrage" in zustand["plan_prompt"])
pruefe("C3 nur die neue Anfrage gemeldet", ausgefuehrt_c == ["neue anfrage"], ausgefuehrt_c)

print("\nD) Grenzen werden eingehalten")
neu()
plan_antwort["queries"] = [{"q": "q" + str(i), "market": "de-de"} for i in range(1, 11)]
for i in range(1, 11):
    treffer_je_anfrage["q" + str(i)] = treffer(20, "y" + str(i))
_rd = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt"], max_queries=4,
))
block_d, ausgefuehrt_d = _rd.block, _rd.queries
pruefe("D1 hoechstens max_queries Anfragen", len(ausgefuehrt_d) == 4, ausgefuehrt_d)
pruefe("D2 Treffer je Anfrage begrenzt",
       block_d.count("[S") == 4 * EU_VERIFY_HITS_PER_QUERY, block_d.count("[S"))
neu()
plan_antwort["queries"] = [{"q": "q1", "market": "de-de"}]
treffer_je_anfrage["q1"] = treffer(1, "z")
asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt " + str(i) for i in range(1, 40)], max_queries=2,
))
gezaehlt = sum(1 for i in range(1, 40) if ("Punkt " + str(i)) in zustand["plan_prompt"])
pruefe("D3 Zahl der Pruefpunkte begrenzt", gezaehlt == EU_VERIFY_MAX_ITEMS, gezaehlt)

print("\nE) Ausfaelle fuehren nicht zum Abbruch")
neu()
plan_antwort["queries"] = [{"q": "kaputt", "market": "de-de"}, {"q": "heil", "market": "de-de"}]
treffer_je_anfrage["heil"] = treffer(2, "h")


async def staan_mit_fehler(q, **kw):
    zustand["suchen"].append({"q": q, "market": kw.get("market")})
    if q == "kaputt":
        raise eur.StaanError("Suche nicht erreichbar")
    return treffer_je_anfrage.get(q, [])


eur.staan_search = staan_mit_fehler
_re = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt"], max_queries=5,
))
eur.staan_search = fake_staan
block_e, usage_e, ausgefuehrt_e = _re.block, _re.usage, _re.queries
pruefe("E1 gescheiterte Suche uebersprungen", ausgefuehrt_e == ["heil"], ausgefuehrt_e)
pruefe("E2 gefundene Belege bleiben erhalten", block_e.count("[S") == 2, block_e.count("[S"))
pruefe("E3 nur bezahlte Suchen berechnet",
       round(usage_e.cost_usd, 6) == round(0.002 + STAAN_COST_PER_QUERY_USD, 6), usage_e.cost_usd)

neu()
plan_antwort["queries"] = []
_rf = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt"], max_queries=5,
))
block_f, ausgefuehrt_f = _rf.block, _rf.queries
pruefe("E4 leere Planung liefert leeren Block", block_f == "" and ausgefuehrt_f == [])

_rg = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=[], max_queries=5,
))
block_g = _rg.block
pruefe("E5 ohne Pruefpunkte keine Suche", block_g == "")

print("\nF) Gefundene Quellen werden einzeln zurueckgegeben")
neu()
plan_antwort["queries"] = [{"q": "eine anfrage", "market": "de-de"}]
treffer_je_anfrage["eine anfrage"] = treffer(3, "q", host="tagesschau.de")
_rh = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt"], max_queries=5,
))
pruefe("F1 jede Quelle einzeln gemeldet", len(_rh.quellen) == 3, len(_rh.quellen))
pruefe("F2 Quelle traegt Titel, Herkunft und Adresse",
       all(q.get("headline") and q.get("source") and q.get("source_url") for q in _rh.quellen),
       _rh.quellen[:1])
pruefe("F3 Adressen stimmen mit dem Block ueberein",
       all(q["source_url"] in _rh.block for q in _rh.quellen))
neu()
plan_antwort["queries"] = [{"q": "a", "market": "de-de"}, {"q": "b", "market": "de-de"}]
treffer_je_anfrage["a"] = treffer(2, "dup")
treffer_je_anfrage["b"] = treffer(2, "dup")
_ri = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt"], max_queries=5,
))
pruefe("F4 doppelte Quellen nur einmal gemeldet", len(_ri.quellen) == 2, len(_ri.quellen))

print("\nG) Der Auftrag verlangt vollstaendige Quellenadressen")
from agents.eu_researcher import build_eu_prompt  # noqa: E402

mit = build_eu_prompt("Auftrag", "\n\nBELEGE\n[S1] Titel | host | https://x.y/z")
ohne = build_eu_prompt("Auftrag", "")
pruefe("G1 Hinweis auf die Adresspflicht vorhanden",
       "vollständige" in mit and "URL" in mit)
pruefe("G2 ohne Belege kein Adresshinweis", "Adresse (die URL" not in ohne)

print("\nH) Ein Teil der Anfragen ist eine Gegenprobe")
neu()
plan_antwort["queries"] = [
    {"q": "ceuta opferzahl offiziell", "market": "de-de", "for_items": [1]},
    {"q": "ceuta opferzahl dementiert kritik", "market": "de-de", "for_items": [1],
     "gegenprobe": True},
]
treffer_je_anfrage["ceuta opferzahl offiziell"] = treffer(2, "a")
treffer_je_anfrage["ceuta opferzahl dementiert kritik"] = treffer(2, "b")
_rj = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["57 Tote gemeldet"], max_queries=7,
))
def markierte_treffer(block):
    """Zaehlt nur Trefferzeilen, nicht die Erklaerung im Kopf."""
    return sum(1 for z in block.splitlines()
               if z.startswith("[S") and "[GEGENPROBE]" in z)


pruefe("H1 Auftrag verlangt Gegenproben", "GEGENPROBEN" in zustand["plan_prompt"])
pruefe("H2 Gegenprobe-Treffer sind gekennzeichnet",
       markierte_treffer(_rj.block) == 2, markierte_treffer(_rj.block))
pruefe("H3 Kopf erklaert die Gegenprobe", "GEGENPROBE" in _rj.block.split("[S1]")[0])
pruefe("H4 stuetzende Treffer bleiben unmarkiert",
       _rj.block.count("\n[S") == 4, _rj.block.count("\n[S"))

print("\nI) Belastungsprobe sucht ausschliesslich Gegenproben")
neu()
plan_antwort["queries"] = [
    {"q": "eine anfrage", "market": "de-de"},
    {"q": "andere anfrage", "market": "de-de"},
]
treffer_je_anfrage["eine anfrage"] = treffer(1, "c")
treffer_je_anfrage["andere anfrage"] = treffer(1, "d")
_rk = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt"], max_queries=4,
    label="Belastungsprobe", nur_gegenproben=True,
))
pruefe("I1 alle Treffer als Gegenprobe gekennzeichnet",
       markierte_treffer(_rk.block) == 2, markierte_treffer(_rk.block))
pruefe("I2 Auftrag fordert so viele Gegenproben wie Anfragen",
       "Davon sind 4 Anfragen" in zustand["plan_prompt"])

print("\nJ) Artikeltexte werden geholt und begrenzt")
neu()
plan_antwort["queries"] = [
    {"q": "mit text", "market": "de-de", "full_content": True},
    {"q": "ohne text", "market": "de-de", "full_content": False},
]
treffer_je_anfrage["mit text"] = [{
    "title": "Titel mit Text", "hostname": "example.org",
    "url": "https://example.org/volltext", "snippet": "Auszug",
    "extra_snippets": [], "full_text": "W" * 5000,
}]
treffer_je_anfrage["ohne text"] = treffer(1, "e")
_rl = asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt"], max_queries=5,
))
pruefe("J1 Volltext-Anfrage wurde mit Volltext gestellt",
       zustand["suchen"][0].get("full_content") is True, zustand["suchen"][0])
pruefe("J2 andere Anfrage ohne Volltext",
       zustand["suchen"][1].get("full_content") is False, zustand["suchen"][1])
pruefe("J3 Artikeltext steht im Belegblock", "Artikeltext:" in _rl.block)
pruefe("J4 Artikeltext ist begrenzt",
       _rl.block.count("W") <= EU_VERIFY_FULLTEXT_CHARS + 10, _rl.block.count("W"))

neu()
plan_antwort["queries"] = [
    {"q": f"anfrage {i}", "market": "de-de", "full_content": True} for i in range(1, 7)
]
for i in range(1, 7):
    treffer_je_anfrage[f"anfrage {i}"] = [{
        "title": "T", "hostname": "example.org", "url": f"https://example.org/{i}",
        "snippet": "s", "extra_snippets": [], "full_text": "Text",
    }]
asyncio.run(eur.plan_verification_searches(
    title="Ceuta", search_items=["Punkt"], max_queries=6,
))
mit_volltext = sum(1 for s in zustand["suchen"] if s.get("full_content"))
pruefe("J5 Zahl der Volltext-Anfragen ist gedeckelt",
       mit_volltext == EU_VERIFY_FULLTEXT_QUERIES, mit_volltext)

print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen")
sys.exit(1 if fail else 0)