feat: Add get_running_workflow_step_by_name()

Author: Alan Christie

tests/test_test_api_adapter.py

@@ -409,3 +409,35 @@ def test_get_workflow_steps_driving_this_step_when_2nd_step():
     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()
+    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={},
+    )
+    rwf_id = response["id"]
+    response, _ = utaa.create_running_workflow_step(
+        running_workflow_id=rwf_id, step="step-2"
+    )
+    rwfs_id = response["id"]
+
+    # Act
+    response, _ = utaa.get_running_workflow_step_by_name(
+        name="step-2", running_workflow_id=rwf_id
+    )
+
+    # Assert
+    assert response["running_workflow"]["id"] == rwf_id
+    assert response["name"] == "step-2"
+    assert response["id"] == rwfs_id

tests/wapi_adapter.py

@@ -178,6 +178,25 @@ def get_running_workflow_step(
         response["id"] = running_workflow_step_id
         return response, 0
 
+    def get_running_workflow_step_by_name(
+        self, *, name: str, running_workflow_id: str
+    ) -> dict[str, Any]:
+        UnitTestWorkflowAPIAdapter.lock.acquire()
+        with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
+            running_workflow_step = Unpickler(pickle_file).load()
+        UnitTestWorkflowAPIAdapter.lock.release()
+
+        print(f"name={name} running_workflow_id={running_workflow_id}")
+        for rwfs_id, record in running_workflow_step.items():
+            print(f"rwfs_id={rwfs_id} record={record}")
+            if record["running_workflow"]["id"] != running_workflow_id:
+                continue
+            if record["name"] == name:
+                response = record
+                response["id"] = rwfs_id
+                return response, 0
+        return {}, 0
+
     def set_running_workflow_step_variables(
         self,
         *,

workflow/workflow_abc.py

@@ -242,6 +242,37 @@ def get_running_workflow_step(
         #          "id": "r-workflow-step-00000000-0000-0000-0000-000000000001",
         #       },
 
+    @abstractmethod
+    def get_running_workflow_step_by_name(
+        self, *, name: str, running_workflow_id: str
+    ) -> tuple[dict[str, Any], int]:
+        """Get a RunningWorkflowStep Record given a step name
+        (and its RUnningWorkflow ID)"""
+        # Should return:
+        # {
+        #       "id": "r-workflow-step-00000000-0000-0000-0000-000000000001",
+        #       "name:": "step-1234",
+        #       "done": False,
+        #       "success": false,
+        #       "error_num": 0,
+        #       "error_msg": "",
+        #       "variables": {
+        #          "x": 1,
+        #          "y": 2,
+        #       },
+        #       "running_workflow": {
+        #          "id": "r-workflow-00000000-0000-0000-0000-000000000001",
+        #       },
+        # }
+        # If not present an empty dictionary should be returned.
+        #
+        # For steps that are not the first in a workflow the following field
+        # can be expected in the response: -
+        #
+        #       "prior_running_workflow_step": {
+        #          "id": "r-workflow-step-00000000-0000-0000-0000-000000000001",
+        #       },
+
     @abstractmethod
     def set_running_workflow_step_variables(
         self,