From 21bac8fda5f04b3939e61cdc4d38beda6df59ce2 Mon Sep 17 00:00:00 2001 From: aaron-kulkarni Date: Wed, 31 Jul 2024 16:48:25 -0400 Subject: [PATCH] test_instruments file. ~215 errors left --- .../hardware_control/test_instruments.py | 252 +++++++++++++----- 1 file changed, 187 insertions(+), 65 deletions(-) diff --git a/api/tests/opentrons/hardware_control/test_instruments.py b/api/tests/opentrons/hardware_control/test_instruments.py index a3d4f369694..65b5b84ed50 100644 --- a/api/tests/opentrons/hardware_control/test_instruments.py +++ b/api/tests/opentrons/hardware_control/test_instruments.py @@ -1,9 +1,25 @@ +import pdb import asyncio +from _pytest.fixtures import SubRequest import mock +from opentrons.hardware_control.ot3api import OT3API +from opentrons_shared_data.pipette.types import PipetteName import pytest from decoy import Decoy -from typing import Iterator +from typing import ( + Any, + Awaitable, + Dict, + Callable, + Coroutine, + Literal, + Iterator, + Tuple, + Optional, + TypeAlias, + Union, +) try: import aionotify # type: ignore[import-untyped] @@ -14,6 +30,7 @@ from opentrons import types from opentrons.hardware_control import API from opentrons.hardware_control.types import Axis, OT3Mount, HardwareFeatureFlags +from opentrons.types import Mount from opentrons_shared_data.errors.exceptions import CommandPreconditionViolated @@ -21,8 +38,18 @@ LEFT_PIPETTE_MODEL = "{}_v1".format(LEFT_PIPETTE_PREFIX) LEFT_PIPETTE_ID = "testy" +# mypy: error +# return-value - Incompatible return value type (got "dict[Mount, dict[str, str]]", expected "dict[Mount, dict[str, str | None] | None]") +# mypy: error +# arg-type - Argument "attached_instruments" to "build_hardware_simulator" of "API" has incompatible type "dict[Mount, dict[str, str | None] | None]"; expected "dict[Mount, dict[str, str | None]] | None" +DummyInstrumentConfig: TypeAlias = Optional[Dict[Mount, Dict[str, Optional[str]]]] +OT3DummyInstrumentConfig: TypeAlias = Dict[ + Union[Mount, OT3Mount], Optional[Dict[str, Optional[str]]] +] +OldDummyInstrumentConfig: TypeAlias = Optional[Dict[Mount, Dict[str, Optional[str]]]] + -def dummy_instruments_attached(): +def dummy_instruments_attached() -> Tuple[DummyInstrumentConfig, int]: return { types.Mount.LEFT: { "model": LEFT_PIPETTE_MODEL, @@ -38,11 +65,11 @@ def dummy_instruments_attached(): @pytest.fixture -def dummy_instruments(): +def dummy_instruments() -> Tuple[DummyInstrumentConfig, int]: return dummy_instruments_attached() -def dummy_instruments_attached_ot3(): +def dummy_instruments_attached_ot3() -> Tuple[OT3DummyInstrumentConfig, int]: return { types.Mount.LEFT: { "model": "p1000_single_v3.3", @@ -55,12 +82,12 @@ def dummy_instruments_attached_ot3(): @pytest.fixture -def dummy_instruments_ot3(): +def dummy_instruments_ot3() -> Tuple[OT3DummyInstrumentConfig, int]: return dummy_instruments_attached_ot3() @pytest.fixture -def mock_api_verify_tip_presence_ot3(request) -> Iterator[mock.AsyncMock]: +def mock_api_verify_tip_presence_ot3(request: SubRequest) -> Iterator[mock.AsyncMock]: if request.config.getoption("--ot2-only"): pytest.skip("testing ot2 only") from opentrons.hardware_control.ot3api import OT3API @@ -69,7 +96,7 @@ def mock_api_verify_tip_presence_ot3(request) -> Iterator[mock.AsyncMock]: yield mock_tip_presence -def wrap_build_ot3_sim(): +def wrap_build_ot3_sim() -> Callable[[Any], Coroutine[Any, Any, OT3API]]: from opentrons.hardware_control.ot3api import OT3API with mock.patch.object( @@ -79,7 +106,9 @@ def wrap_build_ot3_sim(): @pytest.fixture -def ot3_api_obj(request, mock_api_verify_tip_presence_ot3): +def ot3_api_obj( + request: SubRequest, mock_api_verify_tip_presence_ot3: Iterator[mock.AsyncMock] +) -> Callable[[Any], Coroutine[Any, Any, OT3API]]: if request.config.getoption("--ot2-only"): pytest.skip("testing ot2 only") from opentrons.hardware_control.ot3api import OT3API @@ -94,7 +123,7 @@ def ot3_api_obj(request, mock_api_verify_tip_presence_ot3): ], ids=["ot2", "ot3"], ) -def sim_and_instr(request): +def sim_and_instr(request: SubRequest) -> Iterator[Tuple[Any, Any]]: if ( request.node.get_closest_marker("ot2_only") and request.param[0] == wrap_build_ot3_sim @@ -114,7 +143,7 @@ def sim_and_instr(request): @pytest.fixture -def dummy_backwards_compatibility(): +def dummy_backwards_compatibility() -> OldDummyInstrumentConfig: dummy_instruments_attached = { types.Mount.LEFT: { "model": "p20_single_v2.0", @@ -127,20 +156,26 @@ def dummy_backwards_compatibility(): "name": "p300_single_gen2", }, } - return dummy_instruments_attached + return dummy_instruments_attached # type: ignore[return-value] -def get_plunger_speed(api): +def get_plunger_speed(api: Any) -> Any: if isinstance(api, API): return api.plunger_speed else: return api._pipette_handler.plunger_speed -async def test_cache_instruments(sim_and_instr): - sim_builder, (dummy_instruments, _) = sim_and_instr +async def test_cache_instruments( + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ] +) -> None: + sim_builder = sim_and_instr[0] + assert sim_and_instr[1] is not None + dummy_instruments = sim_and_instr[1] hw_api = await sim_builder( - attached_instruments=dummy_instruments, loop=asyncio.get_running_loop() + attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) await hw_api.cache_instruments() @@ -151,10 +186,16 @@ async def test_cache_instruments(sim_and_instr): # typeguard.check_type("left mount dict", attached[types.Mount.LEFT], PipetteDict) -async def test_mismatch_fails(sim_and_instr): - sim_builder, (dummy_instruments, _) = sim_and_instr +async def test_mismatch_fails( + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ] +) -> None: + sim_builder = sim_and_instr[0] + assert sim_and_instr[1] is not None + dummy_instruments = sim_and_instr[1] hw_api = await sim_builder( - attached_instruments=dummy_instruments, loop=asyncio.get_running_loop() + attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) requested_instr = { types.Mount.LEFT: "p20_single_gen2", @@ -164,12 +205,17 @@ async def test_mismatch_fails(sim_and_instr): await hw_api.cache_instruments(requested_instr) -async def test_backwards_compatibility(dummy_backwards_compatibility): +async def test_backwards_compatibility( + dummy_backwards_compatibility: OldDummyInstrumentConfig, +) -> None: hw_api = await API.build_hardware_simulator( attached_instruments=dummy_backwards_compatibility, loop=asyncio.get_running_loop(), ) - requested_instr = {types.Mount.LEFT: "p10_single", types.Mount.RIGHT: "p300_single"} + requested_instr: Optional[Dict[types.Mount, PipetteName]] = { + types.Mount.LEFT: "p10_single", + types.Mount.RIGHT: "p300_single", + } volumes = { types.Mount.LEFT: {"min": 1, "max": 10}, types.Mount.RIGHT: {"min": 30, "max": 300}, @@ -177,7 +223,9 @@ async def test_backwards_compatibility(dummy_backwards_compatibility): await hw_api.cache_instruments(requested_instr) attached = hw_api.attached_instruments + assert requested_instr is not None for mount, name in requested_instr.items(): + assert dummy_backwards_compatibility is not None assert attached[mount]["name"] == dummy_backwards_compatibility[mount]["name"] assert attached[mount]["min_volume"] == volumes[mount]["min"] assert attached[mount]["max_volume"] == volumes[mount]["max"] @@ -185,22 +233,18 @@ async def test_backwards_compatibility(dummy_backwards_compatibility): @pytest.mark.skipif(aionotify is None, reason="inotify not available") async def test_cache_instruments_hc( - monkeypatch, - dummy_instruments, - hardware_controller_lockfile, - is_robot, - cntrlr_mock_connect, -): + monkeypatch: pytest.MonkeyPatch, +) -> None: hw_api_cntrlr = await API.build_hardware_controller( loop=asyncio.get_running_loop(), feature_flags=HardwareFeatureFlags.build_from_ff(), ) - async def mock_driver_model(mount): + async def mock_driver_model(mount: str) -> Optional[str]: attached_pipette = {"left": LEFT_PIPETTE_MODEL, "right": None} return attached_pipette[mount] - async def mock_driver_id(mount): + async def mock_driver_id(mount: str) -> Optional[str]: attached_pipette = {"left": LEFT_PIPETTE_ID, "right": None} return attached_pipette[mount] @@ -223,7 +267,7 @@ async def mock_driver_id(mount): await hw_api_cntrlr.cache_instruments({types.Mount.LEFT: "p300_multi"}) # If we pass a matching expects it should work - await hw_api_cntrlr.cache_instruments({types.Mount.LEFT: LEFT_PIPETTE_PREFIX}) + await hw_api_cntrlr.cache_instruments({types.Mount.LEFT: "p10_single"}) # TODO: (ba, 2023-03-08): no longer true, change this # attached = hw_api_cntrlr.attached_instruments # typeguard.check_type( @@ -232,13 +276,19 @@ async def mock_driver_id(mount): @pytest.mark.ot2_only -async def test_cache_instruments_sim(sim_and_instr): - sim_builder, (dummy_instruments, _) = sim_and_instr +async def test_cache_instruments_sim( + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ] +) -> None: + sim_builder = sim_and_instr[0] + assert sim_and_instr[1] is not None + dummy_instruments = sim_and_instr[1] - def fake_func1(value): + def fake_func1(value: Any) -> Any: return value - def fake_func2(mount, value): + def fake_func2(mount: Any, value: Any) -> Any: return mount, value sim = await sim_builder(loop=asyncio.get_running_loop()) @@ -295,7 +345,7 @@ def fake_func2(mount, value): assert attached[types.Mount.RIGHT]["name"] == "p300_single" # If we specify instruments at init time, we should get them without # passing an expectation - sim = await sim_builder(attached_instruments=dummy_instruments) + sim = await sim_builder(attached_instruments=dummy_instruments[0]) await sim.cache_instruments() attached = sim.attached_instruments # TODO: (ba, 2023-03-08): no longer true, change this @@ -320,10 +370,17 @@ def fake_func2(mount, value): await sim.cache_instruments({types.Mount.LEFT: "p10_sing"}) -async def test_prep_aspirate(sim_and_instr): - sim_builder, (dummy_instruments, dummy_tip_vol) = sim_and_instr +async def test_prep_aspirate( + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ] +) -> None: + sim_builder = sim_and_instr[0] + assert sim_and_instr[1] is not None + dummy_instruments = sim_and_instr[1] + dummy_tip_vol = dummy_instruments[1] hw_api = await sim_builder( - attached_instruments=dummy_instruments, loop=asyncio.get_running_loop() + attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) await hw_api.home() await hw_api.cache_instruments() @@ -351,7 +408,9 @@ async def test_prep_aspirate(sim_and_instr): await hw_api.aspirate(mount, 1, 1.0) -async def test_aspirate_new(dummy_instruments): +async def test_aspirate_new( + dummy_instruments: Tuple[DummyInstrumentConfig, int] +) -> None: hw_api = await API.build_hardware_simulator( attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop(), @@ -372,7 +431,9 @@ async def test_aspirate_new(dummy_instruments): assert pos[Axis.B] == pytest.approx(new_plunger_pos) -async def test_aspirate_old(decoy: Decoy, dummy_instruments): +async def test_aspirate_old( + decoy: Decoy, dummy_instruments: Tuple[DummyInstrumentConfig, int] +) -> None: hw_api = await API.build_hardware_simulator( attached_instruments=dummy_instruments[0], @@ -394,7 +455,11 @@ async def test_aspirate_old(decoy: Decoy, dummy_instruments): assert pos[Axis.B] == pytest.approx(new_plunger_pos) -async def test_aspirate_ot3_50(dummy_instruments_ot3, ot3_api_obj): +async def test_aspirate_ot3_50( + dummy_instruments_ot3: Tuple[OT3DummyInstrumentConfig, int], + ot3_api_obj: Callable[..., Awaitable[Any]], +) -> None: + assert dummy_instruments_ot3 is not None hw_api = await ot3_api_obj( attached_instruments=dummy_instruments_ot3[0], loop=asyncio.get_running_loop() ) @@ -413,7 +478,10 @@ async def test_aspirate_ot3_50(dummy_instruments_ot3, ot3_api_obj): assert pos[Axis.B] == pytest.approx(new_plunger_pos) -async def test_aspirate_ot3_1000(dummy_instruments_ot3, ot3_api_obj): +async def test_aspirate_ot3_1000( + dummy_instruments_ot3: Tuple[OT3DummyInstrumentConfig, int], + ot3_api_obj: Callable[..., Awaitable[Any]], +) -> None: hw_api = await ot3_api_obj( attached_instruments=dummy_instruments_ot3[0], loop=asyncio.get_running_loop() ) @@ -433,7 +501,7 @@ async def test_aspirate_ot3_1000(dummy_instruments_ot3, ot3_api_obj): assert pos[Axis.B] == pytest.approx(new_plunger_pos) -async def test_configure_ot3(ot3_api_obj): +async def test_configure_ot3(ot3_api_obj: Callable[..., Awaitable[Any]]) -> None: instrs = { types.Mount.LEFT: { "model": "p50_multi_v3.3", @@ -468,7 +536,9 @@ async def test_configure_ot3(ot3_api_obj): assert pos[Axis.B] == pytest.approx(71.5) -async def test_dispense_ot2(dummy_instruments): +async def test_dispense_ot2( + dummy_instruments: Tuple[DummyInstrumentConfig, int] +) -> None: hw_api = await API.build_hardware_simulator( attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) @@ -495,7 +565,10 @@ async def test_dispense_ot2(dummy_instruments): assert (await hw_api.current_position(mount))[Axis.B] == plunger_pos_2 -async def test_dispense_ot3(dummy_instruments_ot3, ot3_api_obj): +async def test_dispense_ot3( + dummy_instruments_ot3: Tuple[OT3DummyInstrumentConfig, int], + ot3_api_obj: Callable[..., Awaitable[Any]], +) -> None: hw_api = await ot3_api_obj( attached_instruments=dummy_instruments_ot3[0], loop=asyncio.get_running_loop() ) @@ -527,10 +600,16 @@ async def test_dispense_ot3(dummy_instruments_ot3, ot3_api_obj): ) -async def test_no_pipette(sim_and_instr): - sim_builder, (dummy_instruments, _) = sim_and_instr +async def test_no_pipette( + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ] +) -> None: + sim_builder = sim_and_instr[0] + assert sim_and_instr[1] is not None + dummy_instruments = sim_and_instr[1] hw_api = await sim_builder( - attached_instruments=dummy_instruments, loop=asyncio.get_running_loop() + attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) await hw_api.cache_instruments() aspirate_ul = 3.0 @@ -540,11 +619,18 @@ async def test_no_pipette(sim_and_instr): assert not hw_api._current_volume[types.Mount.RIGHT] -async def test_tip_pickup_moves(sim_and_instr): +async def test_tip_pickup_moves( + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ] +) -> None: """Make sure that tip_pickup_moves does not add a tip to the instrument.""" - sim_builder, (dummy_instruments, _) = sim_and_instr + sim_builder = sim_and_instr[0] + assert sim_and_instr[1] is not None + dummy_instruments = sim_and_instr[1] + hw_api = await sim_builder( - attached_instruments=dummy_instruments, loop=asyncio.get_running_loop() + attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) mount = types.Mount.LEFT await hw_api.home() @@ -563,10 +649,17 @@ async def test_tip_pickup_moves(sim_and_instr): assert not hw_api.hardware_instruments[mount].has_tip -async def test_pick_up_tip(is_robot, sim_and_instr): - sim_builder, (dummy_instruments, _) = sim_and_instr +async def test_pick_up_tip( + is_robot: bool, + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ], +) -> None: + sim_builder = sim_and_instr[0] + assert sim_and_instr[1] is not None + dummy_instruments = sim_and_instr[1] hw_api = await sim_builder( - attached_instruments=dummy_instruments, loop=asyncio.get_running_loop() + attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) mount = types.Mount.LEFT await hw_api.home() @@ -583,7 +676,9 @@ async def test_pick_up_tip(is_robot, sim_and_instr): assert hw_api.hardware_instruments[mount].current_volume == 0 -async def test_pick_up_tip_pos_ot2(is_robot, dummy_instruments): +async def test_pick_up_tip_pos_ot2( + is_robot, dummy_instruments: Tuple[DummyInstrumentConfig, int] +) -> None: hw_api = await API.build_hardware_simulator( attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) @@ -605,8 +700,9 @@ async def test_pick_up_tip_pos_ot2(is_robot, dummy_instruments): assert hw_api._current_position[k] == v, f"{k} position doesnt match" -def assert_move_called(mock_move, speed, lock=None): +def assert_move_called(mock_move: mock.Mock, speed: float, lock: Any = None) -> None: if lock is not None: + pdb.set_trace() mock_move.assert_called_with( mock.ANY, speed=speed, @@ -621,10 +717,17 @@ def assert_move_called(mock_move, speed, lock=None): ) -async def test_aspirate_flow_rate(sim_and_instr): - sim_builder, (dummy_instruments, tip_vol) = sim_and_instr +async def test_aspirate_flow_rate( + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ] +) -> None: + sim_builder = sim_and_instr[0] + assert sim_and_instr[1] is not None + dummy_instruments = sim_and_instr[1] + tip_vol = dummy_instruments[1] hw_api = await sim_builder( - attached_instruments=dummy_instruments, loop=asyncio.get_running_loop() + attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) mount = types.Mount.LEFT await hw_api.home() @@ -679,10 +782,17 @@ async def test_aspirate_flow_rate(sim_and_instr): assert_move_called(mock_move, 5) -async def test_dispense_flow_rate(sim_and_instr): - sim_builder, (dummy_instruments, tip_vol) = sim_and_instr +async def test_dispense_flow_rate( + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ] +) -> None: + sim_builder = sim_and_instr[0] + assert sim_and_instr[1] is not None + dummy_instruments = sim_and_instr[1] + tip_vol = dummy_instruments[1] hw_api = await sim_builder( - attached_instruments=dummy_instruments, loop=asyncio.get_running_loop() + attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) mount = types.Mount.LEFT await hw_api.home() @@ -735,10 +845,17 @@ async def test_dispense_flow_rate(sim_and_instr): assert_move_called(mock_move, 5) -async def test_blowout_flow_rate(sim_and_instr): - sim_builder, (dummy_instruments, tip_vol) = sim_and_instr +async def test_blowout_flow_rate( + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ] +) -> None: + sim_builder = sim_and_instr[0] + assert sim_and_instr[1] is not None + dummy_instruments = sim_and_instr[1] + tip_vol = dummy_instruments[1] hw_api = await sim_builder( - attached_instruments=dummy_instruments, loop=asyncio.get_running_loop() + attached_instruments=dummy_instruments[0], loop=asyncio.get_running_loop() ) mount = types.Mount.LEFT await hw_api.home() @@ -776,7 +893,12 @@ async def test_blowout_flow_rate(sim_and_instr): assert_move_called(mock_move, 15) -async def test_reset_instruments(monkeypatch, sim_and_instr): +async def test_reset_instruments( + monkeypatch: pytest.MonkeyPatch, + sim_and_instr: Tuple[ + Callable[..., Awaitable[Any]], Tuple[DummyInstrumentConfig, float] + ], +) -> None: instruments = { types.Mount.LEFT: { "model": "p1000_single_v3.3",