22 lines
725 B
Python
22 lines
725 B
Python
from ai_orchestrator.domain.enums import NodeStatus, NodeType
|
|
from ai_orchestrator.domain.models import ExecutionGraph, ExecutionNode
|
|
|
|
|
|
def test_graph_marks_pending_node_ready_when_dependencies_are_completed() -> None:
|
|
first = ExecutionNode(task_id="task_1", node_type=NodeType.PLANNER, input_data={})
|
|
second = ExecutionNode(
|
|
task_id="task_1",
|
|
node_type=NodeType.FINALIZER,
|
|
input_data={},
|
|
dependencies=[first.node_id],
|
|
)
|
|
first.mark_completed({"ok": True})
|
|
graph = ExecutionGraph(task_id="task_1", nodes=[first, second])
|
|
|
|
ready = graph.ready_nodes()
|
|
|
|
assert len(ready) == 1
|
|
assert ready[0].node_id == second.node_id
|
|
assert second.status == NodeStatus.READY
|
|
|