File size: 2,908 Bytes
152be26 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 | import pytest
from src.schemas import NodeDefinition, NodeType, WorkflowDefinition
from src.validator import WorkflowValidationError, topological_order, validate_workflow
def make_node(
node_id: str,
*,
depends_on: list[str] | None = None,
node_type: NodeType = NodeType.TASK,
) -> NodeDefinition:
return NodeDefinition(
node_id=node_id,
name=node_id.replace("_", " ").title(),
node_type=node_type,
depends_on=depends_on or [],
)
def make_workflow(nodes: list[NodeDefinition]) -> WorkflowDefinition:
return WorkflowDefinition(
workflow_id="test-workflow",
name="Test Workflow",
nodes=nodes,
)
def test_valid_linear_dag_passes_validation() -> None:
workflow = make_workflow(
[
make_node("intake"),
make_node("validate", depends_on=["intake"]),
make_node("finish", depends_on=["validate"], node_type=NodeType.END),
]
)
assert validate_workflow(workflow) is None
assert topological_order(workflow) == ["intake", "validate", "finish"]
def test_valid_branching_dag_has_deterministic_order() -> None:
workflow = make_workflow(
[
make_node("start"),
make_node("beta", depends_on=["start"]),
make_node("alpha", depends_on=["start"]),
make_node("finish", depends_on=["alpha", "beta"], node_type=NodeType.END),
]
)
assert topological_order(workflow) == ["start", "alpha", "beta", "finish"]
def test_independent_ready_nodes_use_stable_node_id_order() -> None:
workflow = make_workflow([make_node("beta"), make_node("alpha")])
assert topological_order(workflow) == ["alpha", "beta"]
def test_duplicate_node_ids_are_rejected() -> None:
workflow = make_workflow([make_node("same"), make_node("same")])
with pytest.raises(WorkflowValidationError, match="duplicate node_id values: same"):
validate_workflow(workflow)
def test_unknown_dependency_is_rejected() -> None:
workflow = make_workflow(
[make_node("intake"), make_node("validate", depends_on=["missing"])]
)
with pytest.raises(
WorkflowValidationError,
match="node 'validate' depends on unknown node 'missing'",
):
validate_workflow(workflow)
def test_self_dependency_is_rejected() -> None:
workflow = make_workflow([make_node("loop", depends_on=["loop"])])
with pytest.raises(
WorkflowValidationError,
match="node 'loop' cannot depend on itself",
):
validate_workflow(workflow)
def test_dependency_cycle_is_rejected() -> None:
workflow = make_workflow(
[
make_node("alpha", depends_on=["beta"]),
make_node("beta", depends_on=["alpha"]),
]
)
with pytest.raises(WorkflowValidationError, match="contains a cycle"):
validate_workflow(workflow)
|