feat: A test message queue implementation for protocol buffers

Author: Alan Christie

tests/message_queue.py

@@ -0,0 +1,43 @@
+from contextlib import suppress
+from multiprocessing import Event, Process, Queue
+from queue import Empty
+from typing import Callable
+
+from google.protobuf.message import Message
+from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
+from informaticsmatters.protobuf.datamanager.workflow_message_pb2 import WorkflowMessage
+
+
+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]):
+        super().__init__()
+        self._stop = Event()
+        self._queue = Queue()
+        self._receiver = receiver
+
+    def run(self):
+        while not self._stop.is_set():
+            with suppress(Empty):
+                if item := self._queue.get(True, 0.25):
+                    msg = None
+                    # We only support Workflow and Pod messages
+                    # during 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)
+
+    def put(self, msg: Message):
+        """Puts a protocol buffer message onto the queue."""
+        self._queue.put({"class": type(msg).__name__, "bytes": msg.SerializeToString()})
+
+    def stop(self):
+        """A request to stop the process."""
+        self._stop.set()

tests/test_test_message_queue.py

@@ -0,0 +1,59 @@
+import time
+from datetime import datetime, timezone
+
+import pytest
+
+pytestmark = pytest.mark.unit
+
+from google.protobuf.message import Message
+from informaticsmatters.protobuf.datamanager.workflow_message_pb2 import WorkflowMessage
+
+from tests.message_queue import UnitTestMessageQueue
+
+
+class Receiver:
+    """A dummy message receiver."""
+
+    def handle_msg(self, msg: Message) -> None:
+        assert msg
+        print(f"Received message ({type(msg).__name__})")
+
+
+def test_start_and_stop():
+    # Arrange
+    receiver = Receiver()
+    utmq = UnitTestMessageQueue(receiver=receiver.handle_msg)
+    utmq.start()
+
+    # Act
+    utmq.stop()
+    utmq.join()
+
+    # Assert
+
+
+def test_send_messages():
+    # Arrange
+    receiver = Receiver()
+    utmq = UnitTestMessageQueue(receiver=receiver.handle_msg)
+    utmq.start()
+
+    # Act
+    msg = WorkflowMessage()
+    msg.timestamp = f"{datetime.now(timezone.utc).isoformat()}Z"
+    msg.action = "START"
+    msg.running_workflow = "r-workflow-00000000-0000-0000-0000-000000000000"
+    utmq.put(msg)
+
+    msg = WorkflowMessage()
+    msg.timestamp = f"{datetime.now(timezone.utc).isoformat()}Z"
+    msg.action = "STOP"
+    msg.running_workflow = "r-workflow-00000000-0000-0000-0000-000000000000"
+    utmq.put(msg)
+
+    time.sleep(0.5)
+
+    utmq.stop()
+    utmq.join()
+
+    # Assert