feat: API adapter now supports set_running_workflow_step_variables()

Author: Alan Christie

tests/test_test_api_adapter.py

@@ -163,6 +163,31 @@ def test_create_running_workflow_step():
     assert response["id"] == "r-workflow-step-00000000-0000-0000-0000-000000000001"
 
 
+def test_set_running_workflow_step_variables():
+    # Arrange
+    utaa = UnitTestWorkflowAPIAdapter()
+    response = utaa.create_workflow(workflow_definition={"name": "blah"})
+    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"
+    )
+    rwfsid = response["id"]
+
+    # Act
+    utaa.set_running_workflow_step_variables(
+        running_workflow_step_id=rwfsid, variables={"z": 42}
+    )
+
+    # Assert
+    response, _ = utaa.get_running_workflow_step(running_workflow_step_id=rwfsid)
+    assert response["variables"] == {"z": 42}
+
+
 def test_set_running_workflow_step_done_when_success():
     # Arrange
     utaa = UnitTestWorkflowAPIAdapter()
@@ -187,6 +212,7 @@ def test_set_running_workflow_step_done_when_success():
     assert response["success"]
     assert response["error"] is None
     assert response["error_msg"] is None
+    assert response["variables"] == {}
 
 
 def test_set_running_workflow_step_done_when_failed():

tests/wapi_adapter.py

@@ -137,6 +137,7 @@ def create_running_workflow_step(
             "name": step,
             "done": False,
             "success": False,
+            "variables": {},
             "running_workflow": {"id": running_workflow_id},
         }
         running_workflow_step[running_workflow_step_id] = record
@@ -161,18 +162,18 @@ def get_running_workflow_step(
         response["id"] = running_workflow_step_id
         return response, 0
 
-    def set_running_workflow_step_command(
+    def set_running_workflow_step_variables(
         self,
         *,
         running_workflow_step_id: str,
-        command: str,
+        variables: dict[str, Any],
     ) -> None:
         UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
             running_workflow_step = Unpickler(pickle_file).load()
 
         assert running_workflow_step_id in running_workflow_step
-        running_workflow_step[running_workflow_step_id]["command"] = command
+        running_workflow_step[running_workflow_step_id]["variables"] = variables
 
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "wb") as pickle_file:
             Pickler(pickle_file).dump(running_workflow_step)

workflow/workflow_abc.py

@@ -186,12 +186,27 @@ def get_running_workflow_step(
         #       "success": false,
         #       "error": None,
         #       "error_msg": None,
+        #       "variables": {
+        #          "x": 1,
+        #          "y": 2,
+        #       },
         #       "running_workflow": {
         #          "id": "r-workflow-00000000-0000-0000-0000-000000000001"
         #       },
         # }
         # If not present an empty dictionary should be returned.
 
+    @abstractmethod
+    def set_running_workflow_step_variables(
+        self,
+        *,
+        running_workflow_step_id: str,
+        variables: dict[str, Any],
+    ) -> None:
+        """Set the variables used prior to decoding the step command for each step.
+        This can be used to understand step failures but will also be vital
+        when adding variables values to subsequent steps from prior step values."""
+
     @abstractmethod
     def set_running_workflow_step_done(
         self,