feat: Use of test message queue

Author: Alan Christie

tests/message_dispatcher.py

@@ -1,10 +1,15 @@
 from google.protobuf.message import Message
 
+from tests.message_queue import UnitTestMessageQueue
 from workflow.workflow_abc import MessageDispatcher
 
 
 class UnitTestMessageDispatcher(MessageDispatcher):
     """A minimal Message dispatcher to support testing."""
 
+    def __init__(self, msg_queue: UnitTestMessageQueue):
+        super().__init__()
+        self._msg_queue: UnitTestMessageQueue = msg_queue
+
     def send(self, message: Message) -> None:
-        assert message
+        self._msg_queue.put(message)

tests/message_queue.py

@@ -1,7 +1,7 @@
 from contextlib import suppress
 from multiprocessing import Event, Process, Queue
 from queue import Empty
-from typing import Callable
+from typing import Callable, Optional
 
 from google.protobuf.message import Message
 from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
@@ -12,7 +12,7 @@ class UnitTestMessageQueue(Process):
     """A simple asynchronous message passer, used by the Validator
     (and UnitTestInstanceLauncher) to send ProtocolBuffer messages to the Engine."""
 
-    def __init__(self, receiver: Callable[[Message], None]):
+    def __init__(self, receiver: Optional[Callable[[Message], None]] = None):
         super().__init__()
         self._stop = Event()
         self._queue = Queue()
@@ -23,16 +23,17 @@ def run(self):
             with suppress(Empty):
                 if item := self._queue.get(True, 0.25):
                     msg = None
-                    # We only support Workflow and Pod messages
-                    # during testing...
+                    # Convert the message (bytes) back to a ProtocolBuffer object.
+                    # We only support Workflow and Pod messages for testing...
                     if item["class"] == "WorkflowMessage":
                         msg = WorkflowMessage()
                         msg.ParseFromString(item["bytes"])
                     elif item["class"] == "PodMessage":
                         msg = PodMessage()
                         msg.ParseFromString(item["bytes"])
                     assert msg
-                    self._receiver(msg)
+                    if self._receiver:
+                        self._receiver(msg)
 
     def put(self, msg: Message):
         """Puts a protocol buffer message onto the queue."""

tests/test_test_message_dispatcher.py

@@ -1,17 +1,27 @@
 # Tests for the decoder package.
 
+from datetime import datetime, timezone
+
 import pytest
 
 pytestmark = pytest.mark.unit
 
+from informaticsmatters.protobuf.datamanager.workflow_message_pb2 import WorkflowMessage
+
 from tests.message_dispatcher import UnitTestMessageDispatcher
+from tests.message_queue import UnitTestMessageQueue
 
 
 def test_get_nop_job():
     # Arrange
-    utmd = UnitTestMessageDispatcher()
+    utmq = UnitTestMessageQueue()
+    utmd = UnitTestMessageDispatcher(msg_queue=utmq)
+    msg = WorkflowMessage()
+    msg.timestamp = f"{datetime.now(timezone.utc).isoformat()}Z"
+    msg.action = "START"
+    msg.running_workflow = "r-workflow-00000000-0000-0000-0000-000000000000"
 
     # Act
-    utmd.send(1)
+    utmd.send(msg)
 
     # Assert

tests/test_worflow_validator.py

@@ -6,14 +6,16 @@
 
 from tests.database_adapter import UnitTestDatabaseAdapter
 from tests.message_dispatcher import UnitTestMessageDispatcher
+from tests.message_queue import UnitTestMessageQueue
 from tests.test_decoder_minimal import _MINIMAL_WORKFLOW
 from workflow.worklfow_validator import ValidationLevel, WorkflowValidator
 
 
 def test_validate_minimal_for_create():
     # Arrange
+    msg_queue = UnitTestMessageQueue()
     db_adapter = UnitTestDatabaseAdapter()
-    msg_dispatcher = UnitTestMessageDispatcher()
+    msg_dispatcher = UnitTestMessageDispatcher(msg_queue=msg_queue)
     validator = WorkflowValidator(db_adapter=db_adapter, msg_dispatcher=msg_dispatcher)
 
     # Act