feat: Refactor design (to include message dispatcher)

Author: Alan Christie

tests/database_adapter.py

@@ -29,12 +29,9 @@ def get_workflow_by_name(
     def get_job(
         self, *, collection: str, job: str, version: str
     ) -> Optional[Dict[str, Any]]:
-        if collection != _JOB_DEFINITIONS["collection"]:
-            return 1
-        if job not in _JOB_DEFINITIONS["jobs"]:
-            return 2
-        if version != _JOB_DEFINITIONS["jobs"][job]["version"]:
-            return 3
+        assert collection == _JOB_DEFINITIONS["collection"]
+        assert job in _JOB_DEFINITIONS["jobs"]
+
         jd = _JOB_DEFINITIONS["jobs"][job]
         response = {"command": jd["command"]}
         if "variables" in jd:

tests/instance_launcher.py

@@ -0,0 +1,50 @@
+import os
+import subprocess
+from subprocess import CompletedProcess
+from typing import Any, Callable, Dict, List, Optional
+
+from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
+
+from workflow.workflow_abc import InstanceLauncher, LaunchResult
+
+_JOB_DIRECTORY: str = os.path.join(os.path.dirname(__file__), "jobs")
+
+_SUCCESS_LAUNCH_RESULT: LaunchResult = LaunchResult(
+    error=0,
+    error_msg=None,
+    instance_id="instance-00000000-0000-0000-0000-000000000000",
+    task_id="task-00000000-0000-0000-0000-000000000000",
+)
+
+
+class UnitTestInstanceLauncher(InstanceLauncher):
+    """A unit test instance launcher, which runs the
+    Python module that matches the job name in the provided specification.
+    """
+
+    def launch(
+        self,
+        *,
+        project_id: str,
+        workflow_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
+
+        # 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"]
+        job_module = f"{_JOB_DIRECTORY}/{job}.py"
+        assert os.path.isfile(job_module)
+
+        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
+
+        return _SUCCESS_LAUNCH_RESULT

tests/job-definitions/job-definitions.yaml

@@ -10,14 +10,17 @@
 # All jobs used for unit testing must be defined in this file.
 #
 # For each job you must provide: -
-# - Job version (and only one version for each Job)
 # - Job command
+#
+# The command is meaningless as the job that is run is expected to be
+# a Python 3 module called <job>.py in the tests/jobs directory.
+# So a nop job would have a corresponding Python module called nop.py.
+# The test instance_launcher module will run the job.
 
 collection: workflow-engine-unit-test-jobs
 
 jobs:
 
   nop:
-    version: "1.0.0"
     command: >-
       python --version

tests/jobs/nop.py

@@ -0,0 +1,10 @@
+from google.protobuf.message import Message
+
+from workflow.workflow_abc import MessageDispatcher
+
+
+class UnitTestMessageDispatcher(MessageDispatcher):
+    """A minimal Message dispatcher to support testing."""
+
+    def send(self, message: Message) -> None:
+        assert message

tests/message_dispatcher.py

@@ -1,9 +1,6 @@
 # Tests for the decoder package.
-import os
-from typing import Any, Dict
 
 import pytest
-import yaml
 
 pytestmark = pytest.mark.unit
 

tests/test_database_adapter.py renamed to tests/test_test_database_adapter.py

@@ -0,0 +1,25 @@
+# Tests for the decoder package.
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+from tests.instance_launcher import UnitTestInstanceLauncher
+
+
+def test_get_nop_job():
+    # Arrange
+    util = UnitTestInstanceLauncher()
+
+    # Act
+    result = util.launch(
+        project_id="project-00000000-0000-0000-0000-000000000000",
+        workflow_id="workflow-00000000-0000-0000-0000-000000000000",
+        workflow_definition={},
+        step="step-1",
+        step_specification={"job": "nop"},
+        completion_callback=None,
+    )
+
+    # Assert
+    assert result.error == 0

tests/test_test_instance_launcher.py

@@ -0,0 +1,17 @@
+# Tests for the decoder package.
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+from tests.message_dispatcher import UnitTestMessageDispatcher
+
+
+def test_get_nop_job():
+    # Arrange
+    utmd = UnitTestMessageDispatcher()
+
+    # Act
+    utmd.send(1)
+
+    # Assert

tests/test_test_message_dispatcher.py

@@ -5,14 +5,16 @@
 pytestmark = pytest.mark.unit
 
 from tests.database_adapter import UnitTestDatabaseAdapter
+from tests.message_dispatcher import UnitTestMessageDispatcher
 from tests.test_decoder_minimal import _MINIMAL_WORKFLOW
 from workflow.worklfow_validator import ValidationLevel, WorkflowValidator
 
 
 def test_validate_minimal_for_create():
     # Arrange
     db_adapter = UnitTestDatabaseAdapter()
-    validator = WorkflowValidator(db_adapter=db_adapter)
+    msg_dispatcher = UnitTestMessageDispatcher()
+    validator = WorkflowValidator(db_adapter=db_adapter, msg_dispatcher=msg_dispatcher)
 
     # Act
     error = validator.validate(

tests/test_worflow_validator.py

@@ -4,7 +4,10 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Any, Dict, Optional
+from typing import Any, Callable, Dict, Optional
+
+from google.protobuf.message import Message
+from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
 
 
 @dataclass
@@ -29,6 +32,7 @@ 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"""
 
@@ -43,6 +47,11 @@ 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 DatabaseAdapter(ABC):
@@ -79,3 +88,11 @@ def get_job(
         version: str,
     ) -> Optional[Dict[str, Any]]:
         """Get a Job"""
+
+
+class MessageDispatcher(ABC):
+    """The class handling the sending of messages (on the Data Manager message bus)."""
+
+    @abstractmethod
+    def send(self, message: Message) -> None:
+        """Send a message"""