refactor: Major 'launcher' refactor based on needs of exiting DM launcher logic

Author: Alan Christie

tests/api_adapter.py

@@ -52,7 +52,7 @@
 _TASK_PICKLE_FILE: str = f"{_PICKLE_DIRECTORY}/task.pickle"
 
 
-class UnitTestAPIAdapter(WorkflowAPIAdapter):
+class UnitTestWorkflowAPIAdapter(WorkflowAPIAdapter):
     """A minimal API adapter. It serves-up Job Definitions
     from the job-definitions/job-definitions.yaml file and provides basic
     storage for Workflow Definitions and related tables.
@@ -66,7 +66,7 @@ class UnitTestAPIAdapter(WorkflowAPIAdapter):
     def __init__(self):
         super().__init__()
         # Safely initialise the pickle files
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         if not os.path.exists(_PICKLE_DIRECTORY):
             os.makedirs(_PICKLE_DIRECTORY)
         for file in [
@@ -78,10 +78,10 @@ def __init__(self):
         ]:
             with open(file, "wb") as pickle_file:
                 Pickler(pickle_file).dump({})
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
     def create_workflow(self, *, workflow_definition: Dict[str, Any]) -> str:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
             workflow = Unpickler(pickle_file).load()
 
@@ -91,23 +91,23 @@ def create_workflow(self, *, workflow_definition: Dict[str, Any]) -> str:
 
         with open(_WORKFLOW_PICKLE_FILE, "wb") as pickle_file:
             Pickler(pickle_file).dump(workflow)
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         return {"id": workflow_definition_id}
 
     def get_workflow(self, *, workflow_id: str) -> Dict[str, Any]:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
             workflow = Unpickler(pickle_file).load()
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         return {"workflow": workflow[workflow_id]} if workflow_id in workflow else {}
 
     def get_workflow_by_name(self, *, name: str, version: str) -> Dict[str, Any]:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
             workflow = Unpickler(pickle_file).load()
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         item = {}
         for wfid, value in workflow.items():
@@ -126,7 +126,7 @@ def create_running_workflow(
         assert user_id
         assert isinstance(variables, dict)
 
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
             running_workflow = Unpickler(pickle_file).load()
 
@@ -144,7 +144,7 @@ def create_running_workflow(
 
         with open(_RUNNING_WORKFLOW_PICKLE_FILE, "wb") as pickle_file:
             Pickler(pickle_file).dump(running_workflow)
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         return {"id": running_workflow_id}
 
@@ -156,7 +156,7 @@ def set_running_workflow_done(
         error: Optional[int] = None,
         error_msg: Optional[str] = None,
     ) -> None:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
             running_workflow = Unpickler(pickle_file).load()
 
@@ -168,13 +168,13 @@ def set_running_workflow_done(
 
         with open(_RUNNING_WORKFLOW_PICKLE_FILE, "wb") as pickle_file:
             Pickler(pickle_file).dump(running_workflow)
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
     def get_running_workflow(self, *, running_workflow_id: str) -> Dict[str, Any]:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_PICKLE_FILE, "rb") as pickle_file:
             running_workflow = Unpickler(pickle_file).load()
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         if running_workflow_id not in running_workflow:
             return {}
@@ -183,7 +183,7 @@ def get_running_workflow(self, *, running_workflow_id: str) -> Dict[str, Any]:
     def create_running_workflow_step(
         self, *, running_workflow_id: str, step: str
     ) -> str:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
             running_workflow_step = Unpickler(pickle_file).load()
 
@@ -201,17 +201,17 @@ def create_running_workflow_step(
 
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "wb") as pickle_file:
             Pickler(pickle_file).dump(running_workflow_step)
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         return {"id": running_workflow_step_id}
 
     def get_running_workflow_step(
         self, *, running_workflow_step_id: str
     ) -> Dict[str, Any]:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
             running_workflow_step = Unpickler(pickle_file).load()
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         if running_workflow_step_id not in running_workflow_step:
             return {}
@@ -227,7 +227,7 @@ def set_running_workflow_step_done(
         error: Optional[int] = None,
         error_msg: Optional[str] = None,
     ) -> None:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
             running_workflow_step = Unpickler(pickle_file).load()
 
@@ -239,15 +239,15 @@ def set_running_workflow_step_done(
 
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "wb") as pickle_file:
             Pickler(pickle_file).dump(running_workflow_step)
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
     def get_running_workflow_steps(
         self, *, running_workflow_id: str
     ) -> List[Dict[str, Any]]:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_RUNNING_WORKFLOW_STEP_PICKLE_FILE, "rb") as pickle_file:
             running_workflow_step = Unpickler(pickle_file).load()
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         steps = []
         for key, value in running_workflow_step.items():
@@ -257,7 +257,7 @@ def get_running_workflow_steps(
         return {"count": len(steps), "running_workflow_steps": steps}
 
     def create_instance(self, *, running_workflow_step_id: str) -> Dict[str, Any]:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_INSTANCE_PICKLE_FILE, "rb") as pickle_file:
             instances = Unpickler(pickle_file).load()
 
@@ -270,20 +270,20 @@ def create_instance(self, *, running_workflow_step_id: str) -> Dict[str, Any]:
 
         with open(_INSTANCE_PICKLE_FILE, "wb") as pickle_file:
             Pickler(pickle_file).dump(instances)
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         return {"id": instance_id}
 
     def get_instance(self, *, instance_id: str) -> Dict[str, Any]:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_INSTANCE_PICKLE_FILE, "rb") as pickle_file:
             instances = Unpickler(pickle_file).load()
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         return {} if instance_id not in instances else instances[instance_id]
 
     def create_task(self, *, instance_id: str) -> Dict[str, Any]:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_TASK_PICKLE_FILE, "rb") as pickle_file:
             tasks = Unpickler(pickle_file).load()
 
@@ -297,15 +297,15 @@ def create_task(self, *, instance_id: str) -> Dict[str, Any]:
 
         with open(_TASK_PICKLE_FILE, "wb") as pickle_file:
             Pickler(pickle_file).dump(tasks)
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         return {"id": task_id}
 
     def get_task(self, *, task_id: str) -> Dict[str, Any]:
-        UnitTestAPIAdapter.lock.acquire()
+        UnitTestWorkflowAPIAdapter.lock.acquire()
         with open(_TASK_PICKLE_FILE, "rb") as pickle_file:
             tasks = Unpickler(pickle_file).load()
-        UnitTestAPIAdapter.lock.release()
+        UnitTestWorkflowAPIAdapter.lock.release()
 
         return {} if task_id not in tasks else tasks[task_id]
 

tests/instance_launcher.py

@@ -15,22 +15,21 @@
 (so the start of each test begins with an empty project directory).
 """
 
-import json
 import os
 import shutil
 import subprocess
 from datetime import datetime, timezone
 from subprocess import CompletedProcess
-from typing import Any, Dict, List
+from typing import List
 
 from decoder import decoder as job_decoder
 from decoder.decoder import TextEncoding
 from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
 
-from tests.api_adapter import UnitTestAPIAdapter
+from tests.api_adapter import UnitTestWorkflowAPIAdapter
 from tests.config import TEST_PROJECT_ID
 from tests.message_dispatcher import UnitTestMessageDispatcher
-from workflow.workflow_abc import InstanceLauncher, LaunchResult
+from workflow.workflow_abc import InstanceLauncher, LaunchParameters, LaunchResult
 
 # Full path to the 'jobs' directory
 _JOB_PATH: str = os.path.join(os.path.dirname(__file__), "jobs")
@@ -48,65 +47,55 @@ class UnitTestInstanceLauncher(InstanceLauncher):
     """A unit test instance launcher."""
 
     def __init__(
-        self, api_adapter: UnitTestAPIAdapter, msg_dispatcher: UnitTestMessageDispatcher
+        self,
+        wapi_adapter: UnitTestWorkflowAPIAdapter,
+        msg_dispatcher: UnitTestMessageDispatcher,
     ):
         super().__init__()
 
-        self._api_adapter = api_adapter
+        self._api_adapter = wapi_adapter
         self._msg_dispatcher = msg_dispatcher
 
         # Every launcher starts with an empty execution directory...
         print(f"Removing execution directory ({_EXECUTION_DIRECTORY})")
         assert _EXECUTION_DIRECTORY.startswith("tests/project-root")
         shutil.rmtree(_EXECUTION_DIRECTORY, ignore_errors=True)
 
-    def launch(
-        self,
-        *,
-        project_id: str,
-        running_workflow_id: str,
-        running_workflow_step_id: str,
-        step_specification: str,
-        variables: Dict[str, Any],
-    ) -> LaunchResult:
-        assert project_id
-        assert running_workflow_id
-        assert step_specification
-        assert isinstance(step_specification, str)
-        assert isinstance(variables, dict)
-
-        assert project_id == TEST_PROJECT_ID
+    def launch(self, launch_parameters: LaunchParameters) -> LaunchResult:
+        assert launch_parameters
+        assert launch_parameters.project_id == TEST_PROJECT_ID
+        assert launch_parameters.specification
+        assert isinstance(launch_parameters.specification, dict)
 
         os.makedirs(_EXECUTION_DIRECTORY, exist_ok=True)
 
         # 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
+            running_workflow_step_id=launch_parameters.running_workflow_step_id
         )
         assert "running_workflow_step" in response
         # Now simulate the creation of a Task and Instance record
         response = self._api_adapter.create_instance(
-            running_workflow_step_id=running_workflow_step_id
+            running_workflow_step_id=launch_parameters.running_workflow_step_id
         )
         instance_id = response["id"]
         response = self._api_adapter.create_task(instance_id=instance_id)
         task_id = response["id"]
 
         # Apply variables to the step's Job command.
-        step_specification_map = json.loads(step_specification)
         job = self._api_adapter.get_job(
-            collection=step_specification_map["collection"],
-            job=step_specification_map["job"],
+            collection=launch_parameters.specification["collection"],
+            job=launch_parameters.specification["job"],
             version="do-not-care",
         )
         assert job
 
         # Now apply the variables to the command
         decoded_command, status = job_decoder.decode(
             job["command"],
-            variables,
-            running_workflow_step_id,
+            launch_parameters.specification_variables,
+            launch_parameters.running_workflow_step_id,
             TextEncoding.JINJA2_3_0,
         )
         print(f"Decoded command: {decoded_command}")