test: Use of test fixtures

Author: Alan Christie

tests/instance_launcher.py

@@ -5,6 +5,7 @@
 
 from informaticsmatters.protobuf.datamanager.pod_message_pb2 import PodMessage
 
+from tests.message_dispatcher import UnitTestMessageDispatcher
 from workflow.workflow_abc import InstanceLauncher, LaunchResult
 
 _JOB_DIRECTORY: str = os.path.join(os.path.dirname(__file__), "jobs")
@@ -13,8 +14,15 @@
 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).
     """
 
+    def __init__(self, msg_dispatcher: UnitTestMessageDispatcher):
+        super().__init__()
+
+        self._msg_dispatcher = msg_dispatcher
+
     def launch(
         self,
         *,

tests/test_test_instance_launcher.py

@@ -5,14 +5,22 @@
 pytestmark = pytest.mark.unit
 
 from tests.instance_launcher import UnitTestInstanceLauncher
+from tests.message_dispatcher import UnitTestMessageDispatcher
+from tests.message_queue import UnitTestMessageQueue
 
 
-def test_get_nop_job():
+@pytest.fixture
+def basic_launcher():
+    utmq = UnitTestMessageQueue()
+    utmd = UnitTestMessageDispatcher(msg_queue=utmq)
+    return UnitTestInstanceLauncher(msg_dispatcher=utmd)
+
+
+def test_get_nop_job(basic_launcher):
     # Arrange
-    util = UnitTestInstanceLauncher()
 
     # Act
-    result = util.launch(
+    result = basic_launcher.launch(
         project_id="project-00000000-0000-0000-0000-000000000000",
         workflow_id="workflow-00000000-0000-0000-0000-000000000000",
         workflow_definition={},

tests/test_test_message_dispatcher.py

@@ -12,16 +12,20 @@
 from tests.message_queue import UnitTestMessageQueue
 
 
-def test_get_nop_job():
-    # Arrange
+@pytest.fixture
+def basic_dispatcher():
     utmq = UnitTestMessageQueue()
-    utmd = UnitTestMessageDispatcher(msg_queue=utmq)
+    return UnitTestMessageDispatcher(msg_queue=utmq)
+
+
+def test_get_nop_job(basic_dispatcher):
+    # Arrange
     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(msg)
+    basic_dispatcher.send(msg)
 
     # Assert

tests/test_test_message_queue.py

@@ -19,41 +19,41 @@ def handle_msg(self, msg: Message) -> None:
         print(f"Received message ({type(msg).__name__})")
 
 
-def test_start_and_stop():
-    # Arrange
+@pytest.fixture
+def basic_queue():
     receiver = Receiver()
     utmq = UnitTestMessageQueue(receiver=receiver.handle_msg)
     utmq.start()
-
-    # Act
+    yield utmq
     utmq.stop()
     utmq.join()
 
+
+def test_start_and_stop(basic_queue):
+    # Arrange
+
+    # Act
+    pass
+
     # Assert
 
 
-def test_send_messages():
+def test_send_messages(basic_queue):
     # 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)
+    basic_queue.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)
+    basic_queue.put(msg)
 
     time.sleep(0.5)
 
-    utmq.stop()
-    utmq.join()
-
     # Assert

tests/test_worflow_validator.py

@@ -11,15 +11,21 @@
 from workflow.worklfow_validator import ValidationLevel, WorkflowValidator
 
 
-def test_validate_minimal_for_create():
-    # Arrange
-    msg_queue = UnitTestMessageQueue()
+@pytest.fixture
+def basic_validator():
+    # A 'basic' unit-test WorkflowAdapter needs a DB adapter and Message Dispatcher.
+    # For testing outside of the DM the Message Dispatcher also needs a Message Queue
     db_adapter = UnitTestDatabaseAdapter()
+    msg_queue = UnitTestMessageQueue()
     msg_dispatcher = UnitTestMessageDispatcher(msg_queue=msg_queue)
-    validator = WorkflowValidator(db_adapter=db_adapter, msg_dispatcher=msg_dispatcher)
+    return WorkflowValidator(db_adapter=db_adapter, msg_dispatcher=msg_dispatcher)
+
+
+def test_validate_minimal_for_create(basic_validator):
+    # Arrange
 
     # Act
-    error = validator.validate(
+    error = basic_validator.validate(
         level=ValidationLevel.CREATE,
         workflow_definition=_MINIMAL_WORKFLOW,
     )