From 81564e545b03a8b03fe0a623978d82a2d79e1340 Mon Sep 17 00:00:00 2001
From: Shubham Padkonde <shubhampadkonde12@gmail.com>
Date: Tue, 29 Sep 2026 20:11:25 +0530
Subject: [PATCH] fix: validate workflow steps after environment substitution

---
 databusclient/workflow/parser.py |  9 +++---
 tests/test_workflow_parser.py    | 47 +++++++++++++++++++++++++++++++-
 2 files changed, 51 insertions(+), 5 deletions(-)

diff --git a/databusclient/workflow/parser.py b/databusclient/workflow/parser.py
index 267dda1..a694427 100644
--- a/databusclient/workflow/parser.py
+++ b/databusclient/workflow/parser.py
@@ -178,11 +178,12 @@ def parse_workflow(path: str) -> Dict[str, Any]:
     seen_names: set = set()
     validated_steps: List[Dict[str, Any]] = []
     for index, step in enumerate(steps):
-        validated = _validate_step(step, index, seen_names)
-        substituted = _substitute_value(validated, validated["name"])
-        validated_steps.append(substituted)
+        step_name = step.get("name", str(index)) if isinstance(step, dict) else str(index)
+        substituted = _substitute_value(step, step_name)
+        validated = _validate_step(substituted, index, seen_names)
+        validated_steps.append(validated)
 
     return {
         "manifest": raw.get("manifest"),
         "steps": validated_steps,
-    }
\ No newline at end of file
+    }
diff --git a/tests/test_workflow_parser.py b/tests/test_workflow_parser.py
index b82bab7..f2078aa 100644
--- a/tests/test_workflow_parser.py
+++ b/tests/test_workflow_parser.py
@@ -177,4 +177,49 @@ def test_empty_braces_pass_through_unchanged():
         "steps": [{"name": "a", "command": "download", "uri": "${}/data"}]
     })
     result = parse_workflow(path)
-    assert result["steps"][0]["uri"] == "${}/data"
\ No newline at end of file
+    assert result["steps"][0]["uri"] == "${}/data"
+
+
+def test_duplicate_names_after_substitution_raise(monkeypatch):
+    monkeypatch.setenv("STEP_NAME", "fetch")
+    path = _write_yaml({
+        "steps": [
+            {"name": "fetch", "command": "download", "uri": "x"},
+            {"name": "${STEP_NAME}", "command": "download", "uri": "y"},
+        ]
+    })
+    with pytest.raises(WorkflowParseError, match="Duplicate step name 'fetch'"):
+        parse_workflow(path)
+
+
+def test_empty_name_after_substitution_raises(monkeypatch):
+    monkeypatch.setenv("STEP_NAME", "")
+    path = _write_yaml({
+        "steps": [{"name": "${STEP_NAME}", "command": "download", "uri": "x"}]
+    })
+    with pytest.raises(WorkflowParseError, match="valid 'name'"):
+        parse_workflow(path)
+
+
+@pytest.mark.parametrize("command", ["download", "invalid"])
+def test_command_is_validated_after_substitution(monkeypatch, command):
+    monkeypatch.setenv("COMMAND", command)
+    path = _write_yaml({
+        "steps": [{"name": "fetch", "command": "${COMMAND}", "uri": "x"}]
+    })
+    if command == "download":
+        assert parse_workflow(path)["steps"][0]["command"] == "download"
+    else:
+        with pytest.raises(WorkflowParseError, match="invalid command 'invalid'"):
+            parse_workflow(path)
+
+
+def test_on_error_is_validated_after_substitution(monkeypatch):
+    monkeypatch.setenv("ERROR_POLICY", "continue")
+    path = _write_yaml({
+        "steps": [{
+            "name": "fetch", "command": "download", "uri": "x",
+            "on_error": "${ERROR_POLICY}",
+        }]
+    })
+    assert parse_workflow(path)["steps"][0]["on_error"] == "continue"