121 lines
5.2 KiB
Python
121 lines
5.2 KiB
Python
"""Fake-E2E: kompletter Graph in Sekunden. Prüft Kernziele am DB-Endzustand:
|
|
Coverage 100 % (Lücke über Web-Nachrecherche geschlossen), Dublette gemerged,
|
|
Dopplung entfernt, Marker global einmalig, Resume = 0 neue Events."""
|
|
from pathlib import Path
|
|
|
|
from backend import db, fakes, montage
|
|
from .conftest import lauf_bis_ende, topic_anlegen
|
|
|
|
|
|
def _pruefe_endzustand(topic: str):
|
|
t = db.one("SELECT * FROM topics WHERE name=?", topic)
|
|
assert t["status"] == "fertig"
|
|
# Coverage: jeder bestätigte Soll-Punkt hat ≥1 aktives Atom mit Baustein
|
|
luecken = db.query(
|
|
"SELECT s.id FROM soll s WHERE s.topic=? AND s.status='bestaetigt' AND "
|
|
"s.freispruch='' AND NOT EXISTS (SELECT 1 FROM atome a WHERE "
|
|
"a.soll_id=s.id AND a.status='aktiv' AND a.baustein_id IS NOT NULL)", topic)
|
|
assert luecken == []
|
|
assert db.one("SELECT COUNT(*) c FROM soll WHERE topic=? AND "
|
|
"status='bestaetigt'", topic)["c"] == 3
|
|
# Dublette gemerged: HTTP-Paar existiert nur einmal aktiv
|
|
http = db.query("SELECT titel FROM atome WHERE topic=? AND status='aktiv' AND "
|
|
"titel IN ('HTTP', 'Hypertext Transfer Protocol')", topic)
|
|
assert len(http) == 1
|
|
merged = db.one("SELECT id FROM atome WHERE topic=? AND merged_into IS NOT NULL",
|
|
topic)
|
|
assert merged is not None
|
|
guide = Path(montage.guide_pfad(topic)).read_text()
|
|
# Marker global genau einmal je Atom
|
|
aktive = db.query("SELECT id FROM atome WHERE topic=? AND status='aktiv' AND "
|
|
"baustein_id IS NOT NULL", topic)
|
|
for a in aktive:
|
|
assert guide.count(f"<!-- atom: {a['id']} -->") == 1
|
|
# Dopplung raus, Rückbezug als Prosa — KEINE Links im Guide (linear lesbar)
|
|
assert guide.count(fakes.DUP) <= 1
|
|
assert "Wie bereits bei den Grundlagen beschrieben" in guide
|
|
assert "](#" not in guide
|
|
import re as _re
|
|
assert not _re.search(r"^\*\*[^*\n]+\*\*\s+—", guide, _re.MULTILINE)
|
|
# Verdichtung: Splitter-Atome („Vorteil …") sind zu EINEM fusioniert
|
|
vorteile = db.query("SELECT id FROM atome WHERE topic=? AND status='aktiv' "
|
|
"AND titel LIKE 'Vorteil%'", topic)
|
|
assert len(vorteile) == 1
|
|
starts = db.query("SELECT DISTINCT start FROM anker WHERE atom_id=?",
|
|
vorteile[0]["id"])
|
|
assert len(starts) == 2 # beide Belegstellen geerbt (Leser-Duplikate egal)
|
|
# Kein Format-Leak im Guide
|
|
assert "TITEL:" not in guide and "DEFINITION:" not in guide
|
|
assert "## Fehlerbehandlung" in guide
|
|
# Lücken-Atom kam über Web-Nachrecherche
|
|
assert db.one("SELECT id FROM quellen WHERE topic=? AND zweck='luecke' "
|
|
"AND status='geladen'", topic) is not None
|
|
|
|
|
|
async def test_kompletter_lauf():
|
|
topic = topic_anlegen()
|
|
run = await lauf_bis_ende(topic)
|
|
assert run["status"] == "done", run["grund"]
|
|
_pruefe_endzustand(topic)
|
|
|
|
|
|
async def test_resume_keine_neuen_events():
|
|
topic = topic_anlegen()
|
|
run1 = await lauf_bis_ende(topic)
|
|
assert run1["status"] == "done"
|
|
vorher = db.one("SELECT COUNT(*) c FROM events")["c"]
|
|
run2 = await lauf_bis_ende(topic)
|
|
assert run2["status"] == "done"
|
|
assert db.one("SELECT COUNT(*) c FROM events")["c"] == vorher
|
|
|
|
|
|
async def test_pause_und_fortsetzen():
|
|
import asyncio
|
|
from backend import engine
|
|
topic = topic_anlegen()
|
|
run_id = engine.lauf_starten(topic)
|
|
await asyncio.sleep(0.3)
|
|
engine.lauf_pausieren(topic)
|
|
await asyncio.wait_for(engine._laeufe[topic], 10)
|
|
zwischen = db.one("SELECT * FROM runs WHERE id=?", run_id)
|
|
if zwischen["status"] == "done":
|
|
return # Lauf war schneller als die Pause — ok
|
|
assert zwischen["status"] == "paused"
|
|
run = await lauf_bis_ende(topic)
|
|
assert run["status"] == "done"
|
|
assert run["id"] == run_id # Resume nutzt denselben Run
|
|
_pruefe_endzustand(topic)
|
|
# keine Task-Duplikate
|
|
doppel = db.query("SELECT topic, knoten, item, COUNT(*) c FROM tasks "
|
|
"GROUP BY topic, knoten, item HAVING c > 1")
|
|
assert doppel == []
|
|
|
|
|
|
async def test_reset_und_neulauf():
|
|
from backend import config, engine
|
|
from backend.main import _pipeline_daten_loeschen
|
|
topic = topic_anlegen()
|
|
run1 = await lauf_bis_ende(topic)
|
|
assert run1["status"] == "done"
|
|
# Reset muss auch pausierte Läufe schließen (sonst Resume mit altem Budget)
|
|
db.update("runs", "id=?", (run1["id"],), status="paused")
|
|
_pipeline_daten_loeschen(topic)
|
|
db.execute("UPDATE runs SET status='stopped' WHERE topic=? AND "
|
|
"status IN ('running','paused','budget')", topic)
|
|
db.update("topics", "name=?", (topic,), status="neu")
|
|
assert db.one("SELECT COUNT(*) c FROM runs WHERE topic=? AND "
|
|
"status='paused'", topic)["c"] == 0
|
|
assert db.one("SELECT COUNT(*) c FROM tasks WHERE topic=?", topic)["c"] == 0
|
|
assert db.one("SELECT COUNT(*) c FROM atome WHERE topic=?", topic)["c"] == 0
|
|
assert not (config.KORPUS_DIR / topic).exists()
|
|
run2 = await lauf_bis_ende(topic)
|
|
assert run2["status"] == "done"
|
|
_pruefe_endzustand(topic)
|
|
|
|
|
|
async def test_events_haben_skill_hash():
|
|
topic = topic_anlegen()
|
|
await lauf_bis_ende(topic)
|
|
ohne = db.query("SELECT id FROM events WHERE status='ok' AND skill_hash=''")
|
|
assert ohne == []
|