feat: Decoder now returns workflow variable names

Author: Alan Christie

tests/test_decoder.py

@@ -23,6 +23,13 @@
     _SHORTCUT_EXAMPLE_1_WORKFLOW: Dict[str, Any] = yaml.safe_load(workflow_file)
 assert _SHORTCUT_EXAMPLE_1_WORKFLOW
 
+_SIMPLE_PYTHON_MOLPROPS_WORKFLOW_FILE: str = os.path.join(
+    os.path.dirname(__file__), "workflow-definitions", "simple-python-molprops.yaml"
+)
+with open(_SIMPLE_PYTHON_MOLPROPS_WORKFLOW_FILE, "r", encoding="utf8") as workflow_file:
+    _SIMPLE_PYTHON_MOLPROPS_WORKFLOW: Dict[str, Any] = yaml.safe_load(workflow_file)
+assert _SIMPLE_PYTHON_MOLPROPS_WORKFLOW
+
 
 def test_validate_minimal():
     # Arrange
@@ -78,3 +85,56 @@ def test_validate_shortcut_example_1():
 
     # Assert
     assert error is None
+
+
+def test_validate_python_simple_molprops():
+    # Arrange
+
+    # Act
+    error = decoder.validate_schema(_SIMPLE_PYTHON_MOLPROPS_WORKFLOW)
+
+    # Assert
+    assert error is None
+
+
+def test_get_workflow_variables():
+    # Arrange
+
+    # Act
+    wf_variables = decoder.get_variable_names(_SIMPLE_PYTHON_MOLPROPS_WORKFLOW)
+
+    # Assert
+    assert len(wf_variables) == 1
+    assert "candidateMolecules" in wf_variables
+
+
+def test_get_workflow_description():
+    # Arrange
+
+    # Act
+    description = decoder.get_description(_SIMPLE_PYTHON_MOLPROPS_WORKFLOW)
+
+    # Assert
+    assert description == "A simple python experimental workflow"
+
+
+def test_get_workflow_name():
+    # Arrange
+
+    # Act
+    name = decoder.get_name(_SIMPLE_PYTHON_MOLPROPS_WORKFLOW)
+
+    # Assert
+    assert name == "python-workflow"
+
+
+def test_get_workflow_steps():
+    # Arrange
+
+    # Act
+    steps = decoder.get_steps(_SIMPLE_PYTHON_MOLPROPS_WORKFLOW)
+
+    # Assert
+    assert len(steps) == 2
+    assert steps[0]["name"] == "step1"
+    assert steps[1]["name"] == "step2"

tests/workflow-definitions/simple-python-molprops.yaml

@@ -2,11 +2,11 @@
 kind: DataManagerWorkflow
 kind-version: "2024.1"
 name: python-workflow
-description: An simple python experimental workflow
+description: A simple python experimental workflow
 variables:
   inputs:
   - name: candidateMolecules
-  - type: squonk/x-smiles
+    type: squonk/x-smiles
   outputs:
   - name: clusteredMolecules
     from:

workflow/decoder.py

@@ -52,11 +52,24 @@ def get_steps(definition: dict[str, Any]) -> list[dict[str, Any]]:
     return response
 
 
-def get_workflow_name(definition: dict[str, Any]) -> str:
+def get_name(definition: dict[str, Any]) -> str:
     """Given a Workflow definition this function returns its name."""
     return str(definition.get("name", ""))
 
 
-def get_workflow_description(definition: dict[str, Any]) -> str | None:
+def get_description(definition: dict[str, Any]) -> str | None:
     """Given a Workflow definition this function returns its description (if it has one)."""
     return definition.get("description")
+
+
+def get_variable_names(definition: dict[str, Any]) -> list[str]:
+    """Given a Workflow definition this function returns all the names of the
+    variables defined at the workflow level."""
+    wf_variable_names: set[str] = set()
+    variables: dict[str, Any] | None = definition.get("variables")
+    if variables:
+        for input_variable in variables.get("inputs", []):
+            name: str = input_variable["name"]
+            assert name not in wf_variable_names
+            wf_variable_names.add(name)
+    return list(wf_variable_names)