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
10 changes: 10 additions & 0 deletions src/specify_cli/workflows/steps/shell/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,16 @@ def validate(self, config: dict[str, Any]) -> list[str]:
errors.append(
f"Shell step {config.get('id', '?')!r} is missing 'run' field."
)
elif not isinstance(config["run"], str):
# execute() str()-coerces run and invokes it under shell=True, so a
# null or list 'run' would run the Python repr ('None', "['echo']")
# as a command. Reject non-strings at validation, mirroring the
# command-step input/options and gate options type checks. An
# expression like "{{ ... }}" is still a str, so it stays valid.
errors.append(
f"Shell step {config.get('id', '?')!r}: 'run' must be a string, "
f"got {type(config['run']).__name__}."
)
output_format = config.get("output_format")
if output_format is not None and output_format != "json":
errors.append(
Expand Down
20 changes: 20 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -1183,6 +1183,26 @@ def test_validate_missing_run(self):
errors = step.validate({"id": "test"})
assert any("missing 'run'" in e for e in errors)

@pytest.mark.parametrize("bad_run", [None, ["echo", "hi"], 42])
def test_validate_rejects_non_string_run(self, bad_run):
"""A non-string 'run' must be rejected at validation.

execute() str()-coerces run and invokes it under shell=True, so a
null or list run would otherwise run the Python repr as a command.
"""
from specify_cli.workflows.steps.shell import ShellStep

step = ShellStep()
errors = step.validate({"id": "s", "run": bad_run})
assert any("'run' must be a string" in e for e in errors)

def test_validate_accepts_string_and_expression_run(self):
from specify_cli.workflows.steps.shell import ShellStep

step = ShellStep()
assert step.validate({"id": "s", "run": "echo hi"}) == []
assert step.validate({"id": "s", "run": "{{ steps.x.output }}"}) == []


def test_output_format_json_exposes_data(self, tmp_path):
from specify_cli.workflows.steps.shell import ShellStep
Expand Down