Spaces:
Runtime error
Runtime error
| 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) | |