128 lines
5.6 KiB
Python
128 lines
5.6 KiB
Python
"""Struktur-Ebene: Topo-Sortierung, Zyklenbruch, Band-Schnitt — mit synthetischen Atomen."""
|
|
|
|
import db
|
|
import llm
|
|
import struktur
|
|
from conftest import run_anlegen, topic_anlegen
|
|
|
|
|
|
def _atom(topic, titel, ziel_id=None, soll_id=1):
|
|
return db.insert("atome", topic=topic, titel=titel, typ="begriff",
|
|
definition=f"Definition {titel}", status="neu",
|
|
soll_id=soll_id, ziel_id=ziel_id, braucht=db.j([]))
|
|
|
|
|
|
def test_topo_ordnung():
|
|
rang = {1: (0, 1), 2: (0, 2), 3: (0, 3)}
|
|
# 3 braucht 1, 2 braucht 3 → 1, 3, 2
|
|
assert struktur._topo([1, 2, 3], [(3, 1), (2, 3)], rang) == [1, 3, 2]
|
|
|
|
|
|
def test_topo_fremde_kanten_bleiben_draussen():
|
|
# Kante zu Atom 3 (nicht in der Gruppe) darf weder 3 hineinziehen
|
|
# noch Gruppenmitglieder verdrängen (aak-Bug: 22 Atome ohne Baustein)
|
|
rang = {1: (0, 1), 2: (0, 2), 3: (0, 3)}
|
|
out = struktur._topo([1, 2], [(1, 3), (2, 1)], rang)
|
|
assert out == [1, 2]
|
|
|
|
|
|
def test_zyklus_finden():
|
|
assert struktur._finde_zyklus([1, 2], [(1, 2), (2, 1)]) is not None
|
|
assert struktur._finde_zyklus([1, 2, 3], [(2, 1), (3, 2)]) is None
|
|
|
|
|
|
async def test_zyklen_brechen_und_schneiden():
|
|
topic = topic_anlegen()
|
|
run = run_anlegen(topic)
|
|
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
|
ziel = db.insert("lernziele", topic=topic, text="Kann P", soll_id=soll, status="aktiv")
|
|
a = _atom(topic, "A", ziel, soll)
|
|
b = _atom(topic, "B", ziel, soll)
|
|
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art) VALUES(?,?,?,'braucht')",
|
|
(topic, a, b))
|
|
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art) VALUES(?,?,?,'braucht')",
|
|
(topic, b, a))
|
|
ctx = llm.Kontext(run, topic, "minimax")
|
|
ctx.ebene = "struktur"
|
|
await struktur._zyklen_brechen(ctx)
|
|
aktiv = db.query("SELECT * FROM kanten WHERE topic=? AND art='braucht' AND status='aktiv'",
|
|
(topic,))
|
|
assert len(aktiv) == 1 # genau eine Kante gebrochen
|
|
|
|
struktur._bausteine_schneiden(ctx)
|
|
atome = struktur._atome(topic)
|
|
assert all(x["baustein_id"] for x in atome)
|
|
|
|
|
|
async def test_band_split():
|
|
topic = topic_anlegen("band")
|
|
run = run_anlegen(topic)
|
|
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
|
ziel = db.insert("lernziele", topic=topic, text="Kann P", soll_id=soll, status="aktiv")
|
|
for i in range(18): # 18 Atome in EINEM Ziel → muss in ≤8er-Teile splitten
|
|
_atom(topic, f"A{i}", ziel, soll)
|
|
ctx = llm.Kontext(run, topic, "minimax")
|
|
struktur._bausteine_schneiden(ctx)
|
|
bausteine = db.query("SELECT * FROM bausteine WHERE topic=?", (topic,))
|
|
assert len(bausteine) >= 3
|
|
for bs in bausteine:
|
|
n = db.one("SELECT COUNT(*) n FROM atome WHERE baustein_id=?", (bs["id"],))["n"]
|
|
assert 4 <= n <= 8
|
|
befunde = struktur.messen(ctx)
|
|
assert not [x for x in befunde if x["art"] in ("band", "partition", "zyklus")]
|
|
|
|
|
|
async def test_merge_ohne_soll_schranke_und_kapitel():
|
|
# Entkopplung: kleines Ziel merged über Soll-Punkt-Grenzen (Partner per Kante);
|
|
# Kapitel entstehen danach als kontiguierliche Segmente
|
|
topic = topic_anlegen("kapitel")
|
|
run = run_anlegen(topic)
|
|
s1 = db.insert("soll", topic=topic, punkt="P1", status="bestaetigt", belege=db.j([]))
|
|
s2 = db.insert("soll", topic=topic, punkt="P2", status="bestaetigt", belege=db.j([]))
|
|
z1 = db.insert("lernziele", topic=topic, text="Kann P1", soll_id=s1, status="aktiv")
|
|
z2 = db.insert("lernziele", topic=topic, text="Kann P2", soll_id=s2, status="aktiv")
|
|
a = _atom(topic, "A", z1, s1) # 1-Atom-Ziel, anderer Soll-Punkt als z2
|
|
b0 = _atom(topic, "B0", z2, s2)
|
|
for i in (1, 2, 3):
|
|
_atom(topic, f"B{i}", z2, s2)
|
|
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art) VALUES(?,?,?,'braucht')",
|
|
(topic, a, b0))
|
|
ctx = llm.Kontext(run, topic, "minimax")
|
|
ctx.ebene = "struktur"
|
|
struktur._bausteine_schneiden(ctx)
|
|
atome = struktur._atome(topic)
|
|
assert len({x["baustein_id"] for x in atome}) == 1 # gemerged trotz fremdem Soll
|
|
await struktur._kapitel_bilden(ctx)
|
|
bausteine = db.query("SELECT * FROM bausteine WHERE topic=?", (topic,))
|
|
assert bausteine and all(x["kapitel_id"] for x in bausteine)
|
|
assert struktur.messen(ctx) == []
|
|
|
|
|
|
async def test_kapitel_retry_und_fallback(monkeypatch):
|
|
import fake_agents
|
|
topic = topic_anlegen("kapfall")
|
|
run = run_anlegen(topic)
|
|
ziel = db.insert("lernziele", topic=topic, text="Kann X", status="aktiv")
|
|
b = [db.insert("bausteine", topic=topic, ziel_id=ziel, titel=f"B{i}", ord=i,
|
|
status="neu") for i in range(4)]
|
|
ctx = llm.Kontext(run, topic, "minimax")
|
|
ctx.ebene = "struktur"
|
|
|
|
aufrufe = {"n": 0}
|
|
def judge(prompt):
|
|
aufrufe["n"] += 1
|
|
if aufrufe["n"] == 1:
|
|
return [{"titel": "kaputt", "bis": 999999}] # ungültige Grenze
|
|
return [{"titel": "Gutes Kapitel", "bis": b[-1]}]
|
|
monkeypatch.setitem(fake_agents._HANDLER, "Kapitel-Schnitt", judge)
|
|
await struktur._kapitel_bilden(ctx)
|
|
kaps = db.query("SELECT * FROM kapitel WHERE topic=?", (topic,))
|
|
assert aufrufe["n"] == 2 and len(kaps) == 1 and kaps[0]["art"] == "judge"
|
|
|
|
monkeypatch.setitem(fake_agents._HANDLER, "Kapitel-Schnitt",
|
|
lambda p: [{"titel": "x", "bis": 999999}])
|
|
await struktur._kapitel_bilden(ctx) # beide Versuche ungültig → Fallback
|
|
kaps = db.query("SELECT * FROM kapitel WHERE topic=?", (topic,))
|
|
assert kaps and all(k["art"] == "fallback" for k in kaps)
|
|
assert "kapitel_fallback" in {bf["art"] for bf in struktur.messen(ctx)}
|