refactor: More combiner logic

Author: Alan Christie

tests/instance_launcher.py

@@ -76,6 +76,12 @@ 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

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/wapi_adapter.py

@@ -112,6 +112,12 @@ 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, step_name: str
+    ) -> tuple[dict[str, Any], int]:
+        # Need to implement!
+        return {"count": 0, "status": []}, 0
+
     def set_running_workflow_done(
         self,
         *,
@@ -249,44 +255,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:

tests/workflow-definitions/simple-python-split-combine.yaml

@@ -36,5 +36,19 @@ steps:
     from-step:
       name: split
       variable: outputBase
+
+#- name: combine
+#  description: Add some params
+#  specification:
+#    collection: workflow-engine-unit-test-jobs
+#    job: concatenate
+#    version: "1.0.0"
+#    variables:
+#      outputFile: results.smi
+#  plumbing:
+#  - variable: inputFile
+#    from-step:
+#      variable: outputFile
+#      name: parallel
   out:
   - outputFile

workflow/workflow_abc.py

@@ -42,8 +42,12 @@ class LaunchParameters:
     # If only one instance of the step is expected to run
     # this value can be left at 0 (zero). If this step's launch
     # is expected to be executed more than once the value should be
-    # non-zero (and unique for this workflow run).
+    # 1..'N'.
     step_replication_number: int = 0
+    # The total number of replicas of this instance that are expected to be laucnhed.
+    # if step_replication_number is set, this has to be set. It is 'N'.
+    # If step_replication_number is zero this value is ignored.
+    total_number_of_replicas: int = 0
     # The application ID (a custom resource name)
     # used to identify the 'type' of Instance to create.
     # For DM Jobs this will be 'datamanagerjobs.squonk.it'
@@ -144,7 +148,7 @@ def get_running_workflow(
         #       "running_user": "alan",
         #       "running_user_api_token": "123456789",
         #       "done": False,
-        #       "success": false,
+        #       "success": False,
         #       "error_num": 0,
         #       "error_msg": "",
         #       "workflow": {
@@ -171,7 +175,29 @@ def get_running_steps(
         #    "count": 1,
         #    "steps": [
         #       {
-        #           "name:": "step-1234"
+        #           "name": "step-1234"
+        #       }
+        #    ]
+        # }
+
+    @abstractmethod
+    def get_status_of_all_step_instances_by_name(
+        self, *, running_workflow_id: str, step_name: str
+    ) -> tuple[dict[str, Any], int]:
+        """Get a list of step execution statuses for the named step."""
+        # Should return:
+        # {
+        #    "count": 2,
+        #    "status": [
+        #       {
+        #           "done": True,
+        #           "success": True,
+        #           "running_workflow_step_id": "step-0001"
+        #       },
+        #       {
+        #           "done": False,
+        #           "success": False,
+        #           "running_workflow_step_id": "step-0002"
         #       }
         #    ]
         # }
@@ -195,9 +221,9 @@ def get_running_workflow_step(
         """Get a RunningWorkflowStep Record"""
         # Should return:
         # {
-        #       "name:": "step-1234",
+        #       "name": "step-1234",
         #       "done": False,
-        #       "success": false,
+        #       "success": False,
         #       "error_num": 0,
         #       "error_msg": "",
         #       "variables": {
@@ -234,9 +260,9 @@ def get_running_workflow_step_by_name(
         # Should return:
         # {
         #       "id": "r-workflow-step-00000000-0000-0000-0000-000000000001",
-        #       "name:": "step-1234",
+        #       "name": "step-1234",
         #       "done": False,
-        #       "success": false,
+        #       "success": False,
         #       "error_num": 0,
         #       "error_msg": "",
         #       "variables": {
@@ -274,26 +300,6 @@ def set_running_workflow_step_done(
         """Set the success value for a RunningWorkflowStep Record,
         If not successful an error code and message should be provided."""
 
-    @abstractmethod
-    def get_workflow_steps_driving_this_step(
-        self,
-        *,
-        running_workflow_step_id: str,
-    ) -> tuple[dict[str, Any], int]:
-        """Get all the step records that belong to the Workflow for the given
-        RunningWorkflowStep record ID. You are also given the caller's position
-        in the list, which will be -1 if the caller is not present."""
-        # It should return:
-        # {
-        #    "caller_step_index": 0,
-        #    "steps": [
-        #      {
-        #        "name": "step-name"
-        #        "specification": "{}",
-        #       }
-        #     ]
-        # }
-
     @abstractmethod
     def get_instance(self, *, instance_id: str) -> tuple[dict[str, Any], int]:
         """Get an Instance Record"""