init

tests/test_korpus.py

@@ -0,0 +1,105 @@
+"""Ebene 0, Konsens: Vollständigkeits-Invariante (kein Kandidat verschwindet
+still), explizite Ablehnung, uni-Beleg-Regel (1 Beleg reicht)."""
+
+import re
+
+import db
+import fake_agents
+import korpus
+import llm
+from conftest import run_anlegen, topic_anlegen
+
+
+def _seed(art="uni", n_quellen=1):
+    topic = topic_anlegen("k-test", art=art)
+    quellen = [db.insert("quellen", topic=topic, art="datei", titel=f"q{i}.txt",
+                         snapshot=f"q{i}.txt", hash=f"h{i}",
+                         status="extrahiert", rolle="stoff")
+               for i in range(n_quellen)]
+    ctx = llm.Kontext(run_anlegen(topic), topic, "minimax")
+    ctx.ebene = "korpus"
+    return topic, quellen, ctx
+
+
+def _kandidat(topic, punkt, quelle):
+    return db.insert("soll", topic=topic, punkt=punkt, status="kandidat",
+                     belege=db.j([{"quelle": quelle, "zitat": f"Zitat {punkt}"}]))
+
+
+def _ids(prompt):
+    return [int(i) for i in re.findall(r"^(\d+): ", prompt.split("KANDIDATEN:")[1],
+                                       re.MULTILINE)]
+
+
+async def test_nachrunde_ordnet_vergessene_zu(monkeypatch):
+    topic, (q,), ctx = _seed()
+    a = _kandidat(topic, "Alpha", q)
+    b = _kandidat(topic, "Beta", q)
+    c = _kandidat(topic, "Gamma", q)
+
+    def judge(prompt):
+        if "NACHRUNDE" in prompt:  # nur die fehlende id wird angeboten
+            assert _ids(prompt) == [c] and "Alpha" in prompt
+            return [{"punkt": "Alpha", "kandidaten": [c]}]  # wörtlich → Merge
+        return [{"punkt": "Alpha", "kandidaten": [a]},
+                {"punkt": "Beta", "kandidaten": [b]}]  # c stillschweigend vergessen
+
+    monkeypatch.setitem(fake_agents._HANDLER, "Korpus-Soll-Konsens", judge)
+    assert await korpus._konsens(ctx) == 2  # uni: 1 Quelle reicht als Beleg
+    alpha = db.one("SELECT * FROM soll WHERE topic=? AND status='bestaetigt'"
+                   " AND punkt='Alpha'", (topic,))
+    assert len(db.uj(alpha["belege"])) == 2  # Gamma-Beleg in Alpha aufgegangen
+    for i in (a, b, c):
+        assert db.one("SELECT status FROM soll WHERE id=?", (i,))["status"] == "gefaltet"
+    assert korpus.messen(ctx) == []
+
+
+async def test_ablehnung_ist_explizit(monkeypatch):
+    topic, (q,), ctx = _seed()
+    a = _kandidat(topic, "Alpha", q)
+    m = _kandidat(topic, "Einführung ins Thema", q)
+
+    def judge(prompt):
+        return [{"punkt": "Alpha", "kandidaten": [a]},
+                {"abgelehnt": True, "grund": "Meta-Überschrift", "kandidaten": [m]}]
+
+    monkeypatch.setitem(fake_agents._HANDLER, "Korpus-Soll-Konsens", judge)
+    assert await korpus._konsens(ctx) == 1
+    assert db.one("SELECT status FROM soll WHERE id=?", (m,))["status"] == "abgelehnt"
+    assert korpus.messen(ctx) == []  # abgelehnt ist erledigt, kein Befund
+
+
+async def test_hartnaeckig_vergessene_wird_befund(monkeypatch):
+    topic, (q,), ctx = _seed()
+    a = _kandidat(topic, "Alpha", q)
+    c = _kandidat(topic, "Gamma", q)
+
+    def judge(prompt):
+        if "NACHRUNDE" in prompt:
+            return []  # Judge verweigert die Zuordnung dauerhaft
+        return [{"punkt": "Alpha", "kandidaten": [a]}]
+
+    monkeypatch.setitem(fake_agents._HANDLER, "Korpus-Soll-Konsens", judge)
+    await korpus._konsens(ctx)
+    assert db.one("SELECT status FROM soll WHERE id=?", (c,))["status"] == "kandidat"
+    befunde = korpus.messen(ctx)
+    assert [b["art"] for b in befunde] == ["soll_kandidat_offen"]
+    assert befunde[0]["item"] == str(c)
+
+
+async def test_thema_braucht_zwei_quellen(monkeypatch):
+    topic, (q1, q2), ctx = _seed(art="thema", n_quellen=2)
+    x1 = _kandidat(topic, "X", q1)
+    x2 = _kandidat(topic, "X (Synonym)", q2)
+    y = _kandidat(topic, "Y", q1)
+
+    def judge(prompt):
+        if "NACHRUNDE" in prompt:
+            return []
+        return [{"punkt": "X", "kandidaten": [x1, x2]},
+                {"punkt": "Y", "kandidaten": [y]}]
+
+    monkeypatch.setitem(fake_agents._HANDLER, "Korpus-Soll-Konsens", judge)
+    assert await korpus._konsens(ctx) == 1  # X: 2 Quellen; Y: nur 1 → kein Punkt
+    assert db.one("SELECT status FROM soll WHERE id=?", (y,))["status"] == "kandidat"
+    assert {b["art"] for b in korpus.messen(ctx)} == {"soll_kandidat_offen"}