update

tests/test_aktualisieren.py

@@ -0,0 +1,80 @@
+"""Spalten-Aktualisierung: Ebene erneut fahren ohne Reset, ohne Status-Rückschritt."""
+
+import db
+import pipeline
+import pytest
+from conftest import topic_anlegen
+
+
+async def _voll_lauf(topic):
+    pipeline.lauf_starten(topic)
+    await pipeline._laeufe[topic]
+    assert db.one("SELECT status FROM topics WHERE name=?", (topic,))["status"] == "fertig"
+
+
+async def test_aktualisieren_fuellt_nach_ohne_statusaenderung():
+    topic = topic_anlegen("akt1")
+    await _voll_lauf(topic)
+    # eine verifizierte Flashcard „verlieren" + Kaskade nach artefakte abschalten
+    opfer = db.one("SELECT ar.id FROM artefakte ar JOIN atome a ON a.id=ar.atom_id"
+                   " WHERE a.topic=? AND ar.typ='flashcard' AND ar.status='verifiziert'",
+                   (topic,))
+    db.execute("DELETE FROM artefakte WHERE id=?", (opfer["id"],))
+    db.update("topics", "name", topic, auto=db.j({"artefakte": False}))
+    run_id = pipeline.aktualisierung_starten(topic, "artefakte")
+    await pipeline._laeufe[topic]
+    run = db.one("SELECT * FROM runs WHERE id=?", (run_id,))
+    assert run["status"] == "done" and "Auto-Stopp nach artefakte" in run["grund"]
+    # nachgefüllt + Status NICHT zurückgedreht
+    atom = db.one("SELECT atom_id FROM artefakte WHERE id=?", (opfer["id"],)) or {}
+    fehlt = db.query("SELECT a.id FROM atome a WHERE a.topic=? AND a.status NOT IN"
+                     " ('gemerged','verworfen') AND NOT EXISTS (SELECT 1 FROM artefakte ar"
+                     " WHERE ar.atom_id=a.id AND ar.typ='flashcard' AND ar.status='verifiziert')",
+                     (topic,))
+    assert fehlt == []
+    assert db.one("SELECT status FROM topics WHERE name=?", (topic,))["status"] == "fertig"
+
+
+async def test_aktualisieren_kaskadiert_mit_auto():
+    topic = topic_anlegen("akt2")
+    await _voll_lauf(topic)
+    run_id = pipeline.aktualisierung_starten(topic, "struktur")
+    await pipeline._laeufe[topic]
+    run = db.one("SELECT * FROM runs WHERE id=?", (run_id,))
+    assert run["status"] == "done" and not (run["grund"] or "").startswith("Auto-Stopp")
+    assert db.one("SELECT status FROM topics WHERE name=?", (topic,))["status"] == "fertig"
+    letzte = db.one("SELECT ebene FROM runs WHERE id=?", (run_id,))["ebene"]
+    assert letzte == "guide"  # bis zur letzten Ebene durchgelaufen
+
+
+async def test_tief_reverifiziert_artefakte_und_guide():
+    topic = topic_anlegen("akt4")
+    await _voll_lauf(topic)
+    vorher = db.one("SELECT COUNT(*) n FROM artefakte ar JOIN atome a ON a.id=ar.atom_id"
+                    " WHERE a.topic=? AND ar.status='verifiziert'", (topic,))["n"]
+    assert vorher > 0
+    pipeline._tief_zuruecksetzen(topic, "artefakte")
+    offen = db.one("SELECT COUNT(*) n FROM artefakte ar JOIN atome a ON a.id=ar.atom_id"
+                   " WHERE a.topic=? AND ar.status='kandidat'", (topic,))["n"]
+    assert offen == vorher                      # alles zurück in die Prüfung
+    pipeline._tief_zuruecksetzen(topic, "guide")
+    assert db.one("SELECT COUNT(*) n FROM sections s JOIN bausteine b ON b.id=s.baustein_id"
+                  " WHERE b.topic=? AND s.stage='done'", (topic,))["n"] == 0
+    # tief-Lauf beurteilt alles neu (Fake-Panel: ok) → wieder verifiziert, Status bleibt
+    db.update("topics", "name", topic, auto=db.j({"artefakte": False}))
+    pipeline.aktualisierung_starten(topic, "artefakte", tief=True)
+    await pipeline._laeufe[topic]
+    nachher = db.one("SELECT COUNT(*) n FROM artefakte ar JOIN atome a ON a.id=ar.atom_id"
+                     " WHERE a.topic=? AND ar.status='verifiziert'", (topic,))["n"]
+    assert nachher >= vorher
+    assert db.one("SELECT status FROM topics WHERE name=?", (topic,))["status"] == "fertig"
+
+
+async def test_aktualisieren_guards():
+    topic = topic_anlegen("akt3")
+    with pytest.raises(ValueError):
+        pipeline.aktualisierung_starten(topic, "gibtsnicht")
+    pipeline.lauf_starten(topic)
+    with pytest.raises(RuntimeError):
+        pipeline.aktualisierung_starten(topic, "artefakte")
+    await pipeline._laeufe[topic]

tests/test_belege.py

@@ -0,0 +1,54 @@
+"""Beleg-Fenster: Faktenbasis = Quellumgebung des Ankers (Rollen-Trennung)."""
+
+import belege
+import db
+from conftest import topic_anlegen
+
+
+def _quelle(topic, tmp_path, text, name="q.md"):
+    p = tmp_path / name
+    p.write_text(text, encoding="utf-8")
+    return db.insert("quellen", topic=topic, art="datei", titel=name,
+                     snapshot=str(p), hash=name + str(tmp_path), status="atome")
+
+
+def _atom(topic):
+    return db.insert("atome", topic=topic, titel="T", typ="verfahren",
+                     definition="d", status="neu", braucht=db.j([]))
+
+
+def test_fenster_umgibt_anker(tmp_path):
+    topic = topic_anlegen("bel1")
+    text = "vorher " * 100 + "ANKERZITAT HIER" + " nachher" * 300
+    q = _quelle(topic, tmp_path, text)
+    a = _atom(topic)
+    start = text.index("ANKERZITAT")
+    db.insert("anker", atom_id=a, quelle_id=q, start=start,
+              ende=start + len("ANKERZITAT HIER"), zitat="ANKERZITAT HIER")
+    (f,) = belege.fenster_liste(a)
+    assert "ANKERZITAT HIER" in f
+    assert "vorher" in f and "nachher" in f          # Umgebung beidseitig dabei
+    assert len(f) <= 300 + 1200 + 20                  # Fenster-Deckel greift
+
+
+def test_merge_fallback_und_fehlende_datei(tmp_path):
+    topic = topic_anlegen("bel2")
+    text = "start " * 50 + "ZITAT-A und gleich daneben ZITAT-B" + " ende" * 50
+    q = _quelle(topic, tmp_path, text)
+    a = _atom(topic)
+    ia, ib = text.index("ZITAT-A"), text.index("ZITAT-B")
+    db.insert("anker", atom_id=a, quelle_id=q, start=ia, ende=ia + 7, zitat="ZITAT-A")
+    db.insert("anker", atom_id=a, quelle_id=q, start=ib, ende=ib + 7, zitat="ZITAT-B")
+    bloecke = belege.fenster_liste(a)
+    assert len(bloecke) == 1                          # überlappende Fenster gemerged
+    assert "ZITAT-A" in bloecke[0] and "ZITAT-B" in bloecke[0]
+    # Anker ohne Span → Zitat-Fallback
+    b = _atom(topic)
+    db.insert("anker", atom_id=b, quelle_id=q, start=-1, ende=-1, zitat="Nur das Zitat.")
+    assert belege.fenster_liste(b) == ["Nur das Zitat."]
+    # Snapshot-Datei fehlt → Zitat-Fallback (nie schlechter als vorher)
+    c = _atom(topic)
+    q2 = db.insert("quellen", topic=topic, art="datei", titel="tot",
+                   snapshot=str(tmp_path / "gibts-nicht.md"), hash="tot", status="atome")
+    db.insert("anker", atom_id=c, quelle_id=q2, start=5, ende=10, zitat="Fallback-Zitat.")
+    assert belege.fenster_liste(c) == ["Fallback-Zitat."]

tests/test_lesemodus.py

@@ -27,6 +27,61 @@ def test_baustein_karten_nur_verifizierte_flashcards():
     assert karten[0]["box"] == 1                     # Default ohne Leitner-Eintrag
 
 
+def _aussage(topic, atom_id, paar, wahr, text):
+    return db.insert("artefakte", atom_id=atom_id, typ="aussage", status="verifiziert",
+                     inhalt=db.j({"text": text, "wahr": wahr, "paar": paar,
+                                  "erklaerung": "die wahre Fassung"}))
+
+
+def test_pruefung_panels_anti_cheat_und_paar_regel():
+    topic = topic_anlegen("pf1")
+    b, a = _setup(topic)
+    for p in range(4):  # 4 Paare = 8 Aussagen → mind. 1 volles Panel möglich
+        _aussage(topic, a, f"{a}-{p}", True, f"wahr {p}")
+        _aussage(topic, a, f"{a}-{p}", False, f"falsch {p}")
+    panels = main.pruefung_holen(topic, b)
+    assert panels and all(len(p["panel"]) == 4 for p in panels)
+    for p in panels:
+        for x in p["panel"]:
+            assert set(x) == {"id", "text"}  # kein wahr/erklaerung (Anti-Cheat)
+        paare = [db.uj(db.one("SELECT inhalt FROM artefakte WHERE id=?", (x["id"],))
+                       ["inhalt"], {}).get("paar") for x in p["panel"]]
+        assert len(set(paare)) == 4  # kein Paar-Partner im selben Panel
+
+
+def test_pruefung_scoring_exakte_teilmenge():
+    topic = topic_anlegen("pf2")
+    b, a = _setup(topic)
+    ids_wahr = [_aussage(topic, a, f"{a}-{p}", True, f"w{p}") for p in range(2)]
+    ids_falsch = [_aussage(topic, a, f"{a}-{p+2}", False, f"f{p}") for p in range(2)]
+    ids = ids_wahr + ids_falsch
+    r = main.pruefung_antwort(topic, main.PruefungAntwort(ids=ids, angekreuzt=ids_wahr))
+    assert r["richtig"] and r["xp"] == 1
+    assert all(set(x) == {"id", "wahr", "erklaerung"} for x in r["aufloesung"])
+    # eine falsche angekreuzt → −1, Boden 0
+    r = main.pruefung_antwort(topic, main.PruefungAntwort(
+        ids=ids, angekreuzt=ids_wahr + ids_falsch[:1]))
+    assert not r["richtig"] and r["xp"] == 0
+    # Schwelle → fertig + lernstand-Status
+    for _ in range(10):
+        r = main.pruefung_antwort(topic, main.PruefungAntwort(ids=ids, angekreuzt=ids_wahr))
+    assert r["fertig"]
+    stand = [s for s in main.lernstand_holen(topic) if s["baustein_id"] == b][0]
+    assert stand["status"] == "fertig"
+
+
+def test_check_migration_erlaubt_aussage():
+    import sqlite3
+    con = sqlite3.connect(":memory:")
+    con.execute("CREATE TABLE artefakte(id INTEGER PRIMARY KEY, atom_id INTEGER NOT NULL,"
+                " typ TEXT NOT NULL CHECK(typ IN ('flashcard','beispiel')),"
+                " inhalt TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'kandidat')")
+    con.execute("INSERT INTO artefakte(atom_id, typ, inhalt) VALUES(1,'flashcard','{}')")
+    db._init_schema(con)  # muss den CHECK erweitern und Daten erhalten
+    con.execute("INSERT INTO artefakte(atom_id, typ, inhalt) VALUES(1,'aussage','{}')")
+    assert con.execute("SELECT COUNT(*) FROM artefakte").fetchone()[0] == 2
+
+
 def test_lernstand_und_fertig_idempotent():
     topic = topic_anlegen("lm2")
     b, _ = _setup(topic)