75 lines
3.6 KiB
Python
75 lines
3.6 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", "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"],))
|
|
# 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_befunde_atom_ids():
|
|
"""Offene Fix-Aufträge nennen Atom-IDs im Klartext — nach Import müssen sie
|
|
auf die neu vergebenen IDs zeigen, nicht auf tote."""
|
|
topic = topic_anlegen("befunde-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.update("sections", "baustein_id", b["id"],
|
|
befunde=db.j([f"KRITISCH: Marker für Atom {a['id']} fehlt — exakt einfügen."]))
|
|
d = json.loads(json.dumps(transfer.export(topic)))
|
|
transfer.importieren(d)
|
|
atom_ids = {r["id"] for r in db.query("SELECT id FROM atome WHERE topic=?", (topic,))}
|
|
treffer = 0
|
|
for s in db.query("SELECT s.befunde FROM sections s JOIN bausteine b ON b.id=s.baustein_id"
|
|
" WHERE b.topic=?", (topic,)):
|
|
for auftrag in db.uj(s["befunde"]):
|
|
for m in re.findall(r"Atom (\d+)", str(auftrag)):
|
|
assert int(m) in atom_ids, f"toter Befund-Marker {m}"
|
|
treffer += 1
|
|
assert treffer >= 1 # der eingefügte Auftrag wurde geprüft
|