fix: Now handles optional variables (and ability to set step command)

Author: Alan Christie

tests/wapi_adapter.py

@@ -161,6 +161,23 @@ def get_running_workflow_step(
         response["id"] = running_workflow_step_id
         return response, 0
 
+    def set_running_workflow_step_command(
+        self,
+        *,
+        running_workflow_step_id: str,
+        command: str,
+    ) -> 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
+
+        with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "wb") as pickle_file:
+            Pickler(pickle_file).dump(running_workflow_step)
+        UnitTestWorkflowAPIAdapter.lock.release()
+
     def set_running_workflow_step_done(
         self,
         *,

workflow/workflow_abc.py

@@ -156,7 +156,7 @@ def get_running_workflow_step(
         """Get a RunningWorkflowStep Record"""
         # Should return:
         # {
-        #       "step:": "step-1234",
+        #       "name:": "step-1234",
         #       "done": False,
         #       "success": false,
         #       "error": None,
@@ -167,6 +167,15 @@ def get_running_workflow_step(
         # }
         # If not present an empty dictionary should be returned.
 
+    @abstractmethod
+    def set_running_workflow_step_command(
+        self,
+        *,
+        running_workflow_step_id: str,
+        command: str,
+    ) -> None:
+        """Set the command value for a RunningWorkflowStep Record"""
+
     @abstractmethod
     def set_running_workflow_step_done(
         self,

workflow/workflow_engine.py

@@ -134,17 +134,20 @@ def _handle_workflow_start_message(self, r_wfid: str) -> None:
         r_wfsid = response["id"]
 
         # The step's 'specification' is a string - pass it directly to the
-        # launcher along with any appropriate 'variables'. The launcher
+        # launcher along with any (optional) 'variables'. The launcher
         # will apply the variables to step's Job command but we need to handle
         # any launch problems. The validator should have checked to ensure that
         # variable expansion will work, but we must prepare for the unexpected.
 
         project_id = rwf_response["project"]["id"]
-        variables = rwf_response["variables"]
+        variables: dict[str, Any] | None = rwf_response.get("variables")
 
         _LOGGER.info(
-            "RunningWorkflow: %s (name=%s project=%s, variables=%s)",
+            "Launching first step: RunningWorkflow=%s RunningWorkflowStep=%s step=%s"
+            " (name=%s project=%s, variables=%s)",
             r_wfid,
+            r_wfsid,
+            first_step_name,
             rwf_response["name"],
             project_id,
             variables,
@@ -167,6 +170,11 @@ def _handle_workflow_start_message(self, r_wfid: str) -> None:
                 first_step_name, r_wfid, r_wfsid, lr.error_num, lr.error_msg
             )
         else:
+            if lr.command:
+                self._wapi_adapter.set_running_workflow_step_command(
+                    running_workflow_step_id=r_wfsid,
+                    command=lr.command,
+                )
             _LOGGER.info(
                 "Launched first step '%s' (command=%s)", first_step_name, lr.command
             )