61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
"""Graph-Validierung: DAG-Pflicht, Rück-Kanten nur von Gates, Skills vorhanden."""
|
|
import pytest
|
|
import yaml
|
|
|
|
from backend import config, engine, graph
|
|
|
|
|
|
def test_pipeline_yaml_valide():
|
|
g = graph.laden(set(engine.WORKER))
|
|
assert len(g.knoten) >= 20
|
|
assert g.stufen == ["korpus", "inventar", "struktur", "guide", "montage"]
|
|
for stufe in g.stufen[:-1]:
|
|
assert g.gate_der_stufe(stufe) is not None
|
|
|
|
|
|
def _mit_kante(extra) -> None:
|
|
daten = yaml.safe_load(config.PIPELINE_YAML.read_text())
|
|
daten["kanten"].append(extra)
|
|
import tempfile
|
|
from pathlib import Path
|
|
with tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False) as f:
|
|
yaml.safe_dump(daten, f)
|
|
pfad = Path(f.name)
|
|
try:
|
|
graph.laden(set(engine.WORKER), pfad=pfad)
|
|
finally:
|
|
pfad.unlink()
|
|
graph.laden(set(engine.WORKER)) # Original wiederherstellen
|
|
|
|
|
|
def test_zyklus_wird_abgelehnt():
|
|
with pytest.raises(graph.GraphFehler, match="Zyklus"):
|
|
_mit_kante({"von": "montage", "nach": "recherche_plan", "art": "normal"})
|
|
|
|
|
|
def test_ruecklauf_nur_von_gates():
|
|
with pytest.raises(graph.GraphFehler, match="nur von Gates"):
|
|
_mit_kante({"von": "writer", "nach": "suche", "art": "ruecklauf"})
|
|
|
|
|
|
def test_layout_vollstaendig():
|
|
lay = graph.get().layout()
|
|
assert set(lay["knoten"]) == set(graph.get().knoten)
|
|
assert all("spalte" in v and "zeile" in v for v in lay["knoten"].values())
|
|
|
|
|
|
def test_skills_von_eine_wahrheit():
|
|
"""Worker-Skills kommen NUR aus pipeline.yaml; nicht deklarierte
|
|
extra_skills sind ein harter Fehler (Drift bricht laut, nicht still)."""
|
|
assert graph.skills_von("writer")[-1] == "writer"
|
|
assert graph.skills_von("fix", extra="klaerung")[-1] == "klaerung"
|
|
with pytest.raises(graph.GraphFehler, match="nicht deklariert"):
|
|
graph.skills_von("writer", extra="gibtsnicht")
|
|
|
|
|
|
def test_timeout_und_runden_aus_dem_graph():
|
|
assert graph.timeout_fuer("writer") == 420
|
|
assert graph.timeout_fuer("dedup", n=10) == 120 + 8 * 10
|
|
assert graph.timeout_fuer("unbekannter_knoten") == 180 # Default
|
|
assert graph.get().knoten["gate_guide"].runden_max == 12
|