194 lines
9.4 KiB
Python
194 lines
9.4 KiB
Python
"""Anker-Repair: unverankerte Zitate (start=-1) werden erst deterministisch
|
||
neu gematcht — ohne LLM-Call."""
|
||
|
||
import db
|
||
import inventar
|
||
import llm
|
||
from conftest import run_anlegen, topic_anlegen
|
||
|
||
|
||
def test_merge_kollision_und_kette():
|
||
topic = topic_anlegen("mergekante")
|
||
a = db.insert("atome", topic=topic, titel="A", typ="begriff", definition="lang genug",
|
||
status="neu")
|
||
b = db.insert("atome", topic=topic, titel="B", typ="begriff", definition="kurz",
|
||
status="neu")
|
||
c = db.insert("atome", topic=topic, titel="C", typ="begriff", definition="x",
|
||
status="neu")
|
||
# beide haben dieselbe Kante zu C → blindes Umhängen würde UNIQUE verletzen
|
||
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art) VALUES(?,?,?,'verwandt')",
|
||
(topic, a, c))
|
||
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art) VALUES(?,?,?,'verwandt')",
|
||
(topic, b, c))
|
||
# und eine Kante zwischen den Merge-Partnern → würde Selbstkante
|
||
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art) VALUES(?,?,?,'verwandt')",
|
||
(topic, b, a))
|
||
inventar._merge(topic, a, b)
|
||
kanten = db.query("SELECT * FROM kanten WHERE topic=?", (topic,))
|
||
assert len(kanten) == 1 and kanten[0]["von_atom"] == a and kanten[0]["zu_atom"] == c
|
||
# Merge-Kette: (B, C) — B ist schon gemerged, Wurzel A übernimmt
|
||
inventar._merge(topic, b, c)
|
||
assert db.one("SELECT merged_into FROM atome WHERE id=?", (c,))["merged_into"] == a
|
||
|
||
|
||
async def test_anker_rematch_ohne_llm(tmp_path):
|
||
topic = topic_anlegen("rematch")
|
||
run = run_anlegen(topic)
|
||
snap = tmp_path / "q.md"
|
||
snap.write_text("Vorher. Der Satz steht hier drin. Nachher.", encoding="utf-8")
|
||
q = db.insert("quellen", topic=topic, art="datei", titel="q",
|
||
snapshot=str(snap), hash="h", status="atome")
|
||
a = db.insert("atome", topic=topic, titel="T", typ="begriff", definition="d",
|
||
status="ohne_anker")
|
||
# Extraktion fand das Zitat nicht (Whitespace-Differenz), hat es aber gespeichert
|
||
db.insert("anker", atom_id=a, quelle_id=q, start=-1, ende=-1,
|
||
zitat="Der Satz steht hier drin.")
|
||
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "inventar"
|
||
assert await inventar._anker_fixen_batch(ctx, [a]) is True
|
||
row = db.one("SELECT * FROM anker WHERE atom_id=? AND start>=0", (a,))
|
||
assert row is not None
|
||
assert db.one("SELECT status FROM atome WHERE id=?", (a,))["status"] == "neu"
|
||
assert db.query("SELECT id FROM events WHERE run_id=?", (run,)) == [] # kein LLM-Call
|
||
assert inventar.messen(ctx) == [] or all(
|
||
b["art"] != "atom_ohne_anker" for b in inventar.messen(ctx))
|
||
|
||
|
||
async def test_resume_liest_teilextrahierte_quelle_weiter(tmp_path):
|
||
"""Abbruch mitten in der Extraktion (atome_stand leer): der Resume liest die
|
||
Quelle komplett neu, statt sie als fertig zu überspringen."""
|
||
topic = topic_anlegen("resume-ext")
|
||
run = run_anlegen(topic)
|
||
snap = tmp_path / "q.md"
|
||
snap.write_text("Absatz eins über Kompaktheit.\n\nAbsatz zwei über Vollständigkeit.",
|
||
encoding="utf-8")
|
||
# Quelle hat schon EIN verankertes Atom, aber atome_stand ist leer (Abbruch)
|
||
q = db.insert("quellen", topic=topic, art="datei", titel="q",
|
||
snapshot=str(snap), hash="h", status="extrahiert", atome_stand="")
|
||
a = db.insert("atome", topic=topic, titel="Alt", typ="begriff", definition="d",
|
||
status="neu")
|
||
db.insert("anker", atom_id=a, quelle_id=q, start=0, ende=5, zitat="Absatz")
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "inventar"
|
||
await inventar._extrahiere_quelle(ctx, db.one("SELECT * FROM quellen WHERE id=?", (q,)))
|
||
# neu gelesen → Extraktions-Events da, atome_stand jetzt gesetzt
|
||
assert db.query("SELECT id FROM events WHERE run_id=? AND stage='extraktion'", (run,))
|
||
assert db.one("SELECT atome_stand FROM quellen WHERE id=?", (q,))["atome_stand"] == "atome"
|
||
|
||
# Gegenprobe: mit gesetztem Merker wird NICHT neu gelesen
|
||
run2 = run_anlegen(topic)
|
||
ctx2 = llm.Kontext(run2, topic, "minimax")
|
||
ctx2.ebene = "inventar"
|
||
await inventar._extrahiere_quelle(ctx2, db.one("SELECT * FROM quellen WHERE id=?", (q,)))
|
||
assert db.query("SELECT id FROM events WHERE run_id=?", (run2,)) == []
|
||
|
||
|
||
async def test_stichentscheid_braucht_zwei_stimmen(monkeypatch):
|
||
"""Verwerfen (destruktiv) nur bei vollzähligem einstimmigem „fremd"-Panel."""
|
||
topic = topic_anlegen("stich")
|
||
run = run_anlegen(topic)
|
||
db.insert("soll", topic=topic, punkt="P1", status="bestaetigt", belege=db.j([]))
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "inventar"
|
||
|
||
def neu_atom():
|
||
return db.insert("atome", topic=topic, titel="X", typ="begriff",
|
||
definition="d", status="neu")
|
||
|
||
def panel_mit(votes):
|
||
async def _p(ctx, groesse, **kw):
|
||
return list(votes)
|
||
return _p
|
||
|
||
a1 = neu_atom() # 1× fremd → nicht vollzählig → Atom bleibt
|
||
monkeypatch.setattr(llm, "panel", panel_mit([{"soll": "fremd"}]))
|
||
await inventar._soll_stichentscheid(ctx, a1)
|
||
assert db.one("SELECT status FROM atome WHERE id=?", (a1,))["status"] == "neu"
|
||
|
||
a2 = neu_atom() # 2× fremd → verworfen
|
||
monkeypatch.setattr(llm, "panel", panel_mit([{"soll": "fremd"}, {"soll": "fremd"}]))
|
||
await inventar._soll_stichentscheid(ctx, a2)
|
||
assert db.one("SELECT status FROM atome WHERE id=?", (a2,))["status"] == "verworfen"
|
||
|
||
a3 = neu_atom() # 2× ohne soll-Feld → kein fremd-Votum → bleibt
|
||
monkeypatch.setattr(llm, "panel", panel_mit([{}, {}]))
|
||
await inventar._soll_stichentscheid(ctx, a3)
|
||
assert db.one("SELECT status FROM atome WHERE id=?", (a3,))["status"] == "neu"
|
||
|
||
|
||
async def test_anker_batch_ausfall_verwirft_nicht(monkeypatch, tmp_path):
|
||
"""Call-Ausfall (None) ≠ „keine Quellstelle" → Atom bleibt ohne_anker, bewegt=False."""
|
||
import fake_agents
|
||
topic = topic_anlegen("ankerfail")
|
||
run = run_anlegen(topic)
|
||
snap = tmp_path / "q.md"
|
||
snap.write_text("Ganz anderer Text ohne die gesuchte Stelle.", encoding="utf-8")
|
||
q = db.insert("quellen", topic=topic, art="datei", titel="q",
|
||
snapshot=str(snap), hash="h", status="atome")
|
||
a = db.insert("atome", topic=topic, titel="T", typ="begriff", definition="d",
|
||
status="ohne_anker")
|
||
db.insert("anker", atom_id=a, quelle_id=q, start=-1, ende=-1,
|
||
zitat="Ein Zitat das nirgends im Quelltext steht und lang genug ist.")
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Atom-Anker-Fix-Batch", lambda p: {})
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "inventar"
|
||
bewegt = await inventar._anker_fixen_batch(ctx, [a])
|
||
assert db.one("SELECT status FROM atome WHERE id=?", (a,))["status"] == "ohne_anker"
|
||
assert bewegt is False
|
||
|
||
|
||
def test_finde_zitat_casefold_sz_offsets():
|
||
"""casefold-Stufe: ß→ss-Expansionen vor dem Treffer dürfen den Span nicht
|
||
verschieben; Treffer am Textende darf nicht in einen IndexError laufen."""
|
||
import textkit
|
||
text = "Straße und Fußweg. " * 30 + "DER KERNSATZ STEHT AM ENDE."
|
||
span = textkit.finde_zitat(text, "der kernsatz steht am ende.")
|
||
assert span is not None
|
||
assert text[span[0]:span[1]] == "DER KERNSATZ STEHT AM ENDE."
|
||
|
||
|
||
def test_titel_kern_faltet_schreibvarianten():
|
||
import textkit
|
||
assert textkit.titel_kern("ΔTSP2 Tour‑Länge") == textkit.titel_kern("ΔTSP2 Tour-Länge")
|
||
assert textkit.titel_kern("FeedbackVertexSet ∈ NP") == textkit.titel_kern("Feedback Vertex Set ∈ NP")
|
||
assert textkit.titel_kern("!!!") == ""
|
||
|
||
|
||
def test_fallback_nutzt_jaccard_schwelle(monkeypatch):
|
||
import embedding
|
||
a = "Polynomialzeitreduktionen sind transitiv wenn L1 auf L2 und L2 auf L3 reduzierbar sind"
|
||
b = ("Polynomialzeitreduktionen sind transitiv wenn eine Sprache L1 auf L2"
|
||
" und L2 auf L3 in Polynomialzeit reduzierbar ist")
|
||
# Kosinus-Schwelle allein: Paraphrase rutscht durch (aak Lauf 8)
|
||
assert embedding.kandidaten_paare([a, b], 0.75) == []
|
||
paare = embedding.kandidaten_paare([a, b], 0.75, 0.3)
|
||
assert [(i, j) for i, j, _ in paare] == [(0, 1)]
|
||
|
||
|
||
async def test_titel_dublette_wird_gemerged_trotz_ferner_definition():
|
||
topic = topic_anlegen("titeldup")
|
||
run = run_anlegen(topic)
|
||
# identischer Titel, Definitionen lexikalisch fern (Jaccard < 0.3):
|
||
# nur der Titel-Kern-Generator bringt das Paar ans Panel
|
||
a = db.insert("atome", topic=topic, titel="VERTEX COVER Problem", typ="begriff",
|
||
definition="Entscheidungsproblem, ob ein Graph ein Vertex Cover"
|
||
" der Größe höchstens k enthält.",
|
||
status="neu")
|
||
b = db.insert("atome", topic=topic, titel="VERTEX COVER Problem", typ="begriff",
|
||
definition="Gefragt wird nach einer Knotenmenge, die jede Kante"
|
||
" abdeckt und maximal k Elemente hat.",
|
||
status="neu")
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "inventar"
|
||
await inventar._judge_dedup(ctx)
|
||
stati = {r["id"]: r["status"] for r in db.query(
|
||
"SELECT id, status FROM atome WHERE topic=?", (topic,))}
|
||
assert sorted(stati.values()) == ["gemerged", "neu"]
|
||
|
||
|
||
def test_titel_kaputt_flaggt_latex():
|
||
assert inventar._titel_kaputt("Variablen in $\\alpha_k$")
|
||
assert inventar._titel_kaputt("Der \\textsc{VC}-Beweis")
|
||
assert not inventar._titel_kaputt("Vertex Cover und Entscheidungsvariante")
|