Kairo-Jinkendo/backend/tests/test_ap14d_roadmap_graph.py
Lars 4f8725e234
All checks were successful
Deploy Development / deploy (push) Successful in 45s
Test Suite / pytest-backend (push) Successful in 2m17s
Test Suite / lint-backend (push) Successful in 2s
Test Suite / compose-smoke (push) Has been skipped
Test Suite / k6 /api/health Baseline (push) Successful in 18s
Test Suite / playwright-smoke (push) Successful in 13s
AP1.4d: Graph Engine mit blocked/ready Read Models.
Migration 020 fuer Kantenarten und Parallelgruppe; steering/graph berechnet Fortschritt; UI zeigt Bereit/Blockiert in Liste und Graph.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-11 07:59:18 +02:00

86 lines
2.9 KiB
Python

"""Unit tests for roadmap graph engine (AP1.4d)."""
from steering.graph.roadmap_engine import compute_initiative_graph_state
def _item(item_id: str, status: str = "planned", **kwargs):
return {
"id": item_id,
"status": status,
"sort_order": kwargs.get("sort_order", 0),
"sequencing_mode": kwargs.get("sequencing_mode", "sequential"),
"title": kwargs.get("title", item_id),
}
def test_requires_blocks_until_prerequisite_reached():
items = [_item("a", "planned"), _item("b", "reached")]
deps = [
{
"from_item_id": "a",
"to_item_id": "b",
"dependency_type": "requires",
}
]
state = compute_initiative_graph_state(items=items, dependencies=deps)
assert state["items"]["a"]["blocked"] is False
assert state["items"]["a"]["ready"] is True
assert "a" in state["ready_items"]
items[1]["status"] = "planned"
state = compute_initiative_graph_state(items=items, dependencies=deps)
assert state["items"]["a"]["blocked"] is True
assert state["items"]["a"]["blocked_by"] == ["b"]
assert "a" in state["blocked_items"]
def test_sequential_fallback_without_edges():
items = [
_item("first", "planned", sort_order=0),
_item("second", "planned", sort_order=1),
]
state = compute_initiative_graph_state(items=items, dependencies=[])
assert state["items"]["first"]["ready"] is True
assert state["items"]["second"]["blocked"] is True
assert state["items"]["second"]["blocked_by"] == ["first"]
items[0]["status"] = "reached"
state = compute_initiative_graph_state(items=items, dependencies=[])
assert state["items"]["second"]["ready"] is True
def test_parallel_sequencing_mode_skips_order_block():
items = [
_item("first", "planned", sort_order=0),
_item("second", "planned", sort_order=1, sequencing_mode="parallel"),
]
state = compute_initiative_graph_state(items=items, dependencies=[])
assert state["items"]["second"]["ready"] is True
def test_blocks_edge():
items = [_item("blocker", "active"), _item("blocked", "planned")]
deps = [
{
"from_item_id": "blocker",
"to_item_id": "blocked",
"dependency_type": "blocks",
}
]
state = compute_initiative_graph_state(items=items, dependencies=deps)
assert state["items"]["blocked"]["blocked"] is True
assert state["items"]["blocked"]["blocked_by"] == ["blocker"]
items[0]["status"] = "reached"
state = compute_initiative_graph_state(items=items, dependencies=deps)
assert state["items"]["blocked"]["ready"] is True
def test_fulfillment_ratio_from_criteria_progress():
items = [_item("g", "active")]
progress = {"g": {"total": 4, "closed": 2, "open": 2}}
state = compute_initiative_graph_state(
items=items, dependencies=[], criteria_progress=progress
)
assert state["items"]["g"]["fulfillment_ratio"] == 0.5