tests/test_validators.py

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