From 29b7509564deb1dc69caedf5a162bfccee0858ce Mon Sep 17 00:00:00 2001 From: amit lissack Date: Fri, 22 Oct 2021 16:30:09 -0400 Subject: [PATCH] tests for module control integration. --- .../emulation/module_server/client.py | 3 +- .../emulation/module_server/helpers.py | 6 +- .../emulation/module_server/models.py | 2 +- .../emulation/module_server/server.py | 6 +- .../emulation/module_server/test_helpers.py | 140 +++++++++++++++++- 5 files changed, 145 insertions(+), 12 deletions(-) diff --git a/api/src/opentrons/hardware_control/emulation/module_server/client.py b/api/src/opentrons/hardware_control/emulation/module_server/client.py index 2881623890c0..bf60164a1f9a 100644 --- a/api/src/opentrons/hardware_control/emulation/module_server/client.py +++ b/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: diff --git a/api/src/opentrons/hardware_control/emulation/module_server/helpers.py b/api/src/opentrons/hardware_control/emulation/module_server/helpers.py index 279bee34fe20..450bf8ad1d34 100644 --- a/api/src/opentrons/hardware_control/emulation/module_server/helpers.py +++ b/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], diff --git a/api/src/opentrons/hardware_control/emulation/module_server/models.py b/api/src/opentrons/hardware_control/emulation/module_server/models.py index 9ee203ea42ac..f2c4b0c598e0 100644 --- a/api/src/opentrons/hardware_control/emulation/module_server/models.py +++ b/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] \ No newline at end of file + connections: List[Connection] diff --git a/api/src/opentrons/hardware_control/emulation/module_server/server.py b/api/src/opentrons/hardware_control/emulation/module_server/server.py index 62d7743a39a9..13da64103929 100644 --- a/api/src/opentrons/hardware_control/emulation/module_server/server.py +++ b/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 diff --git a/api/tests/opentrons/hardware_control/emulation/module_server/test_helpers.py b/api/tests/opentrons/hardware_control/emulation/module_server/test_helpers.py index f50c9f9194c0..45ad51b6a686 100644 --- a/api/tests/opentrons/hardware_control/emulation/module_server/test_helpers.py +++ b/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 \ No newline at end of file +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, [])