Lars
d7aefed0de
All checks were successful
Deploy Development / deploy (push) Successful in 50s
Test Suite / pytest-backend (push) Successful in 2m18s
Test Suite / lint-backend (push) Successful in 3s
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
Co-authored-by: Cursor <cursoragent@cursor.com>
71 lines
2.1 KiB
Python
71 lines
2.1 KiB
Python
"""Tests for join/branch topology — AP1.15c."""
|
|
|
|
from steering.graph.join_branch import (
|
|
apply_deferred_topology,
|
|
build_topology_metadata,
|
|
)
|
|
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_optional_branch_does_not_block():
|
|
items = [_item("join", "planned"), _item("opt", "planned"), _item("req", "planned")]
|
|
deps = [
|
|
{
|
|
"from_item_id": "join",
|
|
"to_item_id": "req",
|
|
"dependency_type": "requires",
|
|
},
|
|
{
|
|
"from_item_id": "join",
|
|
"to_item_id": "opt",
|
|
"dependency_type": "optional_branch",
|
|
},
|
|
]
|
|
state = compute_initiative_graph_state(items=items, dependencies=deps)
|
|
assert state["items"]["join"]["blocked"] is True
|
|
assert state["items"]["join"]["blocked_by"] == ["req"]
|
|
assert "opt" not in state["items"]["join"]["blocked_by"]
|
|
assert state["items"]["join"]["optional_prerequisites"] == ["opt"]
|
|
|
|
|
|
def test_parallel_group_metadata():
|
|
items = [_item("a"), _item("b"), _item("c")]
|
|
deps = [
|
|
{
|
|
"from_item_id": "b",
|
|
"to_item_id": "c",
|
|
"dependency_type": "parallel_group",
|
|
"group_key": "phase-1",
|
|
}
|
|
]
|
|
topology = build_topology_metadata(dependencies=deps, item_by_id={i["id"]: i for i in items})
|
|
assert set(topology["parallel_groups"]["phase-1"]) == {"b", "c"}
|
|
|
|
|
|
def test_apply_deferred_topology_strips_optional_blockers():
|
|
item_by_id = {"x": _item("x"), "y": _item("y")}
|
|
blocked_by_map = {"x": ["y", "z"]}
|
|
deps = [
|
|
{
|
|
"from_item_id": "x",
|
|
"to_item_id": "y",
|
|
"dependency_type": "optional_branch",
|
|
}
|
|
]
|
|
result = apply_deferred_topology(
|
|
dependencies=deps,
|
|
item_by_id=item_by_id,
|
|
blocked_by_map=blocked_by_map,
|
|
)
|
|
assert result["x"] == ["z"]
|