Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions src/specify_cli/workflows/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -366,6 +366,17 @@ def _validate_steps(

# Determine step type
step_type = step_config.get("type", "command")
if not isinstance(step_type, str):
# Registry keys are strings. Checking an unhashable YAML value
# (for example ``type: [shell]`` or a mapping) against the set
# below raises a raw TypeError before validation can report the
# authoring mistake. Guard every non-string shape first, matching
# the typed validation already applied to workflow and step IDs.
errors.append(
f"Step {step_id!r}: 'type' must be a string, got "
f"{type(step_type).__name__} ({step_type!r})."
)
continue
if step_type not in _get_valid_step_types():
errors.append(
f"Step {step_id!r} has invalid type {step_type!r}."
Expand Down
23 changes: 23 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -4565,6 +4565,29 @@ def test_invalid_step_type(self):
errors = validate_workflow(definition)
assert any("invalid type" in e.lower() for e in errors)

@pytest.mark.parametrize("step_type", [["shell"], {"name": "shell"}])
def test_non_string_step_type_reports_error(self, step_type):
"""Unhashable YAML values must not crash registry membership checks."""
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow

definition = WorkflowDefinition(
{
"workflow": {
"id": "test",
"name": "Test",
"version": "1.0.0",
},
"steps": [{"id": "bad", "type": step_type}],
}
)

errors = validate_workflow(definition)

assert errors == [
f"Step 'bad': 'type' must be a string, got "
f"{type(step_type).__name__} ({step_type!r})."
]

def test_nested_step_validation(self):
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow

Expand Down