feat: Add get_workflow_steps_driving_this_step and prior step ID to create_running_workflow_step

Author: Alan Christie

tests/test_test_api_adapter.py

@@ -267,6 +267,37 @@ def test_get_running_workflow_step():
     assert response["name"] == "step-1"
     assert not response["done"]
     assert response["running_workflow"]["id"] == rwfid
+    assert "prior_running_workflow_step" not in response
+
+
+def test_get_running_workflow_step_with_prior_step():
+    # Arrange
+    utaa = UnitTestWorkflowAPIAdapter()
+    response = utaa.create_workflow(workflow_definition={"name": "blah"})
+    wfid = response["id"]
+    response = utaa.create_running_workflow(
+        user_id="dlister",
+        workflow_id=wfid,
+        project_id=TEST_PROJECT_ID,
+        variables={},
+    )
+    rwfid = response["id"]
+    response, _ = utaa.create_running_workflow_step(
+        running_workflow_id=rwfid,
+        step="step-1",
+        prior_running_workflow_step_id="r-workflow-step-111",
+    )
+    rwfsid = response["id"]
+
+    # Act
+    response, _ = utaa.get_running_workflow_step(running_workflow_step_id=rwfsid)
+
+    # Assert
+    assert response["name"] == "step-1"
+    assert not response["done"]
+    assert response["running_workflow"]["id"] == rwfid
+    assert "prior_running_workflow_step" in response
+    assert response["prior_running_workflow_step"]["id"] == "r-workflow-step-111"
 
 
 def test_create_instance():
@@ -291,3 +322,69 @@ def test_create_and_get_instance():
 
     # Assert
     assert response["running_workflow_step_id"] == "r-workflow-step-000"
+
+
+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"

tests/wapi_adapter.py

@@ -88,14 +88,16 @@ def get_workflow(self, *, workflow_id: str) -> dict[str, Any]:
             response["id"] = workflow_id
         return response, 0
 
-    def get_running_workflow(self, *, running_workflow_id: str) -> dict[str, Any]:
+    def get_running_workflow(
+        self, *, running_workflow_id: str
+    ) -> tuple[dict[str, Any], int]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
             running_workflow = Unpickler(pickle_file).load()
         UnitTestWorkflowAPIAdapter.lock.release()
 
         if running_workflow_id not in running_workflow:
-            return {}
+            return {}, 0
         response = running_workflow[running_workflow_id]
         response["id"] = running_workflow_id
         return response, 0
@@ -123,7 +125,11 @@ def set_running_workflow_done(
         UnitTestWorkflowAPIAdapter.lock.release()
 
     def create_running_workflow_step(
-        self, *, running_workflow_id: str, step: str
+        self,
+        *,
+        running_workflow_id: str,
+        step: str,
+        prior_running_workflow_step_id: str | None = None,
     ) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
@@ -140,6 +146,10 @@ def create_running_workflow_step(
             "variables": {},
             "running_workflow": {"id": running_workflow_id},
         }
+        if prior_running_workflow_step_id:
+            record["prior_running_workflow_step"] = {
+                "id": prior_running_workflow_step_id
+            }
         running_workflow_step[running_workflow_step_id] = record
 
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "wb") as pickle_file:
@@ -201,6 +211,44 @@ 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) -> dict[str, Any]:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_INSTANCE_PICKLE_FILE, "rb") as pickle_file:

workflow/workflow_abc.py

@@ -167,6 +167,7 @@ def create_running_workflow_step(
         *,
         running_workflow_id: str,
         step: str,
+        prior_running_workflow_step_id: str | None = None,
     ) -> tuple[dict[str, Any], int]:
         """Create a RunningWorkflowStep Record (from a RunningWorkflow)"""
         # Should return:
@@ -191,10 +192,17 @@ def get_running_workflow_step(
         #          "y": 2,
         #       },
         #       "running_workflow": {
-        #          "id": "r-workflow-00000000-0000-0000-0000-000000000001"
+        #          "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-worflkow-step-00000000-0000-0000-0000-000000000001",
+        #       },
 
     @abstractmethod
     def set_running_workflow_step_variables(
@@ -219,6 +227,26 @@ 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": "steps": [
+        #      {
+        #        "name": "step-name"
+        #        "specification": "{}",
+        #       }
+        #     ]
+        # }
+
     @abstractmethod
     def get_instance(self, *, instance_id: str) -> tuple[dict[str, Any], int]:
         """Get an Instance Record"""