Spaces:
Runtime error
Runtime error
File size: 2,996 Bytes
595d60e | 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 | import pytest
from app.schemas.agent_spec import AgentSpec, ConnectionConfig, TriggerConfig, ConnectionType, TriggerType, ModeType, Step
from app.validators.workflow_validator import WorkflowValidator
def get_base_spec():
return AgentSpec(
name="Test Agent",
connection=ConnectionConfig(type=ConnectionType.NONE, config={}),
trigger=TriggerConfig(type=TriggerType.MANUAL, config={}),
mode=ModeType.WORKFLOW,
steps=[]
)
def test_workflow_duplicate_step_ids():
validator = WorkflowValidator()
spec = get_base_spec()
spec.steps = [
Step(id="step_1", name="Step A", action="do_a"),
Step(id="step_1", name="Step B", action="do_b")
]
res = validator.validate(spec)
assert res.valid is False
assert any(err.code == "DUPLICATE_STEP_ID" for err in res.errors)
def test_workflow_missing_dependency():
validator = WorkflowValidator()
spec = get_base_spec()
spec.steps = [
Step(id="step_1", name="Step A", action="do_a", depends_on=["step_2"])
]
res = validator.validate(spec)
assert res.valid is False
assert any(err.code == "MISSING_DEPENDENCY" for err in res.errors)
def test_workflow_circular_dependency():
validator = WorkflowValidator()
spec = get_base_spec()
spec.steps = [
Step(id="step_1", name="Step A", action="do_a", depends_on=["step_2"]),
Step(id="step_2", name="Step B", action="do_b", depends_on=["step_1"])
]
res = validator.validate(spec)
assert res.valid is False
assert any(err.code == "CIRCULAR_DEPENDENCY" for err in res.errors)
def test_workflow_data_flow_non_existent():
validator = WorkflowValidator()
spec = get_base_spec()
spec.steps = [
Step(id="step_1", name="Step A", action="do_a", input={"x": "$step_2.out"})
]
res = validator.validate(spec)
assert res.valid is False
assert any(err.code == "INVALID_DATA_FLOW" and "non-existent" in err.message for err in res.errors)
def test_workflow_data_flow_not_depend():
validator = WorkflowValidator()
spec = get_base_spec()
spec.steps = [
Step(id="step_1", name="Step A", action="do_a", output={"out": "str"}),
Step(id="step_2", name="Step B", action="do_b", input={"x": "$step_1.out"}, depends_on=[])
]
res = validator.validate(spec)
assert res.valid is False
assert any(err.code == "INVALID_DATA_FLOW" and "does not depend on it" in err.message for err in res.errors)
def test_workflow_data_flow_missing_field():
validator = WorkflowValidator()
spec = get_base_spec()
spec.steps = [
Step(id="step_1", name="Step A", action="do_a", output={"out": "str"}),
Step(id="step_2", name="Step B", action="do_b", input={"x": "$step_1.missing_field"}, depends_on=["step_1"])
]
res = validator.validate(spec)
assert res.valid is False
assert any(err.code == "INVALID_DATA_FLOW" and "not defined in step" in err.message for err in res.errors)
|