feat: Add get_workflow_input_names_for_step to decoder

Author: Alan Christie

tests/test_decoder.py

@@ -255,3 +255,40 @@ def test_set_variables_from_options_for_step_for_simnple_python_molprops_with_op
     assert "value" in new_variables
     assert new_variables["name"] == "propertyName"
     assert new_variables["value"] == "propertyValue"
+
+
+def test_get_workflow_inputs_for_step_with_name_step1():
+    # Arrange
+
+    # Act
+    inputs = decoder.get_workflow_input_names_for_step(
+        _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW, "step1"
+    )
+
+    # Assert
+    assert len(inputs) == 1
+    assert "candidateMolecules" in inputs
+
+
+def test_get_workflow_inputs_for_step_with_name_step2():
+    # Arrange
+
+    # Act
+    inputs = decoder.get_workflow_input_names_for_step(
+        _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW, "step2"
+    )
+
+    # Assert
+    assert not inputs
+
+
+def test_get_workflow_inputs_for_step_with_unkown_step_name():
+    # Arrange
+
+    # Act
+    inputs = decoder.get_workflow_input_names_for_step(
+        _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW, "unknown"
+    )
+
+    # Assert
+    assert not inputs

workflow/decoder.py

@@ -82,6 +82,30 @@ def get_variable_names(definition: dict[str, Any]) -> list[str]:
     return wf_variable_names
 
 
+def get_workflow_input_names_for_step(
+    definition: dict[str, Any], name: str
+) -> list[str]:
+    """Given a Workflow definition and a step name we return a list of workflow
+    input variable names the step expects. To do this we iterate through the step's
+    inputs to find those that are declared 'from->workflow-input'.
+
+    To get the input (a filename) the caller simply looks these names up
+    in the variable map."""
+    print(f"definition={definition}")
+    inputs: list[str] = []
+    for step in definition.get("steps", {}):
+        if step["name"] == name and "inputs" in step:
+            # Find all the workflow inputs.
+            # This gives us the name of the workflow input variable
+            # and the name of the step input (Job) variable.
+            inputs.extend(
+                step_input["from"]["workflow-input"]
+                for step_input in step["inputs"]
+                if "from" in step_input and "workflow-input" in step_input["from"]
+            )
+    return inputs
+
+
 def set_variables_from_options_for_step(
     definition: dict[str, Any], variables: dict[str, Any], step_name: str
 ) -> dict[str, Any]: