feat: Initial schema for replicate step declaration

Author: Alan Christie

tests/test_decoder.py

@@ -56,6 +56,15 @@
     )
 assert _DUPLICATE_WORKFLOW_VARIABLE_NAMES_WORKFLOW
 
+_SIMPLE_PYTHON_PARALLEL_FILE: str = os.path.join(
+    os.path.dirname(__file__),
+    "workflow-definitions",
+    "simple-python-parallel.yaml",
+)
+with open(_SIMPLE_PYTHON_PARALLEL_FILE, "r", encoding="utf8") as workflow_file:
+    _SIMPLE_PYTHON_PARALLEL_WORKFLOW: Dict[str, Any] = yaml.safe_load(workflow_file)
+assert _SIMPLE_PYTHON_PARALLEL_WORKFLOW
+
 _STEP_SPECIFICATION_VARIABLE_NAMES_WORKFLOW_FILE: str = os.path.join(
     os.path.dirname(__file__),
     "workflow-definitions",
@@ -165,6 +174,16 @@ def test_validate_schema_for_workflow_options():
     assert error is None
 
 
+def test_validate_schema_for_simple_python_parallel():
+    # Arrange
+
+    # Act
+    error = decoder.validate_schema(_SIMPLE_PYTHON_PARALLEL_WORKFLOW)
+
+    # Assert
+    assert error is None
+
+
 def test_get_workflow_variables_for_smiple_python_molprops():
     # Arrange
 
@@ -329,3 +348,36 @@ def test_get_workflow_outputs_for_step_with_unkown_step_name():
 
     # Assert
     assert not outputs
+
+
+def test_get_step_input_variable_names_when_duplicates():
+    # Arrange
+
+    # Act
+    inputs = decoder.get_step_input_variable_names(
+        _SIMPLE_PYTHON_PARALLEL_WORKFLOW, "final-step"
+    )
+
+    # Assert
+    assert len(inputs) == 2
+    assert inputs[0] == "inputFile"
+    assert inputs[1] == "inputFile"
+
+
+def test_get_step_output_variable_names_when_duplicates():
+    # Arrange
+    workflow_filename: str = os.path.join(
+        os.path.dirname(__file__),
+        "workflow-definitions",
+        "duplicate-step-output-variable-names.yaml",
+    )
+    with open(workflow_filename, "r", encoding="utf8") as wf_file:
+        definition: Dict[str, Any] = yaml.safe_load(wf_file)
+
+    # Act
+    outputs = decoder.get_step_output_variable_names(definition, "step-1")
+
+    # Assert
+    assert len(outputs) == 2
+    assert outputs[0] == "outputFile"
+    assert outputs[1] == "outputFile"

tests/test_workflow_validator_for_run_level.py

@@ -233,3 +233,72 @@ def test_validate_duplicate_workflow_variable_names():
     # Assert
     assert error.error_num == 6
     assert error.error_msg == ["Duplicate workflow variable names found: x"]
+
+
+def test_validate_simple_python_parallel():
+    # Arrange
+    workflow_file: str = os.path.join(
+        os.path.dirname(__file__),
+        "workflow-definitions",
+        "simple-python-parallel.yaml",
+    )
+    with open(workflow_file, "r", encoding="utf8") as workflow_file:
+        workflow: dict[str, Any] = yaml.load(workflow_file, Loader=yaml.FullLoader)
+    assert workflow
+
+    # Act
+    error = WorkflowValidator.validate(
+        level=ValidationLevel.TAG,
+        workflow_definition=workflow,
+    )
+
+    # Assert
+    assert error.error_num == 0
+
+
+def test_validate_replicate_using_undeclared_input():
+    # Arrange
+    workflow_file: str = os.path.join(
+        os.path.dirname(__file__),
+        "workflow-definitions",
+        "replicate-using-undeclared-input.yaml",
+    )
+    with open(workflow_file, "r", encoding="utf8") as workflow_file:
+        workflow: dict[str, Any] = yaml.load(workflow_file, Loader=yaml.FullLoader)
+    assert workflow
+
+    # Act
+    error = WorkflowValidator.validate(
+        level=ValidationLevel.TAG,
+        workflow_definition=workflow,
+    )
+
+    # Assert
+    assert error.error_num == 7
+    assert error.error_msg == [
+        "Replicate input variable is not declared: y (step=step-2)"
+    ]
+
+
+def test_validate_duplicate_step_output_variable_names():
+    # Arrange
+    workflow_file: str = os.path.join(
+        os.path.dirname(__file__),
+        "workflow-definitions",
+        "duplicate-step-output-variable-names.yaml",
+    )
+    with open(workflow_file, "r", encoding="utf8") as workflow_file:
+        workflow: dict[str, Any] = yaml.load(workflow_file, Loader=yaml.FullLoader)
+    assert workflow
+
+    # Act
+    error = WorkflowValidator.validate(
+        level=ValidationLevel.TAG,
+        workflow_definition=workflow,
+    )
+
+    # Assert
+    assert error.error_num == 3
+    assert error.error_msg == [
+        "Duplicate step output variable: outputFile (step=step-1)"
+    ]

tests/test_workflow_validator_for_tag_level.py

@@ -109,6 +109,27 @@ def test_validate_shortcut_example_1():
     assert error.error_msg is None
 
 
+def test_validate_simple_python_parallel():
+    # Arrange
+    workflow_file: str = os.path.join(
+        os.path.dirname(__file__),
+        "workflow-definitions",
+        "simple-python-parallel.yaml",
+    )
+    with open(workflow_file, "r", encoding="utf8") as workflow_file:
+        workflow: dict[str, Any] = yaml.load(workflow_file, Loader=yaml.FullLoader)
+    assert workflow
+
+    # Act
+    error = WorkflowValidator.validate(
+        level=ValidationLevel.TAG,
+        workflow_definition=workflow,
+    )
+
+    # Assert
+    assert error.error_num == 0
+
+
 def test_validate_simple_python_molprops():
     # Arrange
     workflow_file: str = os.path.join(
@@ -171,3 +192,51 @@ def test_validate_duplicate_workflow_variable_names():
     # Assert
     assert error.error_num == 6
     assert error.error_msg == ["Duplicate workflow variable names found: x"]
+
+
+def test_validate_replicate_using_undeclared_input():
+    # Arrange
+    workflow_file: str = os.path.join(
+        os.path.dirname(__file__),
+        "workflow-definitions",
+        "replicate-using-undeclared-input.yaml",
+    )
+    with open(workflow_file, "r", encoding="utf8") as workflow_file:
+        workflow: dict[str, Any] = yaml.load(workflow_file, Loader=yaml.FullLoader)
+    assert workflow
+
+    # Act
+    error = WorkflowValidator.validate(
+        level=ValidationLevel.TAG,
+        workflow_definition=workflow,
+    )
+
+    # Assert
+    assert error.error_num == 7
+    assert error.error_msg == [
+        "Replicate input variable is not declared: y (step=step-2)"
+    ]
+
+
+def test_validate_duplicate_step_output_variable_names():
+    # Arrange
+    workflow_file: str = os.path.join(
+        os.path.dirname(__file__),
+        "workflow-definitions",
+        "duplicate-step-output-variable-names.yaml",
+    )
+    with open(workflow_file, "r", encoding="utf8") as workflow_file:
+        workflow: dict[str, Any] = yaml.load(workflow_file, Loader=yaml.FullLoader)
+    assert workflow
+
+    # Act
+    error = WorkflowValidator.validate(
+        level=ValidationLevel.TAG,
+        workflow_definition=workflow,
+    )
+
+    # Assert
+    assert error.error_num == 3
+    assert error.error_msg == [
+        "Duplicate step output variable: outputFile (step=step-1)"
+    ]

tests/workflow-definitions/duplicate-step-output-variable-names.yaml

@@ -0,0 +1,56 @@
+---
+kind: DataManagerWorkflow
+kind-version: "2025.2"
+name: duplicate-step-output-variable-names
+description: A workflow where step-1 has duplicate output variable names
+variable-mapping:
+  inputs:
+  - name: x
+  outputs:
+  - name: y
+    from:
+      step: step-2
+      output: outputFile
+    as: clustered-molecules.smi
+
+steps:
+
+- name: step-1
+  description: Add column 1
+  specification:
+    collection: workflow-engine-unit-test-jobs
+    job: rdkit-molprops
+    version: "1.0.0"
+    variables:
+      name: "col1"
+      value: 123
+  inputs:
+  - input: inputFile
+    from:
+      workflow-input: candidateMolecules
+  - input: inputFile
+    from:
+      workflow-input: candidateMolecules
+  outputs:
+  - output: outputFile
+    as: __step1__out.smi
+  - output: outputFile
+    as: __step1__out.smi
+
+- name: step-2
+  description: Add column 2
+  specification:
+    collection: workflow-engine-unit-test-jobs
+    job: cluster-butina
+    version: "1.0.0"
+    variables:
+      name: "col2"
+      value: "999"
+  inputs:
+  - input: inputFile
+    from:
+      step: step1
+      output: outputFile
+  outputs:
+  - output: outputFile
+    as: __step2__out.smi

tests/workflow-definitions/duplicate-workflow-variable-names.yaml

@@ -6,7 +6,6 @@ description: A workflow with a duplicate variable name in the input and output
 variable-mapping:
   inputs:
   - name: x
-    type: squonk/x-smiles
   outputs:
   - name: x
     from:

tests/workflow-definitions/replicate-using-undeclared-input.yaml

@@ -0,0 +1,54 @@
+---
+kind: DataManagerWorkflow
+kind-version: "2025.2"
+name: replicate-using-undeclared-input
+description: A workflow that replicates from a variable that's not declared
+variable-mapping:
+  inputs:
+  - name: x
+  outputs:
+  - name: y
+    from:
+      step: step2
+      output: outputFile
+    as: clustered-molecules.smi
+
+steps:
+
+- name: step-1
+  description: Add column 1
+  specification:
+    collection: workflow-engine-unit-test-jobs
+    job: rdkit-molprops
+    version: "1.0.0"
+    variables:
+      name: "col1"
+      value: 123
+  inputs:
+  - input: inputFile
+    from:
+      workflow-input: candidateMolecules
+  outputs:
+  - output: outputFile
+    as: __step-1__out.smi
+
+- name: step-2
+  description: Add column 2
+  specification:
+    collection: workflow-engine-unit-test-jobs
+    job: cluster-butina
+    version: "1.0.0"
+    variables:
+      name: "col2"
+      value: "999"
+  replicate:
+    using:
+      input: y
+  inputs:
+  - input: inputFile
+    from:
+      step: step-1
+      output: outputFile
+  outputs:
+  - output: outputFile
+    as: __step-2__out.smi

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

@@ -56,7 +56,6 @@ variables:
 variable-mapping:
   inputs:
   - name: candidateMolecules
-    type: squonk/x-smiles
   outputs:
   - name: clusteredMolecules
     from:

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

@@ -6,7 +6,6 @@ description: A simple python experimental workflow
 variable-mapping:
   inputs:
   - name: candidateMolecules
-    type: squonk/x-smiles
   outputs:
   - name: clusteredMolecules
     from:

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

@@ -6,7 +6,6 @@ description: A simple branching workflow
 variable-mapping:
   inputs:
   - name: candidateMolecules
-    type: squonk/x-smiles
   outputs:
   - name: clusteredMolecules
     from: