93 lines
4.8 KiB
Python
93 lines
4.8 KiB
Python
"""Topic-Transfer (server-push/-pull): Export → Löschen → Import erhält
|
||
Struktur, Verweise und Guide."""
|
||
|
||
import json
|
||
import re
|
||
|
||
import db
|
||
import guide
|
||
import pipeline
|
||
import transfer
|
||
from conftest import topic_anlegen
|
||
|
||
|
||
async def test_export_import_roundtrip():
|
||
topic = topic_anlegen("fake-transfer", art="thema")
|
||
pipeline.lauf_starten(topic)
|
||
await pipeline._laeufe[topic]
|
||
|
||
tabellen = ("quellen", "soll", "themen", "atome", "anker", "kanten", "artefakte",
|
||
"lernziele", "bausteine", "kapitel", "sections", "auftraege",
|
||
"runs", "befunde", "events", "leitner")
|
||
vorher = {t: len(db.query(f"SELECT * FROM {t}", ())) for t in tabellen}
|
||
md_vorher = guide.guide_markdown(topic)
|
||
|
||
d = json.loads(json.dumps(transfer.export(topic))) # JSON-Roundtrip wie über ssh
|
||
transfer.importieren(d)
|
||
|
||
nachher = {t: len(db.query(f"SELECT * FROM {t}", ())) for t in tabellen}
|
||
assert nachher == vorher
|
||
# Guide identisch bis auf die neu vergebenen Atom-ids in den Markern
|
||
norm = lambda s: re.sub(r"atom: \d+", "atom: X", s)
|
||
assert norm(guide.guide_markdown(topic)) == norm(md_vorher)
|
||
# Verweise zeigen auf existierende Zeilen
|
||
for a in db.query("SELECT * FROM atome WHERE topic=? AND status NOT IN"
|
||
" ('gemerged','verworfen')", (topic,)):
|
||
assert db.one("SELECT id FROM anker WHERE atom_id=?", (a["id"],))
|
||
assert a["soll_id"] and db.one("SELECT id FROM soll WHERE id=?", (a["soll_id"],))
|
||
# Themen-Verweise remappt: bestätigte Punkte und Bausteine zeigen auf echte Themen
|
||
for p in db.query("SELECT * FROM soll WHERE topic=? AND status='bestaetigt'", (topic,)):
|
||
assert p["thema_id"] and db.one("SELECT id FROM themen WHERE id=?", (p["thema_id"],))
|
||
for b in db.query("SELECT * FROM bausteine WHERE topic=?", (topic,)):
|
||
assert b["thema_id"] and db.one("SELECT id FROM themen WHERE id=?", (b["thema_id"],))
|
||
for s in db.query("SELECT belege FROM soll WHERE topic=? AND status='bestaetigt'",
|
||
(topic,)):
|
||
for b in db.uj(s["belege"]):
|
||
assert db.one("SELECT id FROM quellen WHERE id=?", (b["quelle"],))
|
||
for b in db.query("SELECT * FROM bausteine WHERE topic=?", (topic,)):
|
||
assert db.one("SELECT id FROM kapitel WHERE id=?", (b["kapitel_id"],))
|
||
assert db.one("SELECT baustein_id FROM sections WHERE baustein_id=?", (b["id"],))
|
||
# Marker-Remap: jede Marker-id im importierten Text existiert (aak: 421/421 tot)
|
||
atom_ids = {a["id"] for a in db.query("SELECT id FROM atome WHERE topic=?", (topic,))}
|
||
for s in db.query("SELECT s.* FROM sections s JOIN bausteine b ON b.id=s.baustein_id"
|
||
" WHERE b.topic=?", (topic,)):
|
||
for feld in ("text_lang", "text_kompakt"):
|
||
for m in re.findall(r"<!--\s*atom:\s*(\d+)\s*\|", s[feld] or ""):
|
||
assert int(m) in atom_ids, f"toter Marker {m}"
|
||
|
||
|
||
async def test_import_remappt_auftrag_atom_ids():
|
||
"""Aufträge nennen Atom-IDs im Klartext — nach Import müssen sie auf die neu
|
||
vergebenen IDs zeigen. Alt-Exporte mit befunde-Key: KRITISCH-Arten werden
|
||
Zeilen, Rest fliegt (wie die Schema-Migration)."""
|
||
topic = topic_anlegen("auftrag-remap", art="thema")
|
||
pipeline.lauf_starten(topic)
|
||
await pipeline._laeufe[topic]
|
||
b = db.one("SELECT * FROM bausteine WHERE topic=? LIMIT 1", (topic,))
|
||
a = db.one("SELECT id FROM atome WHERE baustein_id=?", (b["id"],))
|
||
db.insert("auftraege", baustein_id=b["id"], art="falsch",
|
||
detail=f"Atom {a['id']} widerspricht dem Text")
|
||
d = json.loads(json.dumps(transfer.export(topic)))
|
||
# Alt-Export simulieren: befunde-Freitext an einer Section, braucht-Spalte
|
||
# an Atomen, braucht-Kante — alles muss crashfrei toleriert werden
|
||
d["sections"][0]["befunde"] = db.j([
|
||
f"KRITISCH (luecke): Atom {a['id']} fehlt inhaltlich",
|
||
"Langtext, Absatz 3 hat 120 Wörter (Regel: 40–90): teilen."]) # → verwerfen
|
||
for z in d["atome"]:
|
||
z["braucht"] = "[]"
|
||
d["kanten"].append({"id": 999, "topic": topic, "von_atom": d["atome"][0]["id"],
|
||
"zu_atom": d["atome"][-1]["id"], "art": "braucht",
|
||
"status": "aktiv"})
|
||
transfer.importieren(d)
|
||
assert db.query("SELECT * FROM kanten WHERE art='braucht'") == []
|
||
atom_ids = {r["id"] for r in db.query("SELECT id FROM atome WHERE topic=?", (topic,))}
|
||
zeilen = db.query("SELECT au.* FROM auftraege au JOIN bausteine b"
|
||
" ON b.id=au.baustein_id WHERE b.topic=?", (topic,))
|
||
assert {z["art"] for z in zeilen} == {"falsch", "luecke"} # det-Zeile verworfen
|
||
treffer = 0
|
||
for z in zeilen:
|
||
for m in re.findall(r"Atom (\d+)", z["detail"]):
|
||
assert int(m) in atom_ids, f"toter Auftrag-Verweis {m}"
|
||
treffer += 1
|
||
assert treffer == 2
|