86 lines
2.9 KiB
Python
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
|