feat: Add get_workflow_output_values_for_step

Author: Alan Christie

tests/test_decoder.py

@@ -292,3 +292,40 @@ def test_get_workflow_inputs_for_step_with_unkown_step_name():
 
     # Assert
     assert not inputs
+
+
+def test_get_workflow_outputs_for_step_with_name_step1():
+    # Arrange
+
+    # Act
+    outputs = decoder.get_workflow_output_values_for_step(
+        _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW, "step1"
+    )
+
+    # Assert
+    assert not outputs
+
+
+def test_get_workflow_outputs_for_step_with_name_step2():
+    # Arrange
+
+    # Act
+    outputs = decoder.get_workflow_output_values_for_step(
+        _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW, "step2"
+    )
+
+    # Assert
+    assert len(outputs) == 1
+    assert "clustered-molecules.smi" in outputs
+
+
+def test_get_workflow_outputs_for_step_with_unkown_step_name():
+    # Arrange
+
+    # Act
+    outputs = decoder.get_workflow_output_values_for_step(
+        _SIMPLE_PYTHON_MOLPROPS_WITH_OPTIONS_WORKFLOW, "unknown"
+    )
+
+    # Assert
+    assert not outputs

workflow/decoder.py

@@ -91,7 +91,6 @@ def get_workflow_input_names_for_step(
 
     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:
@@ -106,6 +105,24 @@ def get_workflow_input_names_for_step(
     return inputs
 
 
+def get_workflow_output_values_for_step(
+    definition: dict[str, Any], name: str
+) -> list[str]:
+    """Given a Workflow definition and a step name we return a list of workflow
+    out variable names the step creates. To do this we iterate through the workflows's
+    outputs to find those that are declared 'from' our step."""
+    wf_outputs = definition.get("variable-mapping", {}).get("outputs", {})
+    outputs: list[str] = []
+    outputs.extend(
+        output["as"]
+        for output in wf_outputs
+        if "from" in output
+        and "step" in output["from"]
+        and output["from"]["step"] == name
+    )
+    return outputs
+
+
 def set_variables_from_options_for_step(
     definition: dict[str, Any], variables: dict[str, Any], step_name: str
 ) -> dict[str, Any]: