fix: Add get_status_of_all_step_instances_by_name implementation (and fix step replicas)

Author: Alan Christie

tests/instance_launcher.py

@@ -89,6 +89,7 @@ def launch(self, *, launch_parameters: LaunchParameters) -> LaunchResult:
             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/wapi_adapter.py

@@ -113,10 +113,24 @@ def get_running_steps(
         return {"count": 0, "steps": []}, 0
 
     def get_status_of_all_step_instances_by_name(
-        self, *, running_workflow_id: str, step_name: str
+        self, *, running_workflow_id: str, name: str
     ) -> tuple[dict[str, Any], int]:
-        # Need to implement!
-        return {"count": 0, "status": []}, 0
+        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,
@@ -146,6 +160,7 @@ def create_running_workflow_step(
         running_workflow_id: str,
         step: str,
         replica: int = 0,
+        replicas: int = 0,
         prior_running_workflow_step_id: str | None = None,
     ) -> tuple[dict[str, Any], int]:
         if replica:
@@ -164,6 +179,7 @@ def create_running_workflow_step(
             "done": False,
             "success": False,
             "replica": replica,
+            "replicas": replicas,
             "variables": {},
             "running_workflow": {"id": running_workflow_id},
         }

workflow/workflow_abc.py

@@ -182,7 +182,7 @@ def get_running_steps(
 
     @abstractmethod
     def get_status_of_all_step_instances_by_name(
-        self, *, running_workflow_id: str, step_name: str
+        self, *, name: str, running_workflow_id: str
     ) -> tuple[dict[str, Any], int]:
         """Get a list of step execution statuses for the named step."""
         # Should return:
@@ -226,6 +226,8 @@ def get_running_workflow_step(
         #       "success": False,
         #       "error_num": 0,
         #       "error_msg": "",
+        #       "replica": 0,
+        #       "replicas": 0,
         #       "variables": {
         #          "x": 1,
         #          "y": 2,