update

tests/test_transfer.py

@@ -0,0 +1,44 @@
+"""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", "atome", "anker", "kanten", "artefakte",
+                "lernziele", "bausteine", "kapitel", "sections", "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"],))
+    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"],))