213 lines
10 KiB
Python
213 lines
10 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)}
|
||
|
||
|
||
async def test_level_kalibrierung(monkeypatch):
|
||
import fake_agents
|
||
topic = topic_anlegen("level")
|
||
run = run_anlegen(topic)
|
||
ziel = db.insert("lernziele", topic=topic, text="Kann X", status="aktiv")
|
||
a1 = db.insert("atome", topic=topic, titel="Kern", typ="begriff", definition="d",
|
||
level="M", status="neu", ziel_id=ziel, braucht=db.j([]))
|
||
a2 = db.insert("atome", topic=topic, titel="Detail", typ="aussage", definition="d",
|
||
level="M", status="neu", ziel_id=ziel, braucht=db.j([]))
|
||
|
||
def judge(prompt):
|
||
return [{"atom": a1, "level": "E"}, {"atom": a2, "level": "S"},
|
||
{"atom": 99999, "level": "E"}, {"atom": a2, "level": "X"}] # Müll ignorieren
|
||
monkeypatch.setitem(fake_agents._HANDLER, "Level-Kalibrierung", judge)
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
await struktur._level_kalibrieren(ctx)
|
||
assert db.one("SELECT level FROM atome WHERE id=?", (a1,))["level"] == "E"
|
||
assert db.one("SELECT level FROM atome WHERE id=?", (a2,))["level"] == "S"
|
||
|
||
|
||
def test_level_konflikt_absenkung():
|
||
# v(E) braucht z(S), z braucht w(M) → Kaskade senkt z UND w auf E
|
||
topic = topic_anlegen("konflikt")
|
||
run = run_anlegen(topic)
|
||
ziel = db.insert("lernziele", topic=topic, text="Kann X", status="aktiv")
|
||
def atom(titel, level):
|
||
return db.insert("atome", topic=topic, titel=titel, typ="begriff",
|
||
definition="d", level=level, status="neu", ziel_id=ziel,
|
||
braucht=db.j([]))
|
||
v, z, w = atom("V", "E"), atom("Z", "S"), atom("W", "M")
|
||
for von, zu in ((v, z), (z, w)):
|
||
db.execute("INSERT INTO kanten(topic, von_atom, zu_atom, art)"
|
||
" VALUES(?,?,?,'braucht')", (topic, von, zu))
|
||
struktur._level_konflikte_loesen(topic)
|
||
assert db.one("SELECT level FROM atome WHERE id=?", (z,))["level"] == "E"
|
||
assert db.one("SELECT level FROM atome WHERE id=?", (w,))["level"] == "E"
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
assert not [b for b in struktur.messen(ctx) if b["art"] == "level_konflikt"]
|
||
|
||
|
||
async def test_level_split_ordnung_kapitel():
|
||
# Ein Ziel mit 4×E + 4×M → zwei Bausteine mit level, E-ord < M-ord,
|
||
# Kapitel je Durchgang, kein level_mix/kapitel_level
|
||
topic = topic_anlegen("split")
|
||
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(4):
|
||
db.insert("atome", topic=topic, titel=f"E{i}", typ="begriff", definition="d",
|
||
level="E", status="neu", soll_id=soll, ziel_id=ziel, braucht=db.j([]))
|
||
db.insert("atome", topic=topic, titel=f"M{i}", typ="begriff", definition="d",
|
||
level="M", status="neu", soll_id=soll, ziel_id=ziel, braucht=db.j([]))
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
ctx.ebene = "struktur"
|
||
struktur._bausteine_schneiden(ctx)
|
||
bausteine = db.query("SELECT * FROM bausteine WHERE topic=? ORDER BY ord", (topic,))
|
||
assert [b["level"] for b in bausteine] == ["E", "M"]
|
||
await struktur._kapitel_bilden(ctx)
|
||
kaps = db.query("SELECT * FROM kapitel WHERE topic=? ORDER BY ord", (topic,))
|
||
assert [k["level"] for k in kaps] == ["E", "M"]
|
||
arten = {b["art"] for b in struktur.messen(ctx)}
|
||
assert not arten & {"level_mix", "kapitel_level", "band", "partition"}
|
||
|
||
|
||
def test_kein_merge_ueber_level():
|
||
# kleine E-Gruppe darf NICHT mit kleiner M-Gruppe mergen
|
||
topic = topic_anlegen("levelmerge")
|
||
run = run_anlegen(topic)
|
||
soll = db.insert("soll", topic=topic, punkt="P", status="bestaetigt", belege=db.j([]))
|
||
z1 = db.insert("lernziele", topic=topic, text="Kann P1", soll_id=soll, status="aktiv")
|
||
z2 = db.insert("lernziele", topic=topic, text="Kann P2", soll_id=soll, status="aktiv")
|
||
for i in range(2):
|
||
db.insert("atome", topic=topic, titel=f"E{i}", typ="begriff", definition="d",
|
||
level="E", status="neu", soll_id=soll, ziel_id=z1, braucht=db.j([]))
|
||
db.insert("atome", topic=topic, titel=f"M{i}", typ="begriff", definition="d",
|
||
level="M", status="neu", soll_id=soll, ziel_id=z2, braucht=db.j([]))
|
||
ctx = llm.Kontext(run, topic, "minimax")
|
||
struktur._bausteine_schneiden(ctx)
|
||
for b in db.query("SELECT * FROM bausteine WHERE topic=?", (topic,)):
|
||
levels = {a["level"] for a in struktur._atome(topic)
|
||
if a["baustein_id"] == b["id"]}
|
||
assert levels == {b["level"]}
|