feat: Launcher now creates PodMessage

Author: Alan Christie

tests/instance_launcher.py

@@ -1,7 +1,8 @@
 import os
 import subprocess
+from datetime import datetime, timezone
 from subprocess import CompletedProcess
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Dict, List
 
 from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
 
@@ -32,15 +33,27 @@ def launch(
         *,
         project_id: str,
         workflow_id: str,
+        running_workflow_step_id: str,
         workflow_definition: Dict[str, Any],
         step: str,
         step_specification: Dict[str, Any],
-        completion_callback: Optional[Callable[[PodMessage], None]],
     ) -> LaunchResult:
         assert project_id
         assert workflow_id
         assert step_specification
 
+        # We're passed a RunningWorkflowStep ID but a record is expected to have been
+        # created bt the caller, we simply create instance records.
+        response = self._api_adapter.get_running_workflow_step(
+            running_workflow_step_id=running_workflow_step_id
+        )
+        assert "running_workflow_step" in response
+        # Now simulate the creation of an Instance record
+        response = self._api_adapter.create_instance(
+            running_workflow_step_id=running_workflow_step_id
+        )
+        instance_id = response["instance_id"]
+
         # Just run the Python module that matched the 'job' in the step specification.
         # Don't care about 'version' or 'collection'.
         job: str = step_specification["job"]
@@ -52,10 +65,20 @@ def launch(
         completed_process: CompletedProcess = subprocess.run(job_cmd, check=True)
         assert completed_process.returncode == 0
 
+        # Simulate a PodMessage (that will contain the instance ID),
+        # filling-in only the fields that are of use to the Engine.
+        pod_message = PodMessage()
+        pod_message.timestamp = f"{datetime.now(timezone.utc).isoformat()}Z"
+        pod_message.phase = "Completed"
+        pod_message.instance = instance_id
+        pod_message.has_exit_code = True
+        pod_message.exit_code = 0
+        self._msg_dispatcher.send(pod_message)
+
         return LaunchResult(
             error=0,
             error_msg=None,
-            instance_id="instance-00000000-0000-0000-0000-000000000000",
+            instance_id=instance_id,
             task_id="task-00000000-0000-0000-0000-000000000000",
             command=" ".join(job_cmd),
         )

tests/test_test_instance_launcher.py

@@ -15,22 +15,31 @@ def basic_launcher():
     api_adapter = UnitTestAPIAdapter()
     message_queue = UnitTestMessageQueue()
     message_dispatcher = UnitTestMessageDispatcher(msg_queue=message_queue)
-    return UnitTestInstanceLauncher(
+    instance_launcher = UnitTestInstanceLauncher(
         api_adapter=api_adapter, msg_dispatcher=message_dispatcher
     )
+    return [api_adapter, instance_launcher]
 
 
 def test_get_nop_job(basic_launcher):
     # Arrange
+    utaa = basic_launcher[0]
+    launcher = basic_launcher[1]
+    response = utaa.create_workflow(workflow_definition={"name": "blah"})
+    response = utaa.create_running_workflow(workflow_definition_id=response["id"])
+    response = utaa.create_running_workflow_step(
+        running_workflow_id=response["id"], step="step-1"
+    )
+    rwfsid = response["id"]
 
     # Act
-    result = basic_launcher.launch(
+    result = launcher.launch(
         project_id="project-00000000-0000-0000-0000-000000000000",
         workflow_id="workflow-00000000-0000-0000-0000-000000000000",
+        running_workflow_step_id=rwfsid,
         workflow_definition={},
         step="step-1",
         step_specification={"job": "nop"},
-        completion_callback=None,
     )
 
     # Assert

workflow/workflow_abc.py

@@ -4,10 +4,9 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 from google.protobuf.message import Message
-from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
 
 
 @dataclass
@@ -33,7 +32,6 @@ def launch(
         workflow_definition: Dict[str, Any],
         step: str,
         step_specification: Dict[str, Any],
-        completion_callback: Optional[Callable[[PodMessage], None]],
     ) -> LaunchResult:
         """Launch a (Job) Instance"""
 
@@ -48,11 +46,6 @@ def launch(
         # step_specification by the WE? Remember that we have to deal with
         # "input Handlers" that manipulate the specification variables.
         # See _instance_preamble() in the DM's api_instance.py module.
-        #
-        # The completion_callback is only used in local testing and is a function
-        # that should be able to process a PodMessage that indicates a workflow Job
-        # has completed. When the WorkflowEngine is embedded in the data Manager
-        # the Data Manager will not permit the use of this parameter.
 
 
 class APIAdapter(ABC):