testing

robot-server/robot_server/service/notifications/__init__.py

@@ -6,13 +6,14 @@
     get_notification_client,
     clean_up_notification_client,
 )
-from .publisher_notifier import get_notify_publishers
+from .publisher_notifier import PublisherNotifier, get_notify_publishers
 from .publishers import (
     MaintenanceRunsPublisher,
     RunsPublisher,
     get_maintenance_runs_publisher,
     get_runs_publisher,
 )
+from .change_notifier import ChangeNotifier
 
 __all__ = [
     # main export
@@ -28,4 +29,7 @@
     "get_notify_publishers",
     "get_maintenance_runs_publisher",
     "get_runs_publisher",
+    # for testing
+    "PublisherNotifier",
+    "ChangeNotifier",
 ]

robot-server/robot_server/service/notifications/publisher_notifier.py

@@ -68,6 +68,3 @@ def get_notify_publishers(
     publisher_notifier = _publisher_notifier_accessor.get_from(app_state)
     assert isinstance(publisher_notifier, PublisherNotifier)
     return publisher_notifier.notify_publishers
-
-
-# TOME: Left to do. 1) Verify this works conceptually for the runs_publisher route. 2) Add testing. 3) Lint + docstrings.

robot-server/tests/maintenance_runs/test_run_data_manager.py

@@ -93,7 +93,6 @@ def subject(
     )
 
 
-# TOME: The problem is engine_state_summary isn't working for some reason.
 async def test_create(
     decoy: Decoy,
     mock_maintenance_engine_store: MaintenanceEngineStore,

robot-server/tests/service/notifications/test_change_notifier.py

@@ -0,0 +1,56 @@
+"""Tests for the ChangeNotifier interface."""
+import asyncio
+import pytest
+from opentrons.protocol_engine.state.change_notifier import ChangeNotifier
+
+
+async def test_single_subscriber() -> None:
+    """Test that a single subscriber can wait for a notification."""
+    subject = ChangeNotifier()
+    result = asyncio.create_task(subject.wait())
+
+    # ensure that the wait actually waits by delaying and
+    # checking that the task has not resolved
+    await asyncio.sleep(0.1)
+    assert result.done() is False
+
+    asyncio.get_running_loop().call_soon(subject.notify)
+
+    await result
+
+
+@pytest.mark.parametrize("_test_repetition", range(10))
+async def test_multiple_subscribers(_test_repetition: int) -> None:
+    """Test that multiple subscribers can wait for a notification.
+
+    This test checks that the subscribers are awoken in the order they
+    subscribed. This may or may not be guarenteed according to the
+    implementations of both ChangeNotifier and the event loop.
+    This test functions as a canary, given that our code may relies
+    on this ordering for determinism.
+
+    This test runs multiple times to check for flakyness.
+    """
+    subject = ChangeNotifier()
+    results = []
+
+    async def _do_task_1() -> None:
+        await subject.wait()
+        results.append(1)
+
+    async def _do_task_2() -> None:
+        await subject.wait()
+        results.append(2)
+
+    async def _do_task_3() -> None:
+        await subject.wait()
+        results.append(3)
+
+    task_1 = asyncio.create_task(_do_task_1())
+    task_2 = asyncio.create_task(_do_task_2())
+    task_3 = asyncio.create_task(_do_task_3())
+
+    asyncio.get_running_loop().call_soon(subject.notify)
+    await asyncio.gather(task_1, task_2, task_3)
+
+    assert results == [1, 2, 3]

robot-server/tests/service/notifications/test_publisher_notifier.py

@@ -0,0 +1,74 @@
+import asyncio
+from unittest.mock import Mock, MagicMock
+
+from robot_server.service.notifications import (
+    PublisherNotifier,
+    ChangeNotifier,
+)
+
+
+async def test_initialize() -> None:
+    """It should create a new task."""
+    publisher_notifier = PublisherNotifier()
+
+    await publisher_notifier.initialize()
+
+    assert asyncio.get_running_loop()
+
+
+def test_notify_publishers() -> None:
+    """Invoke the change notifier's notify method."""
+    change_notifier = MagicMock()
+    publisher_notifier = PublisherNotifier(change_notifier)
+
+    publisher_notifier.notify_publishers()
+
+    change_notifier.notify.assert_called_once()
+
+
+def test_register_publish_callbacks() -> None:
+    """It should extend the list of callbacks within a given list of callbacks."""
+    publisher_notifier = PublisherNotifier()
+    callback1 = Mock()
+    callback2 = Mock()
+
+    publisher_notifier.register_publish_callbacks([callback1, callback2])
+
+    assert len(publisher_notifier._callbacks) == 2
+    assert publisher_notifier._callbacks[0] == callback1
+    assert publisher_notifier._callbacks[1] == callback2
+
+
+async def test_wait_for_event() -> None:
+    """It should wait for an event to occur, then invoke each callback."""
+    change_notifier = ChangeNotifier()
+    publisher_notifier = PublisherNotifier(change_notifier)
+
+    callback_called = False
+    callback_2_called = False
+
+    async def callback() -> None:
+        """Mock callback."""
+        nonlocal callback_called
+        callback_called = True
+
+    async def callback_2() -> None:
+        """Mock callback."""
+        nonlocal callback_2_called
+        callback_2_called = True
+
+    publisher_notifier.register_publish_callbacks([callback, callback_2])
+
+    async def trigger_callbacks() -> None:
+        """Mock trigger for callbacks."""
+        await asyncio.sleep(0.1)
+        change_notifier.notify()
+
+    task = asyncio.create_task(publisher_notifier.initialize())
+
+    await asyncio.gather(trigger_callbacks(), task)
+
+    assert callback_called
+    assert callback_2_called
+
+    task.cancel()