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: 9 additions & 1 deletion src/specify_cli/workflows/_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -601,7 +601,15 @@ def _validate_and_install_local(yaml_path: Path, source_label: str) -> None:
except (ValueError, yaml.YAMLError) as exc:
console.print(f"[red]Error:[/red] Invalid workflow YAML: {exc}")
raise typer.Exit(1)
if not definition.id or not definition.id.strip():
# Non-string ids (e.g. unquoted ``id: 123`` or ``id: 0``) fall through
# to validate_workflow below, which reports a typed error instead of
# crashing on ``.strip()`` here. Only None/empty/whitespace-only ids
# are rejected as missing.
if (
definition.id is None
or definition.id == ""
or (isinstance(definition.id, str) and not definition.id.strip())
):
console.print("[red]Error:[/red] Workflow definition has an empty or missing 'id'")
raise typer.Exit(1)

Expand Down
39 changes: 34 additions & 5 deletions src/specify_cli/workflows/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -129,26 +129,49 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
errors: list[str] = []

# -- Schema version ---------------------------------------------------
if definition.schema_version not in ("1.0", "1"):
# str() so an unquoted ``schema_version: 1.0`` (YAML float) is accepted —
# rejecting it would print "Unsupported schema_version 1.0. Expected '1.0'."
if str(definition.schema_version) != "1.0":
errors.append(
f"Unsupported schema_version {definition.schema_version!r}. "
f"Expected '1.0'."
)

# -- Top-level fields -------------------------------------------------
if not definition.id:
# YAML parses unquoted scalars like ``id: 123`` or ``version: 1.0`` as
# int/float; check types before regex/string operations so authoring
# mistakes surface as validation errors instead of tracebacks. Only
# ``None``/empty-string count as missing so falsey non-strings
# (``id: 0``, ``name: false``) still get the typed error.
if definition.id is None or definition.id == "":
errors.append("Workflow is missing 'workflow.id'.")
elif not isinstance(definition.id, str):
errors.append(
f"'workflow.id' must be a string, got "
f"{type(definition.id).__name__} ({definition.id!r})."
)
elif not _ID_PATTERN.match(definition.id):
errors.append(
f"Workflow ID {definition.id!r} must be lowercase alphanumeric "
f"with hyphens."
)

if not definition.name:
if definition.name is None or definition.name == "":
errors.append("Workflow is missing 'workflow.name'.")
elif not isinstance(definition.name, str):
errors.append(
f"'workflow.name' must be a string, got "
f"{type(definition.name).__name__} ({definition.name!r})."
)

if not definition.version:
if definition.version is None or definition.version == "":
errors.append("Workflow is missing 'workflow.version'.")
elif not isinstance(definition.version, str):
errors.append(
f"'workflow.version' must be a string, got "
f"{type(definition.version).__name__} ({definition.version!r}) — "
f'quote it in YAML (version: "1.0.0").'
)
elif not re.match(r"^\d+\.\d+\.\d+$", definition.version):
errors.append(
f"Workflow version {definition.version!r} is not valid "
Expand Down Expand Up @@ -256,9 +279,15 @@ def _validate_steps(
continue

step_id = step_config.get("id")
if not step_id:
if step_id is None or step_id == "":
errors.append("Step is missing 'id' field.")
continue
if not isinstance(step_id, str):
errors.append(
f"Step ID must be a string, got "
f"{type(step_id).__name__} ({step_id!r})."
)
continue

if ":" in step_id:
errors.append(
Expand Down
145 changes: 145 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -2627,6 +2627,101 @@ def test_invalid_id_format(self):
errors = validate_workflow(definition)
assert any("lowercase alphanumeric" in e for e in errors)

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

definition = WorkflowDefinition.from_string("""
workflow:
id: 123
name: "Test"
version: "1.0.0"
steps:
- id: step-one
command: speckit.specify
""")
errors = validate_workflow(definition)
assert any("workflow.id" in e and "string" in e for e in errors)

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

definition = WorkflowDefinition.from_string("""
workflow:
id: "test"
name: 123
version: "1.0.0"
steps:
- id: step-one
command: speckit.specify
""")
errors = validate_workflow(definition)
assert any("workflow.name" in e and "string" in e for e in errors)

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

definition = WorkflowDefinition.from_string("""
workflow:
id: "test"
name: "Test"
version: 1.0
steps:
- id: step-one
command: speckit.specify
""")
errors = validate_workflow(definition)
assert any("workflow.version" in e and "quote" in e for e in errors)

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

definition = WorkflowDefinition.from_string("""
workflow:
id: "test"
name: "Test"
version: "1.0.0"
steps:
- id: 123
command: speckit.specify
""")
errors = validate_workflow(definition)
assert any("Step ID" in e and "string" in e for e in errors)

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

definition = WorkflowDefinition.from_string("""
workflow:
id: 0
name: false
version: 0.0
steps:
- id: 0
command: speckit.specify
""")
errors = validate_workflow(definition)
assert any("'workflow.id' must be a string" in e for e in errors)
assert any("'workflow.name' must be a string" in e for e in errors)
assert any("'workflow.version' must be a string" in e for e in errors)
assert any("Step ID must be a string" in e for e in errors)
assert not any("missing" in e for e in errors)

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

definition = WorkflowDefinition.from_string("""
schema_version: 1.0
workflow:
id: "test"
name: "Test"
version: "1.0.0"
steps:
- id: step-one
command: speckit.specify
""")
errors = validate_workflow(definition)
assert errors == []

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

Expand Down Expand Up @@ -6882,3 +6977,53 @@ def test_non_gate_step_without_type_is_not_a_gate(self):
},
)
assert _gate_outcome(state) is None


class TestWorkflowAddNonStringScalars:
"""`workflow add` reports clean errors for non-string YAML scalars (#3420)."""

@pytest.mark.parametrize(
("field_yaml", "expected"),
[
('id: 123\n name: "Probe"\n version: "1.0.0"', "workflow.id"),
('id: "probe"\n name: "Probe"\n version: 1.0', "workflow.version"),
],
)
def test_add_reports_validation_error_not_traceback(
self, project_dir, monkeypatch, field_yaml, expected
):
from typer.testing import CliRunner
from specify_cli import app

monkeypatch.chdir(project_dir)
wf = project_dir / "workflow.yml"
wf.write_text(
"schema_version: \"1.0\"\n"
f"workflow:\n {field_yaml}\n"
"steps:\n - id: s1\n type: shell\n run: \"echo hi\"\n",
encoding="utf-8",
)
runner = CliRunner()
result = runner.invoke(app, ["workflow", "add", str(wf)])
assert result.exit_code == 1
assert result.exception is None or isinstance(result.exception, SystemExit)
assert expected in result.output

def test_add_non_string_step_id_reports_validation_error(
self, project_dir, monkeypatch
):
from typer.testing import CliRunner
from specify_cli import app

monkeypatch.chdir(project_dir)
wf = project_dir / "workflow.yml"
wf.write_text(
"workflow:\n id: \"probe\"\n name: \"Probe\"\n version: \"1.0.0\"\n"
"steps:\n - id: 123\n type: shell\n run: \"echo hi\"\n",
encoding="utf-8",
)
runner = CliRunner()
result = runner.invoke(app, ["workflow", "add", str(wf)])
assert result.exit_code == 1
assert result.exception is None or isinstance(result.exception, SystemExit)
assert "Step ID" in result.output