tests for module control integration.

api/src/opentrons/hardware_control/emulation/module_server/client.py

@@ -3,8 +3,7 @@
 from typing import Optional
 
 from opentrons.hardware_control.emulation.module_server.models import Message
-from opentrons.hardware_control.emulation.module_server.server import \
-    MessageDelimiter
+from opentrons.hardware_control.emulation.module_server.server import MessageDelimiter
 
 
 class ModuleServerClient:

api/src/opentrons/hardware_control/emulation/module_server/helpers.py

@@ -1,8 +1,7 @@
 import asyncio
-from typing import Sequence, Set, Callable, List, Coroutine, Awaitable
+from typing import Sequence, Set, Callable, List, Awaitable
 
-from opentrons.hardware_control.emulation.module_server.client import \
-    ModuleServerClient
+from opentrons.hardware_control.emulation.module_server.client import ModuleServerClient
 from opentrons.hardware_control.emulation.module_server.models import Message
 from opentrons.hardware_control.emulation.module_server.server import log
 from opentrons.hardware_control.emulation.types import ModuleType
@@ -39,7 +38,6 @@ async def message_to_notify(message: Message, notify_method: NotifyMethod) -> No
         await notify_method([], [])
 
 
-
 async def wait_emulators(
     client: ModuleServerClient,
     modules: Sequence[ModuleType],

api/src/opentrons/hardware_control/emulation/module_server/models.py

@@ -17,4 +17,4 @@ class Message(BaseModel):
     """A message sent to module server clients."""
 
     status: Literal["connected", "disconnected", "dump"]
-    connections: List[Connection]
+    connections: List[Connection]

api/src/opentrons/hardware_control/emulation/module_server/server.py

@@ -2,8 +2,10 @@
 import logging
 from typing import Dict, Set
 
-from opentrons.hardware_control.emulation.module_server.models import Connection, \
-    Message
+from opentrons.hardware_control.emulation.module_server.models import (
+    Connection,
+    Message,
+)
 from opentrons.hardware_control.emulation.proxy import ProxyListener
 from opentrons.hardware_control.emulation.settings import ModuleServerSettings
 from typing_extensions import Final

api/tests/opentrons/hardware_control/emulation/module_server/test_helpers.py

@@ -1,9 +1,143 @@
+from typing import List
+
 import pytest
 from mock import AsyncMock
+from opentrons.drivers.rpi_drivers.types import USBPort
 from opentrons.hardware_control.emulation.module_server import helpers
+from opentrons.hardware_control.emulation.module_server import models
+from opentrons.hardware_control.modules import ModuleAtPort
+
+
+@pytest.fixture
+def mock_callback() -> AsyncMock:
+    """Callback mock."""
+    return AsyncMock(spec=helpers.NotifyMethod)
+
+
+@pytest.fixture
+def connections() -> List[models.Connection]:
+    """Connection models."""
+    return [
+        models.Connection(
+            url=f"url{i}", module_type=f"module_type{i}", identifier=f"identifier{i}"
+        )
+        for i in range(5)
+    ]
+
+
+@pytest.fixture
+def modules_at_port() -> List[ModuleAtPort]:
+    """Connection models."""
+    return [
+        ModuleAtPort(
+            port=f"url{i}",
+            name=f"module_type{i}",
+            usb_port=USBPort(name=f"identifier{i}", sub_names=[]),
+        )
+        for i in range(5)
+    ]
+
+
+async def test_message_to_notify_connected_empty(mock_callback: AsyncMock) -> None:
+    """It should call the call back with the correct modules to add."""
+    message = models.Message(status="connected", connections=[])
+    await helpers.ModuleListener.message_to_notify(
+        message=message, notify_method=mock_callback
+    )
+    mock_callback.assert_called_once_with([], [])
+
+
+async def test_message_to_notify_connected_one(
+    mock_callback: AsyncMock,
+    connections: List[models.Connection],
+    modules_at_port: List[ModuleAtPort],
+) -> None:
+    """It should call the call back with the correct modules to add."""
+    message = models.Message(status="connected", connections=connections[:1])
+    await helpers.ModuleListener.message_to_notify(
+        message=message, notify_method=mock_callback
+    )
+    mock_callback.assert_called_once_with(modules_at_port[:1], [])
+
+
+async def test_message_to_notify_connected_many(
+    mock_callback: AsyncMock,
+    connections: List[models.Connection],
+    modules_at_port: List[ModuleAtPort],
+) -> None:
+    """It should call the call back with the correct modules to add."""
+    message = models.Message(status="connected", connections=connections)
+    await helpers.ModuleListener.message_to_notify(
+        message=message, notify_method=mock_callback
+    )
+    mock_callback.assert_called_once_with(modules_at_port, [])
+
+
+async def test_message_to_notify_disconnected_empty(mock_callback: AsyncMock) -> None:
+    """It should call the call back with the correct modules to remove."""
+    message = models.Message(status="disconnected", connections=[])
+    await helpers.ModuleListener.message_to_notify(
+        message=message, notify_method=mock_callback
+    )
+    mock_callback.assert_called_once_with([], [])
+
+
+async def test_message_to_notify_disconnected_one(
+    mock_callback: AsyncMock,
+    connections: List[models.Connection],
+    modules_at_port: List[ModuleAtPort],
+) -> None:
+    """It should call the call back with the correct modules to remove."""
+    message = models.Message(status="disconnected", connections=connections[:1])
+    await helpers.ModuleListener.message_to_notify(
+        message=message, notify_method=mock_callback
+    )
+    mock_callback.assert_called_once_with([], modules_at_port[:1])
+
+
+async def test_message_to_notify_disconnected_many(
+    mock_callback: AsyncMock,
+    connections: List[models.Connection],
+    modules_at_port: List[ModuleAtPort],
+) -> None:
+    """It should call the call back with the correct modules to remove."""
+    message = models.Message(status="disconnected", connections=connections)
+    await helpers.ModuleListener.message_to_notify(
+        message=message, notify_method=mock_callback
+    )
+    mock_callback.assert_called_once_with([], modules_at_port)
+
+
+async def test_message_to_notify_dump_empty(mock_callback: AsyncMock) -> None:
+    """It should call the call back with the correct modules to load."""
+    message = models.Message(status="dump", connections=[])
+    await helpers.ModuleListener.message_to_notify(
+        message=message, notify_method=mock_callback
+    )
+    mock_callback.assert_called_once_with([], [])
+
 
+async def test_message_to_notify_dump_one(
+    mock_callback: AsyncMock,
+    connections: List[models.Connection],
+    modules_at_port: List[ModuleAtPort],
+) -> None:
+    """It should call the call back with the correct modules to load."""
+    message = models.Message(status="dump", connections=connections[:1])
+    await helpers.ModuleListener.message_to_notify(
+        message=message, notify_method=mock_callback
+    )
+    mock_callback.assert_called_once_with(modules_at_port[:1], [])
 
-@pytest
 
-async def test_message_to_notify():
-    pass
+async def test_message_to_notify_dump_many(
+    mock_callback: AsyncMock,
+    connections: List[models.Connection],
+    modules_at_port: List[ModuleAtPort],
+) -> None:
+    """It should call the call back with the correct modules to load."""
+    message = models.Message(status="dump", connections=connections)
+    await helpers.ModuleListener.message_to_notify(
+        message=message, notify_method=mock_callback
+    )
+    mock_callback.assert_called_once_with(modules_at_port, [])