Skip to content
Open
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
8 changes: 8 additions & 0 deletions src/specify_cli/workflows/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -1400,6 +1400,14 @@ def _resolve_inputs(
) -> dict[str, Any]:
"""Resolve workflow inputs against definitions and provided values."""
resolved: dict[str, Any] = {}
# execute()/resume() accept UNVALIDATED definitions (load_workflow does
# not validate). A non-mapping ``inputs:`` block (bare ``inputs:`` ->
# None, or ``inputs: []``) is stored raw, so iterating ``.items()`` here
# would crash the run with AttributeError. Treat a non-mapping inputs
# block as "no inputs"; validate_workflow reports the malformed shape
# via its own isinstance check.
if not isinstance(definition.inputs, dict):
return {}
for name, input_def in definition.inputs.items():
if not isinstance(input_def, dict):
continue
Expand Down
17 changes: 17 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -3612,6 +3612,23 @@ def test_from_string(self, sample_workflow_yaml):
assert definition.id == "test-workflow"
assert len(definition.inputs) == 2

@pytest.mark.parametrize(
"block",
[
"workflow:\n id: w\n name: W\nsteps: []\ninputs: []\n", # list
"workflow:\n id: w\n name: W\nsteps: []\ninputs:\n", # null
],
)
def test_resolve_inputs_tolerates_non_mapping_inputs(self, block):
# execute()/resume() run UNVALIDATED definitions; a non-mapping `inputs:`
# block (list/null) is stored raw and would crash _resolve_inputs at
# `.items()`. It must be treated as "no inputs" instead.
from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine

definition = WorkflowDefinition.from_string(block)
resolved = WorkflowEngine()._resolve_inputs(definition, {}) # must not raise
assert resolved == {}

def test_from_string_invalid(self):
from specify_cli.workflows.engine import WorkflowDefinition

Expand Down