Lars
1166a0c7f1
All checks were successful
Deploy Development / deploy (push) Successful in 48s
Test Suite / pytest-backend (push) Successful in 2m20s
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 19s
Test Suite / playwright-smoke (push) Successful in 14s
Migration 023, Engine, API und ADP für Durchführungsplan getrennt vom Gate-Graph; Docs und Sprint-Assignment synchronisiert. Co-authored-by: Cursor <cursoragent@cursor.com>
117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
"""Unit tests for execution graph engine (AP1.16b)."""
|
|
|
|
from steering.graph.execution_engine import (
|
|
compute_execution_graph_state,
|
|
compute_planning_debt,
|
|
)
|
|
|
|
|
|
def _action(action_id: str, status: str = "open", **kwargs):
|
|
return {
|
|
"id": action_id,
|
|
"status": status,
|
|
"sort_order": kwargs.get("sort_order", 0),
|
|
"title": kwargs.get("title", action_id),
|
|
"action_kind": kwargs.get("action_kind", "delivery"),
|
|
"roadmap_item_id": kwargs.get("roadmap_item_id"),
|
|
}
|
|
|
|
|
|
def test_requires_blocks_until_predecessor_done():
|
|
actions = [_action("b", "open"), _action("a", "open")]
|
|
deps = [
|
|
{
|
|
"predecessor_action_id": "b",
|
|
"successor_action_id": "a",
|
|
"dependency_kind": "requires",
|
|
}
|
|
]
|
|
state = compute_execution_graph_state(actions=actions, dependencies=deps)
|
|
assert state["items"]["a"]["blocked"] is True
|
|
assert state["items"]["a"]["blocked_by"] == ["b"]
|
|
assert "a" in state["blocked_actions"]
|
|
assert "a" not in state["ready_actions"]
|
|
|
|
actions[0]["status"] = "done"
|
|
state = compute_execution_graph_state(actions=actions, dependencies=deps)
|
|
assert state["items"]["a"]["ready"] is True
|
|
assert "a" in state["ready_actions"]
|
|
|
|
|
|
def test_sequential_fallback_without_edges():
|
|
actions = [
|
|
_action("first", "open", sort_order=0),
|
|
_action("second", "open", sort_order=1),
|
|
]
|
|
state = compute_execution_graph_state(actions=actions, dependencies=[])
|
|
assert state["items"]["first"]["ready"] is True
|
|
assert state["items"]["second"]["blocked"] is True
|
|
assert state["items"]["second"]["blocked_by"] == ["first"]
|
|
|
|
actions[0]["status"] = "done"
|
|
state = compute_execution_graph_state(actions=actions, dependencies=[])
|
|
assert state["items"]["second"]["ready"] is True
|
|
|
|
|
|
def test_blocks_edge():
|
|
actions = [_action("blocker", "in_progress"), _action("blocked", "open")]
|
|
deps = [
|
|
{
|
|
"predecessor_action_id": "blocker",
|
|
"successor_action_id": "blocked",
|
|
"dependency_kind": "blocks",
|
|
}
|
|
]
|
|
state = compute_execution_graph_state(actions=actions, dependencies=deps)
|
|
assert state["items"]["blocked"]["blocked"] is True
|
|
|
|
actions[0]["status"] = "done"
|
|
state = compute_execution_graph_state(actions=actions, dependencies=deps)
|
|
assert state["items"]["blocked"]["ready"] is True
|
|
|
|
|
|
def test_critical_path():
|
|
actions = [
|
|
_action("a", "done", sort_order=0),
|
|
_action("b", "open", sort_order=1),
|
|
_action("c", "open", sort_order=2),
|
|
]
|
|
deps = [
|
|
{
|
|
"predecessor_action_id": "a",
|
|
"successor_action_id": "b",
|
|
"dependency_kind": "requires",
|
|
},
|
|
{
|
|
"predecessor_action_id": "b",
|
|
"successor_action_id": "c",
|
|
"dependency_kind": "requires",
|
|
},
|
|
]
|
|
state = compute_execution_graph_state(actions=actions, dependencies=deps)
|
|
assert state["critical_path"] == ["a", "b", "c"]
|
|
|
|
|
|
def test_planning_debt_active_gate_without_actions():
|
|
actions = [_action("x", roadmap_item_id="gate-2")]
|
|
roadmap_items = [
|
|
{"id": "gate-1", "status": "active", "title": "G5"},
|
|
{"id": "gate-2", "status": "active", "title": "G6"},
|
|
]
|
|
debts = compute_planning_debt(actions=actions, roadmap_items=roadmap_items)
|
|
assert len(debts) == 1
|
|
assert debts[0]["roadmap_item_id"] == "gate-1"
|
|
|
|
|
|
def test_scope_filters_to_gate():
|
|
actions = [
|
|
_action("a", roadmap_item_id="g1", sort_order=0),
|
|
_action("b", roadmap_item_id="g2", sort_order=0),
|
|
]
|
|
state = compute_execution_graph_state(
|
|
actions=actions,
|
|
dependencies=[],
|
|
scope_roadmap_item_id="g1",
|
|
)
|
|
assert set(state["items"].keys()) == {"a"}
|