This commit is contained in:
team3
2026-07-06 17:32:52 +02:00
parent cc27e53b9e
commit 73ac3e532f
19 changed files with 565 additions and 523 deletions

View File

@@ -0,0 +1,40 @@
"""Auto-Repair-Loop-Primitive: Stopp bei 100 %, Stillstand, Limit."""
import auto_loop
async def test_fertig_ohne_repair():
async def rep():
raise AssertionError("darf bei 100 % nie reparieren")
r = await auto_loop.auto_repair_loop("x", 10.0, rep)
assert r == {"ebene": "x", "note": 10.0, "runden": 0, "grund": "fertig"}
async def test_verbessert_bis_voll():
noten = iter([9.0, 10.0])
async def rep():
return next(noten)
r = await auto_loop.auto_repair_loop("inv", 8.0, rep) # 8.0 → 9.0 → 10.0
assert r["grund"] == "fertig" and r["runden"] == 2
async def test_stillstand():
async def rep():
return 8.0 # Repair bewegt nichts
r = await auto_loop.auto_repair_loop("art", 8.0, rep)
assert r["grund"] == "stillstand" and r["runden"] == 1
assert r["note"] == 8.0
async def test_limit():
stand = {"n": 1.0}
async def rep():
stand["n"] += 0.1 # verbessert stetig, erreicht aber nie 10.0
return stand["n"]
r = await auto_loop.auto_repair_loop("guide", 1.0, rep, max_iter=10)
assert r["grund"] == "limit" and r["runden"] == 10

View File

@@ -960,15 +960,22 @@ async def _run_flow(ctx, files, timeout=30, **kw):
timeout=timeout)
async def test_qa_gate_pauses_on_bad_note(board_env, monkeypatch):
"""Note unter Schwelle → Flow endet sauber, Board-2-Karten warten in generate."""
async def test_qa_gate_auto_stillstand_pauses(board_env, monkeypatch):
"""Auto an, aber Repair schließt die Befunde nicht → Stillstand → Flow pausiert,
Board-2-Karten warten in generate."""
import qa as qa_mod
import repair as repair_mod
db, ctx, files = board_env
async def bad_qa(topic, llm=False):
return {"note": 5.0, "topic": TOPIC, "quoten": {}, "fremd": [], "artefakte": {}}
async def stuck_repair(topic, ebene=None):
return {"note": 5.0} # bewegt nichts
monkeypatch.setattr(qa_mod, "qa_report", bad_qa)
monkeypatch.setattr(qa_mod, "_write_report", lambda r: None)
monkeypatch.setattr(repair_mod, "repair_befunde", stuck_repair)
monkeypatch.setattr(bi, "_QA_GATE_POLL", 0.05)
await _seed(db)
ok = await _run_flow(ctx, files)
@@ -978,6 +985,51 @@ async def test_qa_gate_pauses_on_bad_note(board_env, monkeypatch):
assert await db.kanban_count(TOPIC, "done_artefact", board="artefacts") == 0
async def test_qa_gate_auto_reaches_100(board_env, monkeypatch):
"""Auto an, Repair hebt die Note auf 100 % → Gate öffnet, Board 2 läuft durch."""
import qa as qa_mod
import repair as repair_mod
db, ctx, files = board_env
async def bad_qa(topic, llm=False):
return {"note": 8.0, "topic": TOPIC, "quoten": {}, "fremd": [], "artefakte": {}}
async def good_repair(topic, ebene=None):
return {"note": 10.0}
monkeypatch.setattr(qa_mod, "qa_report", bad_qa)
monkeypatch.setattr(qa_mod, "_write_report", lambda r: None)
monkeypatch.setattr(repair_mod, "repair_befunde", good_repair)
monkeypatch.setattr(bi, "_QA_GATE_POLL", 0.05)
await _seed(db)
ok = await _run_flow(ctx, files)
assert ok
assert await db.kanban_count(TOPIC, "done_artefact", board="artefacts") >= 5
async def test_qa_gate_auto_off_pauses(board_env, monkeypatch):
"""Auto aus → nach der QA anhalten (Kontrollpunkt), egal welche Note; kein Repair."""
import qa as qa_mod
import repair as repair_mod
db, ctx, files = board_env
async def good_qa(topic, llm=False):
return {"note": 10.0, "topic": TOPIC, "quoten": {}, "fremd": [], "artefakte": {}}
async def no_repair(topic, ebene=None):
raise AssertionError("Auto aus darf nicht reparieren")
monkeypatch.setattr(qa_mod, "qa_report", good_qa)
monkeypatch.setattr(qa_mod, "_write_report", lambda r: None)
monkeypatch.setattr(repair_mod, "repair_befunde", no_repair)
monkeypatch.setattr(bi, "_QA_GATE_POLL", 0.05)
await _seed(db)
ok = await _run_flow(ctx, files, auto_inventory=False)
assert ok
warten = await db.kanban_cards(TOPIC, board="artefacts", stage="generate")
assert len(warten) == 4 # Board 2 wartet trotz Note 10.0 (Auto aus = Kontrollpunkt)
async def test_qa_gate_force_overrides(board_env, monkeypatch):
"""qa_force=True („Trotzdem fortsetzen") übersteuert die schlechte Note."""
import qa as qa_mod
@@ -994,7 +1046,7 @@ async def test_qa_gate_force_overrides(board_env, monkeypatch):
ok = await _run_flow(ctx, files, qa_force=True)
assert ok
assert await db.kanban_count(TOPIC, "done_artefact", board="artefacts") >= 5
assert called["n"] <= 1 # Gate-Lauf übersprungen; höchstens Abschluss-QA
assert called["n"] <= 2 # Gate-QA übersprungen (force); nur Artefakt-Loop + Abschluss-QA
async def test_qa_gate_off_means_no_qa_call(board_env, monkeypatch):
@@ -1011,7 +1063,7 @@ async def test_qa_gate_off_means_no_qa_call(board_env, monkeypatch):
await _seed(db)
ok = await _run_flow(ctx, files)
assert ok
assert called["n"] == 1 # kein Gate-Lauf; nur die Abschluss-QA der Lauf-Summary
assert called["n"] == 2 # kein Gate-Lauf; Artefakt-Loop-QA + Abschluss-QA der Lauf-Summary
async def test_qa_gate_fail_open(board_env, monkeypatch):

View File

@@ -146,41 +146,6 @@ def test_env_file_wins(tmp_path, monkeypatch):
assert os.environ["X_CREATOR_TESTKEY"] == "neu"
async def test_restart_artefact_card_wipes_only_that_block(testdb):
import board_inventory as bi
db = testdb
for norm in ("alpha", "beta"):
await db.kanban_upsert_card(TOPIC, "artefacts", norm, "ablock", "done_artefact",
{"title": norm.title(), "raw": {norm: ["S"]}, "facts": {}})
await db.upsert_subblock(TOPIC, norm, "s1", norm.title(), "Sub Eins")
await db.upsert_question_pattern(TOPIC, norm, "s1", norm.title(), "Sub Eins", "Frage?")
await db.put_sub_artifact(TOPIC, norm, "s1", "flashcard", norm.title(), "Sub Eins", "{}")
assert await bi.restart_artefact_card(TOPIC, "alpha") is True
assert (await db.kanban_get_card(TOPIC, "artefacts", "alpha"))["stage"] == "generate"
assert await db.list_subblocks(TOPIC, "alpha") == []
assert len(await db.list_subblocks(TOPIC, "beta")) == 1 # untouched
assert await bi.restart_artefact_card(TOPIC, "gibtsnicht") is False
async def test_guide_reset_card_single(testdb):
import guide_board as gb
db = testdb
for n in ("alpha", "beta"):
await db.upsert_guide_card(TOPIC, "Guide", n, n.title())
await db.set_guide_card(TOPIC, "Guide", n, stage="done", status="ok",
writer_rounds=2, md="# SECTION Text", gate_info="x")
await db.put_lernziel(TOPIC, n, "z1", "Ziel eins")
assert await gb.reset_card(TOPIC, "Guide", "alpha", 0) is True
cards = {c["block_norm"]: c for c in await db.list_guide_cards(TOPIC, "Guide")}
assert cards["alpha"]["stage"] == "lernziele" and cards["alpha"]["md"] == "" and cards["alpha"]["writer_rounds"] == 0
assert cards["beta"]["stage"] == "done" and cards["beta"]["md"] # untouched
assert await db.list_lernziele(TOPIC) and all(z["block_norm"] != "alpha" for z in await db.list_lernziele(TOPIC))
# ab_stage 3 (pruefer) behält md
assert await gb.reset_card(TOPIC, "Guide", "beta", 3) is True
cards = {c["block_norm"]: c for c in await db.list_guide_cards(TOPIC, "Guide")}
assert cards["beta"]["stage"] == "pruefer" and cards["beta"]["md"]
async def test_completeness_route(testdb, tmp_path, monkeypatch):
import routes, paths
db = testdb

View File

@@ -58,25 +58,6 @@ def test_auftraege_schwelle_und_kritisch():
assert len(z) == 2 and k is True
async def test_reset_from_stage(testdb):
db = testdb
await db.upsert_guide_card(TOPIC, FMT, "a", "A")
await db.upsert_guide_card(TOPIC, FMT, "b", "B")
await db.set_guide_card(TOPIC, FMT, "a", stage="done", md="text", writer_rounds=2)
await db.set_guide_card(TOPIC, FMT, "b", stage="pruefer", md="text")
await db.put_lernziel(TOPIC, "a", "z1", "Ziel")
# reset ab writer (idx 2): beide Karten zurück, md geleert, Ziele bleiben
moved = await gb.reset_from_stage(TOPIC, FMT, 2)
assert moved == 2
cards = {c["block_norm"]: c for c in await db.list_guide_cards(TOPIC, FMT)}
assert cards["a"]["stage"] == "writer" and cards["a"]["md"] == "" and cards["a"]["writer_rounds"] == 0
assert cards["b"]["stage"] == "writer"
assert await db.list_lernziele(TOPIC, "a")
# reset ab lernziele (idx 0): Ziele weg
await gb.reset_from_stage(TOPIC, FMT, 0)
assert not await db.list_lernziele(TOPIC, "a")
assert (await db.list_guide_cards(TOPIC, FMT))[0]["stage"] == "lernziele"
async def test_done_step(testdb):
db = testdb

View File

@@ -0,0 +1,48 @@
"""Guide-Ebene der Auto-Sequenz: generieren → QA → Repair-Loop bis 100 %/Stillstand."""
import blocks
import guide
import guide_qa
import guide_board
def _patch(monkeypatch, noten, calls):
async def fake_gen(*a, **k):
calls["gen"] += 1
async def fake_rep(topic, fmt):
calls["rep"] += 1
return []
async def fake_qa(topic, llm=False):
return {"note_guide": noten.pop(0) if len(noten) > 1 else noten[0]}
monkeypatch.setattr(guide, "generate_guide", fake_gen)
monkeypatch.setattr(guide_board, "repair_karten", fake_rep)
monkeypatch.setattr(guide_qa, "guide_qa_report", fake_qa)
async def test_guide_fertig_ohne_repair(testdb, monkeypatch):
blocks._blocks_errors.pop("G1", None)
calls = {"gen": 0, "rep": 0}
_patch(monkeypatch, [10.0], calls)
await blocks._guide_ebene("G1", "", "claude", lambda: False, lambda *a, **k: None)
assert calls["gen"] == 1 and calls["rep"] == 0 # nur initial gebaut
assert "G1" not in blocks._blocks_errors
async def test_guide_repariert_bis_100(testdb, monkeypatch):
blocks._blocks_errors.pop("G2", None)
calls = {"gen": 0, "rep": 0}
_patch(monkeypatch, [7.0, 10.0], calls) # erst 7.0, nach einem Repair 10.0
await blocks._guide_ebene("G2", "", "claude", lambda: False, lambda *a, **k: None)
assert calls["rep"] == 1
assert "G2" not in blocks._blocks_errors
async def test_guide_stillstand_meldet_systemfehler(testdb, monkeypatch):
blocks._blocks_errors.pop("G3", None)
calls = {"gen": 0, "rep": 0}
_patch(monkeypatch, [6.0], calls) # Repair bewegt die Note nie
await blocks._guide_ebene("G3", "", "claude", lambda: False, lambda *a, **k: None)
assert "G3" in blocks._blocks_errors
assert "Guide" in blocks._blocks_errors["G3"]