Merge pull request #38 from InformaticsMatters/concat

Initial combiner mechanism

Author: alanbchristie

tests/instance_launcher.py

@@ -76,13 +76,20 @@ def launch(self, *, launch_parameters: LaunchParameters) -> LaunchResult:
 
         os.makedirs(EXECUTION_DIRECTORY, exist_ok=True)
 
+        if launch_parameters.step_replication_number:
+            assert (
+                launch_parameters.step_replication_number
+                <= launch_parameters.total_number_of_replicas
+            )
+
         # Create a running workflow step
         assert launch_parameters.running_workflow_id
         assert launch_parameters.step_name
         response, _ = self._api_adapter.create_running_workflow_step(
             running_workflow_id=launch_parameters.running_workflow_id,
             step=launch_parameters.step_name,
             replica=launch_parameters.step_replication_number,
+            replicas=launch_parameters.total_number_of_replicas,
         )
         assert "id" in response
         rwfs_id: str = response["id"]

tests/job-definitions/job-definitions.yaml

@@ -131,13 +131,34 @@ jobs:
 
   concatenate:
     command: >-
-      concatenate.py {% for ifile in  inputFile %}{{ ifile }} {% endfor %} --outputFile {{ outputFile }}
+      concatenate.py --inputFile {{ inputFile }} --outputFile {{ outputFile }}
+    # Simulate a multiple input files Job (combiner)...
+    variables:
+      inputs:
+        properties:
+          inputFile:
+            type: files
+      options:
+        type: object
+        properties:
+          inputDirPrefix:
+            title: Optional inoput directory prefix
+            type: string
+      outputs:
+        properties:
+          outputBase:
+            creates: '{{ outputFile }}'
+            type: file
 
   splitsmiles:
     command: >-
       copyf.py {{ inputFile }}
-    # Simulate multiple output files...
+    # Simulate a multiple output files Job (splitetr)...
     variables:
+      inputs:
+        properties:
+          inputFile:
+            type: file
       outputs:
         properties:
           outputBase:

tests/jobs/concatenate.py

@@ -2,13 +2,16 @@
 
 parser = argparse.ArgumentParser(
     prog="addcol",
-    description="Takes a list of files and writes them into single outputfile",
+    description="Takes an optional directory prefix and a file,"
+    " and combines all the input files that are found"
+    " into single outputfile",
 )
-parser.add_argument("inputFile", nargs="+", type=argparse.FileType("r"))
+parser.add_argument("--inputDirPrefix")
+parser.add_argument("--inputFile", required=True)
 parser.add_argument("-o", "--outputFile", required=True)
 args = parser.parse_args()
 
 
 with open(args.outputFile, "wt", encoding="utf8") as ofile:
-    for f in args.inputFile:
-        ofile.write(f.read())
+    with open(args.inputFile, "rt", encoding="utf8") as ifile:
+        ofile.write(ifile.read())

tests/test_decoder.py

@@ -184,41 +184,3 @@ def test_get_workflow_steps():
     assert len(steps) == 2
     assert steps[0]["name"] == "step1"
     assert steps[1]["name"] == "step2"
-
-
-def test_get_step_input_variable_names_when_duplicates():
-    # Arrange
-    workflow_filename: str = os.path.join(
-        os.path.dirname(__file__),
-        "workflow-definitions",
-        "duplicate-step-input-output-variable-names.yaml",
-    )
-    with open(workflow_filename, "r", encoding="utf8") as wf_file:
-        definition: Dict[str, Any] = yaml.safe_load(wf_file)
-
-    # Act
-    inputs = decoder.get_step_input_variable_names(definition, "step-1")
-
-    # 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-input-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-2")
-
-    # Assert
-    assert len(outputs) == 2
-    assert outputs[0] == "outputFile"
-    assert outputs[1] == "outputFile"

tests/test_test_wapi_adapter.py

@@ -421,72 +421,6 @@ def test_create_instance_and_get_step_instance_directory_by_name():
     assert response["instance_directory"] == f".{i_id}"
 
 
-def test_get_workflow_steps_driving_this_step_when_1st_step():
-    # Arrange
-    utaa = UnitTestWorkflowAPIAdapter()
-    response = utaa.create_workflow(
-        workflow_definition={
-            "name": "blah",
-            "steps": [{"name": "step-1"}, {"name": "step-2"}, {"name": "step-3"}],
-        }
-    )
-    response = utaa.create_running_workflow(
-        user_id="dlister",
-        workflow_id=response["id"],
-        project_id=TEST_PROJECT_ID,
-        variables={},
-    )
-    response, _ = utaa.create_running_workflow_step(
-        running_workflow_id=response["id"], step="step-1"
-    )
-    rwfs_id = response["id"]
-
-    # Act
-    response, _ = utaa.get_workflow_steps_driving_this_step(
-        running_workflow_step_id=rwfs_id
-    )
-
-    # Assert
-    assert response["caller_step_index"] == 0
-    assert len(response["steps"]) == 3
-    assert response["steps"][0]["name"] == "step-1"
-    assert response["steps"][1]["name"] == "step-2"
-    assert response["steps"][2]["name"] == "step-3"
-
-
-def test_get_workflow_steps_driving_this_step_when_2nd_step():
-    # Arrange
-    utaa = UnitTestWorkflowAPIAdapter()
-    response = utaa.create_workflow(
-        workflow_definition={
-            "name": "blah",
-            "steps": [{"name": "step-1"}, {"name": "step-2"}, {"name": "step-3"}],
-        }
-    )
-    response = utaa.create_running_workflow(
-        user_id="dlister",
-        workflow_id=response["id"],
-        project_id=TEST_PROJECT_ID,
-        variables={},
-    )
-    response, _ = utaa.create_running_workflow_step(
-        running_workflow_id=response["id"], step="step-2"
-    )
-    rwfs_id = response["id"]
-
-    # Act
-    response, _ = utaa.get_workflow_steps_driving_this_step(
-        running_workflow_step_id=rwfs_id
-    )
-
-    # Assert
-    assert response["caller_step_index"] == 1
-    assert len(response["steps"]) == 3
-    assert response["steps"][0]["name"] == "step-1"
-    assert response["steps"][1]["name"] == "step-2"
-    assert response["steps"][2]["name"] == "step-3"
-
-
 def test_get_running_workflow_step_by_name():
     # Arrange
     utaa = UnitTestWorkflowAPIAdapter()

tests/test_workflow_engine_examples.py

@@ -398,12 +398,12 @@ def test_workflow_engine_simple_python_molprops_with_options(basic_engine):
     assert project_file_exists(output_file_2)
 
 
-def test_workflow_engine_simple_python_fanout(basic_engine):
+def test_workflow_engine_simple_python_split_combine(basic_engine):
     # Arrange
     md, da = basic_engine
 
     da.mock_get_running_workflow_step_output_values_for_output(
-        step_name="first-step",
+        step_name="split",
         output_variable="outputBase",
         output=["chunk_1.smi", "chunk_2.smi"],
     )
@@ -427,8 +427,8 @@ def test_workflow_engine_simple_python_fanout(basic_engine):
     r_wfid = start_workflow(
         md,
         da,
-        "simple-python-fanout",
-        {"candidateMolecules": input_file_1},
+        "simple-python-split-combine",
+        {"candidateMolecules": input_file_1, "combination": "combination.smi"},
     )
 
     # Assert
@@ -439,10 +439,8 @@ def test_workflow_engine_simple_python_fanout(basic_engine):
     print("response")
     pprint(response)
 
-    assert response["count"] == 3
-    assert response["running_workflow_steps"][0]["done"]
-    assert response["running_workflow_steps"][0]["success"]
-    assert response["running_workflow_steps"][1]["done"]
-    assert response["running_workflow_steps"][1]["success"]
-    assert response["running_workflow_steps"][2]["done"]
-    assert response["running_workflow_steps"][2]["success"]
+    assert response["count"] == 4
+    rwf_steps = response["running_workflow_steps"]
+    for rwf_step in rwf_steps:
+        assert rwf_step["done"]
+        assert rwf_step["success"]

tests/test_workflow_validator_for_run_level.py

@@ -215,27 +215,3 @@ def test_validate_simple_python_molprops_with_missing_input():
     assert error.error_msg == [
         "Missing workflow variable values for: candidateMolecules"
     ]
-
-
-def test_validate_duplicate_step_output_variable_names():
-    # Arrange
-    workflow_filename: str = os.path.join(
-        os.path.dirname(__file__),
-        "workflow-definitions",
-        "duplicate-step-input-output-variable-names.yaml",
-    )
-    with open(workflow_filename, "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-2)"
-    ]

tests/test_workflow_validator_for_tag_level.py

@@ -149,27 +149,3 @@ def test_validate_simple_python_molprops_with_options():
     # Assert
     assert error.error_num == 0
     assert error.error_msg is None
-
-
-def test_validate_duplicate_step_output_variable_names():
-    # Arrange
-    workflow_filename: str = os.path.join(
-        os.path.dirname(__file__),
-        "workflow-definitions",
-        "duplicate-step-input-output-variable-names.yaml",
-    )
-    with open(workflow_filename, "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-2)"
-    ]

tests/wapi_adapter.py

@@ -112,6 +112,26 @@ def get_running_steps(
         # Does nothing at the moment - this is used for the STOP logic.
         return {"count": 0, "steps": []}, 0
 
+    def get_status_of_all_step_instances_by_name(
+        self, *, running_workflow_id: str, name: str
+    ) -> tuple[dict[str, Any], int]:
+        UnitTestWorkflowAPIAdapter.lock.acquire()
+        with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
+            running_workflow_step = Unpickler(pickle_file).load()
+        UnitTestWorkflowAPIAdapter.lock.release()
+
+        steps: list[dict[str, Any]] = []
+        for rwfs_id, record in running_workflow_step.items():
+            if record["running_workflow"]["id"] != running_workflow_id:
+                continue
+            if record["name"] == name:
+                response = record
+                response["id"] = rwfs_id
+                if record["replica"] == 0:
+                    _ = response.pop("replica")
+                steps.append(response)
+        return {"count": len(steps), "status": steps}, 0
+
     def set_running_workflow_done(
         self,
         *,
@@ -140,10 +160,11 @@ def create_running_workflow_step(
         running_workflow_id: str,
         step: str,
         replica: int = 0,
+        replicas: int = 1,
         prior_running_workflow_step_id: str | None = None,
     ) -> tuple[dict[str, Any], int]:
-        if replica:
-            assert replica > 0
+        assert replica >= 0
+        assert replicas > replica
 
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
@@ -158,6 +179,7 @@ def create_running_workflow_step(
             "done": False,
             "success": False,
             "replica": replica,
+            "replicas": replicas,
             "variables": {},
             "running_workflow": {"id": running_workflow_id},
         }
@@ -249,44 +271,6 @@ def set_running_workflow_step_done(
             Pickler(pickle_file).dump(running_workflow_step)
         UnitTestWorkflowAPIAdapter.lock.release()
 
-    def get_workflow_steps_driving_this_step(
-        self,
-        *,
-        running_workflow_step_id: str,
-    ) -> tuple[dict[str, Any], int]:
-        # To accomplish this we get the running workflow for the step,
-        # then the workflow, then the steps from that workflow.
-        # We return a dictionary and an HTTP response code.
-        UnitTestWorkflowAPIAdapter.lock.acquire()
-        with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
-            running_workflow_step = Unpickler(pickle_file).load()
-        UnitTestWorkflowAPIAdapter.lock.release()
-
-        assert running_workflow_step_id in running_workflow_step
-
-        running_workflow_id: str = running_workflow_step[running_workflow_step_id][
-            "running_workflow"
-        ]["id"]
-        rwf_response, _ = self.get_running_workflow(
-            running_workflow_id=running_workflow_id
-        )
-        assert rwf_response
-        workflow_id: str = rwf_response["workflow"]["id"]
-        wf_response, _ = self.get_workflow(workflow_id=workflow_id)
-        assert wf_response
-        # Find the caller's python in the step sequence (-1 if not found)
-        caller_step_index: int = -1
-        index: int = 0
-        for step in wf_response["steps"]:
-            if step["name"] == running_workflow_step[running_workflow_step_id]["name"]:
-                caller_step_index = index
-                break
-            index += 1
-        return {
-            "caller_step_index": caller_step_index,
-            "steps": wf_response["steps"].copy(),
-        }, 0
-
     def get_instance(self, *, instance_id: str) -> tuple[dict[str, Any], int]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_INSTANCE_PICKLE_FILE, "rb") as pickle_file: