update

backend/tests/invarianten.py

@@ -0,0 +1,96 @@
+"""Invarianten nach einem (Fake-)E2E-Lauf: was IMMER gelten muss, egal welches Szenario.
+
+Nutzt bewusst eigene, schlichte Prüfungen statt Pipeline-Heuristiken (Muster qa.py) —
+geteilte blinde Flecken machen den Check wertlos. Rückgabe: Liste von Verstößen,
+leer = alles konsistent.
+"""
+
+import json
+
+import database as db
+from textkit import _norm_title
+
+_LEVELS_OK = {"beginner", "advanced", "expert"}
+_RELEVANZ_OK = {"relevant", "peripheral"}
+
+
+def _json(path):
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+
+
+async def pruefe_invarianten(topic: str, files: dict | None = None,
+                             mit_artefakten: bool = True) -> list[str]:
+    fehler: list[str] = []
+    subs = [dict(r) for r in await db.list_subblocks(topic)]
+    cons = [r for r in subs if r["status"] == "consensus"]
+
+    for r in cons:
+        wo = f"{r['block']}/{r['sub_title']}"
+        fk = None
+        try:
+            fk = json.loads(r["facts"]) if r["facts"] else None
+        except ValueError:
+            fehler.append(f"facts unparsebar: {wo}")
+        if not (isinstance(fk, dict) and (fk.get("key_points") or fk.get("cited_facts"))):
+            fehler.append(f"consensus-Sub ohne facts: {wo}")
+        if r["level"] not in _LEVELS_OK:
+            fehler.append(f"consensus-Sub ohne gültiges level: {wo}")
+        if r["relevance"] not in _RELEVANZ_OK:
+            fehler.append(f"consensus-Sub ohne relevance: {wo}")
+
+    if mit_artefakten:
+        art = [dict(r) for r in await db.get_sub_artefakte(topic)]
+        fragen = [dict(r) for r in await db.list_question_pattern(topic)]
+        versorgt = {(r["block_norm"], r["sub_norm"]) for r in art}
+        versorgt |= {(r["block_norm"], r["sub_norm"]) for r in fragen}
+        lebend = {(r["block_norm"], r["sub_norm"]) for r in subs if r["status"] in ("consensus", "variant")}
+        for r in cons:
+            if r["relevance"] == "relevant" and (r["block_norm"], r["sub_norm"]) not in versorgt:
+                fehler.append(f"relevanter Sub ohne Frage/Artefakt: {r['block']}/{r['sub_title']}")
+        for bn, sn in sorted({(r["block_norm"], r["sub_norm"]) for r in art} |
+                             {(r["block_norm"], r["sub_norm"]) for r in fragen}):
+            if (bn, sn) not in lebend:
+                fehler.append(f"Waise (Ziel-Sub existiert nicht): {bn}/{sn}")
+
+    # keine hängengebliebenen Karten
+    for c in await db.kanban_cards(topic):
+        if c["stage"] == "dead":
+            fehler.append(f"dead-Karte: {c['board']}/{c['card_id']}")
+
+    if files is not None:
+        if not files["final"].exists():
+            fehler.append("blocks.md fehlt")
+        sc = _json(files["sidecar"])
+        if not isinstance(sc, dict):
+            fehler.append("sidecar-Datei fehlt/unparsebar")
+        else:  # Sidecar und DB-consensus müssen dieselbe Sub-Menge tragen
+            db_menge = {(r["block_norm"], r["sub_norm"]) for r in cons}
+            sc_menge = {(_norm_title(bt), _norm_title(str(s.get("title", ""))))
+                        for bt, ss in sc.items() for s in ss if isinstance(s, dict)}
+            for extra in sorted(sc_menge - db_menge):
+                fehler.append(f"Sidecar-Sub fehlt in DB: {extra}")
+            for extra in sorted(db_menge - sc_menge):
+                fehler.append(f"DB-consensus fehlt im Sidecar: {extra}")
+    return fehler
+
+
+async def pruefe_guide_invarianten(topic: str, format_name: str = "Guide") -> list[str]:
+    """Jeder relevante consensus-Sub trägt einen Sub-Marker im Guide (Muster
+    guide_qa.marker_fehlend, ohne LLM)."""
+    import guide_qa
+    fehler: list[str] = []
+    cards = [dict(r) for r in await db.list_guide_cards(topic, format_name)]
+    if not cards:
+        return ["keine Guide-Karten"]
+    for c in cards:
+        if c["status"] != "ok" or not (c.get("md") or "").strip():
+            fehler.append(f"Guide-Karte nicht ok: {c['block']} ({c['status']})")
+    subs_rel: dict[str, set] = {}
+    for r in await db.list_subblocks(topic):
+        if r["status"] == "consensus" and r["relevance"] != "peripheral":
+            subs_rel.setdefault(r["block_norm"], set()).add(r["sub_norm"])
+    fehler += [f"Sub-Marker fehlt: {m}" for m in guide_qa.marker_fehlend(cards, subs_rel)]
+    return fehler