Spaces:
Runtime error
Runtime error
| from typing import Any | |
| import pytest | |
| from pydantic import ValidationError as PydanticValidationError | |
| from workflows.structs import CallType, InputField, ModelStep, OutputField, Workflow | |
| from workflows.validators import ValidationError, ValidationErrorType, WorkflowValidator, _parse_variable_reference | |
| # Test Data | |
| def create_empty_step(step_id: str = "step1") -> ModelStep: | |
| """Creates a basic valid step for testing""" | |
| return ModelStep( | |
| id=step_id, | |
| name="Test Step", | |
| model="gpt-4", | |
| provider="openai", | |
| call_type=CallType.LLM, | |
| temperature=0.7, | |
| system_prompt="Test prompt", | |
| input_fields=[], | |
| output_fields=[], | |
| ) | |
| # Test Data | |
| def create_basic_step(step_id: str = "step1") -> ModelStep: | |
| """Creates a basic valid step for testing""" | |
| return ModelStep( | |
| id=step_id, | |
| name="Test Step", | |
| model="gpt-4", | |
| provider="openai", | |
| call_type=CallType.LLM, | |
| temperature=0.7, | |
| system_prompt="Test prompt", | |
| input_fields=[InputField(name="input", description="test", variable="external_input")], | |
| output_fields=[OutputField(name="output", description="test", type="str")], | |
| ) | |
| def create_basic_workflow(steps: list[ModelStep] | None = None) -> Workflow: | |
| """Creates a basic valid workflow for testing""" | |
| if steps is None: | |
| steps = [create_empty_step()] | |
| return Workflow(inputs=[], outputs={}, steps={step.id: step for step in steps}) | |
| # Additional Test Data | |
| def create_step_with_fields( | |
| step_id: str, input_fields: list[InputField], output_fields: list[OutputField] | |
| ) -> ModelStep: | |
| """Creates a step with specific input and output fields""" | |
| return ModelStep( | |
| id=step_id, | |
| name="Test Step", | |
| model="gpt-4", | |
| provider="openai", | |
| call_type=CallType.LLM, | |
| temperature=0.7, | |
| system_prompt="Test prompt", | |
| input_fields=input_fields, | |
| output_fields=output_fields, | |
| ) | |
| def create_valid_workflow() -> Workflow: | |
| # Create a step with input and output fields | |
| step = create_step_with_fields( | |
| "step1", | |
| [InputField(name="input", description="test", variable="external_input")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| # Create workflow with the single step | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step1.output"} | |
| return workflow | |
| # Basic Workflow Validation Tests | |
| class TestBasicWorkflowValidation: | |
| def test_empty_workflow(self): | |
| """Test validation of empty workflow""" | |
| validator = WorkflowValidator() | |
| workflow = Workflow(inputs=["input"], outputs={"output": "input"}, steps={}) | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.GENERAL | |
| assert "must contain at least one step" in validator.errors[0].message | |
| def test_workflow_without_inputs(self): | |
| """Test validation of workflow without inputs""" | |
| validator = WorkflowValidator() | |
| workflow = create_basic_workflow() | |
| workflow.inputs = [] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.GENERAL | |
| assert "must contain at least one input" in validator.errors[0].message | |
| def test_workflow_without_outputs(self): | |
| """Test validation of workflow without outputs""" | |
| validator = WorkflowValidator() | |
| workflow = create_basic_workflow() | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.GENERAL | |
| assert "must contain at least one output" in validator.errors[0].message | |
| def test_single_step_workflow(self): | |
| """Test validation of valid single-step workflow""" | |
| validator = WorkflowValidator() | |
| # Create a step with input and output fields | |
| workflow = create_valid_workflow() | |
| assert validator.validate(workflow) | |
| assert len(validator.errors) == 0 | |
| # Step Validation Tests | |
| class TestStepValidation: | |
| def test_missing_required_fields(self): | |
| """Test validation of step with missing required fields""" | |
| validator = WorkflowValidator() | |
| step = ModelStep( | |
| id="step1", | |
| name="", # Missing name | |
| model="", # Missing model | |
| provider="", # Missing provider | |
| call_type=CallType.LLM, # Missing call_type | |
| temperature=0.7, | |
| system_prompt="Test prompt", | |
| input_fields=[], | |
| output_fields=[], | |
| ) | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step1.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.STEP | |
| def test_invalid_step_id(self): | |
| """Test validation of step with invalid ID format""" | |
| validator = WorkflowValidator() | |
| step = create_basic_step("123invalid") # Invalid ID format | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step1.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.NAMING | |
| def test_llm_temperature_validation_invalid(self): | |
| """Test validation of LLM step temperature""" | |
| validator = WorkflowValidator() | |
| # Test invalid temperature | |
| step = create_basic_step() | |
| step.temperature = -0.5 # Invalid temperature | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step1.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.RANGE | |
| def test_llm_temperature_validation_missing(self): | |
| # Test missing temperature | |
| validator = WorkflowValidator() | |
| step = create_basic_step() | |
| step.temperature = None # Missing temperature | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step1.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.STEP | |
| def test_llm_system_prompt_validation(self): | |
| """Test validation of LLM step system prompt""" | |
| validator = WorkflowValidator() | |
| # Test missing system prompt | |
| step = create_basic_step() | |
| step.system_prompt = "" # Missing system prompt | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step1.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.STEP | |
| # Test too long system prompt | |
| step = create_basic_step() | |
| step.system_prompt = "x" * 4001 # Too long | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step1.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.LENGTH | |
| # Field Validation Tests | |
| class TestFieldValidation: | |
| def test_input_field_validation(self): | |
| """Test validation of input fields""" | |
| validator = WorkflowValidator() | |
| # Test missing required fields | |
| step = create_basic_step() | |
| step.input_fields = [InputField(name="", description="", variable="")] | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.STEP | |
| # Test invalid field name | |
| step = create_basic_step() | |
| step.input_fields = [InputField(name="123invalid", description="test", variable="test")] | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.NAMING | |
| # Test too long description | |
| step = create_basic_step() | |
| step.input_fields = [InputField(name="test", description="x" * 201, variable="test")] | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.LENGTH | |
| def test_output_field_validation(self): | |
| """Test validation of output fields""" | |
| validator = WorkflowValidator() | |
| # Test missing required fields | |
| step = create_basic_step() | |
| step.output_fields = [OutputField(name="", description="", type="str")] | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.STEP | |
| # Test invalid field name | |
| step = create_basic_step() | |
| step.output_fields = [OutputField(name="123invalid", description="test", type="str")] | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.NAMING | |
| def test_field_name_length(self): | |
| """Test validation of field name length""" | |
| validator = WorkflowValidator() | |
| # Test too long field name | |
| step = create_basic_step() | |
| step.input_fields = [InputField(name="x" * 51, description="test", variable="test")] | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.LENGTH | |
| def test_field_description_length(self): | |
| """Test validation of field description length""" | |
| validator = WorkflowValidator() | |
| # Test too long description | |
| step = create_basic_step() | |
| step.input_fields = [InputField(name="test", description="x" * 201, variable="test")] | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.LENGTH | |
| def test_whitespace_only_strings(self): | |
| """Test validation of whitespace-only strings""" | |
| validator = WorkflowValidator() | |
| # Test whitespace-only field name | |
| step = create_basic_step() | |
| step.input_fields = [InputField(name=" ", description="test", variable="test")] | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.NAMING | |
| def test_special_characters(self): | |
| """Test validation of special characters in names""" | |
| validator = WorkflowValidator() | |
| # Test special characters in field name | |
| step = create_basic_step() | |
| step.input_fields = [InputField(name="test@field", description="test", variable="test")] | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.NAMING | |
| # Variable Reference Tests | |
| class TestVariableReference: | |
| def test_external_input_validation(self): | |
| """Test validation of external input variables""" | |
| validator = WorkflowValidator() | |
| # Test invalid external input format | |
| workflow = create_valid_workflow() | |
| workflow.inputs = ["step1.field"] # Invalid format | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.VARIABLE | |
| def test_step_output_reference(self): | |
| """Test validation of step output references""" | |
| validator = WorkflowValidator() | |
| # Test invalid output reference | |
| workflow = create_basic_workflow() | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "nonexistent_step.field"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.VARIABLE | |
| # Test valid output reference | |
| step = create_basic_step() | |
| step.output_fields = [OutputField(name="field", description="test", type="str")] | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.field"} | |
| assert validator.validate(workflow) | |
| assert len(validator.errors) == 0 | |
| # DAG Validation Tests | |
| class TestDAGValidation: | |
| def test_cycle_detection(self): | |
| """Test detection of cycles in workflow""" | |
| validator = WorkflowValidator() | |
| # Create a workflow with a cycle | |
| step1 = create_step_with_fields( | |
| "step1", | |
| [InputField(name="input", description="test", variable="step3.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| step2 = create_step_with_fields( | |
| "step2", | |
| [InputField(name="input", description="test", variable="step1.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| step3 = create_step_with_fields( | |
| "step3", | |
| [InputField(name="input", description="test", variable="step2.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| workflow = create_basic_workflow([step1, step2, step3]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step3.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.DAG | |
| def test_orphaned_steps(self): | |
| """Test detection of orphaned steps""" | |
| validator = WorkflowValidator() | |
| # Create a workflow with an orphaned step | |
| step1 = create_step_with_fields( | |
| "step1", | |
| [InputField(name="input", description="test", variable="step2.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| step2 = create_step_with_fields( | |
| "step2", | |
| [InputField(name="input", description="test", variable="step1.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| step3 = create_step_with_fields( | |
| "step3", | |
| [], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| workflow = create_basic_workflow([step1, step2, step3]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step3.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.DAG | |
| # Variable Dependency Tests | |
| class TestVariableDependencies: | |
| def test_circular_dependencies(self): | |
| """Test detection of circular variable dependencies""" | |
| validator = WorkflowValidator() | |
| # Create a workflow with circular variable dependencies | |
| step1 = create_step_with_fields( | |
| "step1", | |
| [InputField(name="input", description="test", variable="step2.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| step2 = create_step_with_fields( | |
| "step2", | |
| [InputField(name="input", description="test", variable="step1.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| workflow = create_basic_workflow([step1, step2]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step2.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.DAG | |
| def test_valid_dependencies(self): | |
| """Test validation of valid variable dependencies""" | |
| validator = WorkflowValidator() | |
| # Create a workflow with valid dependencies | |
| step1 = create_step_with_fields( | |
| "step1", | |
| [InputField(name="input", description="test", variable="external_input")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| step2 = create_step_with_fields( | |
| "step2", | |
| [InputField(name="input", description="test", variable="step1.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| step3 = create_step_with_fields( | |
| "step3", | |
| [InputField(name="input", description="test", variable="step2.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| workflow = create_basic_workflow([step1, step2, step3]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step3.output"} | |
| assert validator.validate(workflow) | |
| assert len(validator.errors) == 0 | |
| # Type Compatibility Tests | |
| class TestTypeCompatibility: | |
| def test_basic_type_compatibility(self): | |
| """Test validation of basic type compatibility""" | |
| validator = WorkflowValidator() | |
| # Create steps with type mismatch | |
| step1 = create_step_with_fields( | |
| "step1", | |
| [InputField(name="input", description="test", variable="external_input")], | |
| [OutputField(name="output", description="test", type="int")], | |
| ) | |
| step2 = create_step_with_fields( | |
| "step2", | |
| [InputField(name="input", description="test", variable="step1.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| workflow = create_basic_workflow([step1, step2]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step2.output"} | |
| assert validator.validate(workflow) | |
| # Complex Workflow Tests | |
| class TestComplexWorkflows: | |
| def test_multi_output_workflow(self): | |
| """Test validation of workflow with multiple outputs""" | |
| validator = WorkflowValidator() | |
| # Create a workflow with multiple outputs | |
| step1 = create_step_with_fields( | |
| "step1", | |
| [], | |
| [ | |
| OutputField(name="output1", description="test", type="str"), | |
| OutputField(name="output2", description="test", type="int"), | |
| ], | |
| ) | |
| step2 = create_step_with_fields( | |
| "step2", | |
| [InputField(name="input", description="test", variable="step1.output1")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| workflow = create_basic_workflow([step1, step2]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output1": "step1.output1", "output2": "step1.output2", "output3": "step2.output"} | |
| assert validator.validate(workflow) | |
| assert len(validator.errors) == 0 | |
| def test_complex_dependencies(self): | |
| """Test validation of workflow with complex dependencies""" | |
| validator = WorkflowValidator() | |
| # Create a workflow with complex dependencies | |
| step1 = create_step_with_fields( | |
| "step1", | |
| [InputField(name="input", description="test", variable="external_input")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| step2 = create_step_with_fields( | |
| "step2", | |
| [InputField(name="input", description="test", variable="step1.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| step3 = create_step_with_fields( | |
| "step3", | |
| [ | |
| InputField(name="input1", description="test", variable="step1.output"), | |
| InputField(name="input2", description="test", variable="step2.output"), | |
| ], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| workflow = create_basic_workflow([step1, step2, step3]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step3.output"} | |
| assert validator.validate(workflow) | |
| assert len(validator.errors) == 0 | |
| # Log Probability Validation Tests | |
| class TestLogProbabilityValidation: | |
| def test_logprob_step_validation(self): | |
| """Test validation of log probability step references""" | |
| validator = WorkflowValidator() | |
| # Create a workflow with multiple steps | |
| step1 = create_step_with_fields( | |
| "step1", | |
| [InputField(name="input", description="test", variable="external_input")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| step2 = create_step_with_fields( | |
| "step2", | |
| [InputField(name="input", description="test", variable="step1.output")], | |
| [OutputField(name="output", description="test", type="str")], | |
| ) | |
| workflow = create_basic_workflow([step1, step2]) | |
| workflow.inputs = ["external_input"] | |
| workflow.outputs = {"output": "step2.output"} | |
| # Validate the workflow first | |
| assert validator.validate(workflow) | |
| validator.errors = [] # Clear any previous errors | |
| # Test that a valid step ID is accepted | |
| valid_logprob_step = "step1" | |
| assert valid_logprob_step in workflow.steps | |
| # A validator for logprob_step would check if the step exists in workflow.steps | |
| # Test that an invalid step ID is caught | |
| invalid_logprob_step = "nonexistent_step" | |
| assert invalid_logprob_step not in workflow.steps | |
| # A validator for logprob_step would report an error for a non-existent step | |
| # Output Structure Tests | |
| class TestOutputStructure: | |
| def test_workflow_output_structure(self): | |
| """Test the expected structure of workflow outputs""" | |
| # Sample output dictionary matching WorkflowOutput structure | |
| output: dict[str, dict | None] = { | |
| "final_outputs": {}, | |
| "intermediate_outputs": {}, | |
| "step_contents": {}, | |
| "logprob": None, | |
| } | |
| # Verify that all expected keys are present | |
| assert "final_outputs" in output | |
| assert "intermediate_outputs" in output | |
| assert "step_contents" in output | |
| assert "logprob" in output | |
| # Test with populated values | |
| output = { | |
| "final_outputs": {"output": "result"}, | |
| "intermediate_outputs": {"step1.output": "result", "input": "value"}, | |
| "step_contents": {"step1": "Full content"}, | |
| "logprob": -2.5, | |
| } | |
| assert output["final_outputs"] == {"output": "result"} | |
| assert output["intermediate_outputs"]["step1.output"] == "result" | |
| assert output["step_contents"]["step1"] == "Full content" | |
| assert output["logprob"] == -2.5 | |
| def test_model_step_result_structure(self): | |
| """Test the expected structure of model step results""" | |
| # Sample result dictionary matching ModelStepResult structure | |
| result: dict[str, Any] = {"outputs": {}, "content": None, "logprob": None} | |
| # Verify that all expected keys are present | |
| assert "outputs" in result | |
| assert "content" in result | |
| assert "logprob" in result | |
| # Test with populated values | |
| result = {"outputs": {"field": "value"}, "content": "Full response", "logprob": -1.5} | |
| assert result["outputs"] == {"field": "value"} | |
| assert result["content"] == "Full response" | |
| assert result["logprob"] == -1.5 | |
| # External Input Tests | |
| class TestExternalInputs: | |
| def test_external_input_existence(self): | |
| """Test validation of external input existence""" | |
| validator = WorkflowValidator() | |
| # Test missing external input | |
| step = create_step_with_fields( | |
| "step1", [InputField(name="input", description="test", variable="missing_input")], [] | |
| ) | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["valid_input"] | |
| workflow.outputs = {"output": "step1.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.VARIABLE | |
| def test_external_input_naming_conflicts(self): | |
| """Test validation of external input naming conflicts""" | |
| validator = WorkflowValidator() | |
| # Test conflict between external input and step output | |
| step = create_step_with_fields("step1", [], [OutputField(name="output", description="test", type="str")]) | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["step1.output"] # Conflict with step output | |
| workflow.outputs = {"output": "step1.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.VARIABLE | |
| # Edge Cases | |
| class TestEdgeCases: | |
| def test_empty_workflow_with_inputs(self): | |
| """Test validation of empty workflow with inputs""" | |
| validator = WorkflowValidator() | |
| workflow = Workflow(inputs=["input"], outputs={}, steps={}) | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.GENERAL | |
| def test_workflow_with_empty_outputs(self): | |
| """Test validation of workflow with empty outputs""" | |
| validator = WorkflowValidator() | |
| workflow = create_valid_workflow() | |
| workflow.outputs = {} # Empty output reference | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.GENERAL | |
| def test_workflow_with_none_outputs(self): | |
| """Test validation of workflow with empty outputs""" | |
| validator = WorkflowValidator() | |
| workflow = create_valid_workflow() | |
| workflow.outputs = {"output": None} # Empty output reference | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.GENERAL | |
| def test_workflow_with_duplicate_output_names(self): | |
| """Test validation of workflow with duplicate output names""" | |
| validator = WorkflowValidator() | |
| step = create_step_with_fields( | |
| "step1", | |
| [], | |
| [ | |
| OutputField(name="output", description="test", type="str"), | |
| OutputField(name="output", description="test", type="str"), | |
| ], | |
| ) | |
| workflow = create_basic_workflow([step]) | |
| workflow.inputs = ["input"] | |
| workflow.outputs = {"output": "step1.output"} | |
| assert not validator.validate(workflow) | |
| assert len(validator.errors) == 1 | |
| assert validator.errors[0].error_type == ValidationErrorType.STEP | |
| # Extended validator tests for actual implementation | |
| class TestExtendedValidation: | |
| def test_parse_variable_reference(self): | |
| """Test the _parse_variable_reference method""" | |
| validator = WorkflowValidator() | |
| # Test external input reference | |
| step_id, field_name = _parse_variable_reference("input_var") | |
| assert step_id is None | |
| assert field_name == "input_var" | |
| # Test step output reference | |
| step_id, field_name = _parse_variable_reference("step1.output") | |
| assert step_id == "step1" | |
| assert field_name == "output" | |
| def test_is_valid_identifier(self): | |
| """Test the _is_valid_identifier method""" | |
| validator = WorkflowValidator() | |
| # Valid identifiers | |
| assert validator._is_valid_identifier("valid_name") | |
| assert validator._is_valid_identifier("ValidName") | |
| assert validator._is_valid_identifier("name123") | |
| # Invalid identifiers | |
| assert not validator._is_valid_identifier("") # Empty | |
| assert not validator._is_valid_identifier(" ") # Whitespace | |
| assert not validator._is_valid_identifier("123name") # Starts with number | |
| assert not validator._is_valid_identifier("name-with-hyphens") # Has hyphens | |
| assert not validator._is_valid_identifier("name.with.dots") # Has dots | |
| def test_is_valid_external_input(self): | |
| """Test the _is_valid_external_input method""" | |
| validator = WorkflowValidator() | |
| # Valid external inputs | |
| assert validator._is_valid_external_input("input_var") | |
| # Invalid external inputs | |
| assert not validator._is_valid_external_input("") # Empty | |
| assert not validator._is_valid_external_input("input.var") # Contains dot | |
| assert not validator._is_valid_external_input("123input") # Starts with number | |