fix: Better doc and launcher now handles failures

Author: Alan Christie

tests/api_adapter.py

@@ -15,6 +15,7 @@
     _JOB_DEFINITIONS: Dict[str, Any] = yaml.load(jd_file, Loader=yaml.FullLoader)
 assert _JOB_DEFINITIONS
 
+# Table UUID formats
 _INSTANCE_ID_FORMAT: str = "instance-00000000-0000-0000-0000-{id:012d}"
 _TASK_ID_FORMAT: str = "task-00000000-0000-0000-0000-{id:012d}"
 _WORKFLOW_DEFINITION_ID_FORMAT: str = "workflow-00000000-0000-0000-0000-{id:012d}"
@@ -23,6 +24,7 @@
     "r-workflow-step-00000000-0000-0000-0000-{id:012d}"
 )
 
+# Pickle files (representing each 'Table')
 _WORKFLOW_PICKLE_FILE: str = "workflow.pickle"
 _RUNNING_WORKFLOW_PICKLE_FILE: str = "running-workflow.pickle"
 _RUNNING_WORKFLOW_STEP_PICKLE_FILE: str = "running-workflow-step.pickle"
@@ -33,7 +35,11 @@
 class UnitTestAPIAdapter(APIAdapter):
     """A minimal API adapter. It serves-up Job Definitions
     from the job-definitions/job-definitions.yaml file and provides basic
-    (in-memory) storage for Workflow Definitions and related tables."""
+    storage for Workflow Definitions and related tables.
+
+    Because the adapter is used by the multi-processing test suite, it uses both a lock
+    and pickle files to store data, so that data can be shared between processes.
+    """
 
     mp_lock = Lock()
 
@@ -74,9 +80,7 @@ def get_workflow(self, *, workflow_id: str) -> Dict[str, Any]:
             workflow = Unpickler(pickle_file).load()
         UnitTestAPIAdapter.mp_lock.release()
 
-        if workflow_id not in workflow:
-            return {}
-        return {"workflow": workflow[workflow_id]}
+        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.mp_lock.acquire()
@@ -224,9 +228,7 @@ def get_instance(self, *, instance_id: str) -> Dict[str, Any]:
             instances = Unpickler(pickle_file).load()
         UnitTestAPIAdapter.mp_lock.release()
 
-        if instance_id not in instances:
-            return {}
-        return instances[instance_id]
+        return {} if instance_id not in instances else instances[instance_id]
 
     def create_task(self, *, instance_id: str) -> Dict[str, Any]:
         UnitTestAPIAdapter.mp_lock.acquire()
@@ -253,9 +255,7 @@ def get_task(self, *, task_id: str) -> Dict[str, Any]:
             tasks = Unpickler(pickle_file).load()
         UnitTestAPIAdapter.mp_lock.release()
 
-        if task_id not in tasks:
-            return {}
-        return tasks[task_id]
+        return {} if task_id not in tasks else tasks[task_id]
 
     def get_job(
         self, *, collection: str, job: str, version: str

tests/instance_launcher.py

@@ -16,8 +16,19 @@
 class UnitTestInstanceLauncher(InstanceLauncher):
     """A unit test instance launcher, which runs the
     Python module that matches the job name in the provided specification.
-    It also uses the UnitTestMessageDispatcher to send the simulated
-    'end of instance' PodMessage (to the WorkflowEngine).
+
+    The Python module used to satisfy the step matches the job name in the
+    step specification. If the step_specification's 'job' is 'my_job', then the launcher
+    will run the Python module 'my_job.py' in the 'jobs' directory. The
+    module is run synchronously - i.e. the launch() method waits for the
+    module to complete.
+
+    It then uses the UnitTestMessageDispatcher to send a simulated
+    'end of instance' PodMessage  that will be received by the WorkflowEngine's
+    'handle_message()' method. The 'exit code' of the module is passed to the
+    WorkflowEngine through the PodMessage - so if the module fails (i.e. returns
+    a non-zero exit code) then the WorkflowEngine will see that the PodMessage.
+    This allows you to write jobs that fail and see how the WorkflowEngine responds.
     """
 
     def __init__(
@@ -63,8 +74,7 @@ def launch(
 
         job_cmd: List[str] = ["python", job_module]
         print(f"Running job command: {job_cmd}")
-        completed_process: CompletedProcess = subprocess.run(job_cmd, check=True)
-        assert completed_process.returncode == 0
+        completed_process: CompletedProcess = subprocess.run(job_cmd, check=False)
 
         # Simulate a PodMessage (that will contain the instance ID),
         # filling-in only the fields that are of use to the Engine.
@@ -74,7 +84,7 @@ def launch(
         pod_message.instance = instance_id
         pod_message.task = task_id
         pod_message.has_exit_code = True
-        pod_message.exit_code = 0
+        pod_message.exit_code = completed_process.returncode
         self._msg_dispatcher.send(pod_message)
 
         return LaunchResult(