From feeb999f68dd9dd50cf786d16b4ae056be8870b3 Mon Sep 17 00:00:00 2001 From: Jeremy Leon Date: Thu, 14 Nov 2024 15:42:08 -0500 Subject: [PATCH] fix(api): refactor protocol api integration tests to prevent thread leakage (#16834) Fixes an issue found where after a recent mergeback commit into edge, API unit tests were hanging and failing due to thread leakage in protocol api integration tests. --- .../protocol_api_integration/conftest.py | 28 ++++ .../test_liquid_classes.py | 32 +++-- .../protocol_api_integration/test_modules.py | 46 ++++--- .../test_pipette_movement_deck_conflicts.py | 129 ++++++++++++------ .../protocol_api_integration/test_trashes.py | 94 ++++++------- 5 files changed, 211 insertions(+), 118 deletions(-) create mode 100644 api/tests/opentrons/protocol_api_integration/conftest.py diff --git a/api/tests/opentrons/protocol_api_integration/conftest.py b/api/tests/opentrons/protocol_api_integration/conftest.py new file mode 100644 index 00000000000..fa98ccbb039 --- /dev/null +++ b/api/tests/opentrons/protocol_api_integration/conftest.py @@ -0,0 +1,28 @@ +"""Fixtures for protocol api integration tests.""" + +import pytest +from _pytest.fixtures import SubRequest +from typing import Generator + +from opentrons import simulate, protocol_api +from opentrons.protocol_api.core.engine import ENGINE_CORE_API_VERSION + + +@pytest.fixture +def simulated_protocol_context( + request: SubRequest, +) -> Generator[protocol_api.ProtocolContext, None, None]: + """Return a protocol context with requested version and robot.""" + version, robot_type = request.param + context = simulate.get_protocol_api(version=version, robot_type=robot_type) + try: + yield context + finally: + if context.api_version >= ENGINE_CORE_API_VERSION: + # TODO(jbl, 2024-11-14) this is a hack of a hack to close the hardware and the PE thread when a test is + # complete. At some point this should be replaced with a more holistic way of safely cleaning up these + # threads so they don't leak and cause tests to fail when `get_protocol_api` is called too many times. + simulate._LIVE_PROTOCOL_ENGINE_CONTEXTS.close() + else: + # If this is a non-PE context we need to clean up the hardware thread manually + context._hw_manager.hardware.clean_up() diff --git a/api/tests/opentrons/protocol_api_integration/test_liquid_classes.py b/api/tests/opentrons/protocol_api_integration/test_liquid_classes.py index 6621a790801..1a6e19f85be 100644 --- a/api/tests/opentrons/protocol_api_integration/test_liquid_classes.py +++ b/api/tests/opentrons/protocol_api_integration/test_liquid_classes.py @@ -3,22 +3,30 @@ from decoy import Decoy from opentrons_shared_data.robot.types import RobotTypeEnum -from opentrons import simulate +from opentrons.protocol_api import ProtocolContext from opentrons.config import feature_flags as ff @pytest.mark.ot2_only +@pytest.mark.parametrize( + "simulated_protocol_context", [("2.20", "OT-2")], indirect=True +) def test_liquid_class_creation_and_property_fetching( - decoy: Decoy, mock_feature_flags: None + decoy: Decoy, + mock_feature_flags: None, + simulated_protocol_context: ProtocolContext, ) -> None: """It should create the liquid class and provide access to its properties.""" decoy.when(ff.allow_liquid_classes(RobotTypeEnum.OT2)).then_return(True) - protocol_context = simulate.get_protocol_api(version="2.20", robot_type="OT-2") - pipette_left = protocol_context.load_instrument("p20_single_gen2", mount="left") - pipette_right = protocol_context.load_instrument("p300_multi", mount="right") - tiprack = protocol_context.load_labware("opentrons_96_tiprack_20ul", "1") + pipette_left = simulated_protocol_context.load_instrument( + "p20_single_gen2", mount="left" + ) + pipette_right = simulated_protocol_context.load_instrument( + "p300_multi", mount="right" + ) + tiprack = simulated_protocol_context.load_labware("opentrons_96_tiprack_20ul", "1") - glycerol_50 = protocol_context.define_liquid_class("fixture_glycerol50") + glycerol_50 = simulated_protocol_context.define_liquid_class("fixture_glycerol50") assert glycerol_50.name == "fixture_glycerol50" assert glycerol_50.display_name == "Glycerol 50%" @@ -50,11 +58,13 @@ def test_liquid_class_creation_and_property_fetching( glycerol_50.display_name = "bar" # type: ignore with pytest.raises(ValueError, match="Liquid class definition not found"): - protocol_context.define_liquid_class("non-existent-liquid") + simulated_protocol_context.define_liquid_class("non-existent-liquid") -def test_liquid_class_feature_flag() -> None: +@pytest.mark.parametrize( + "simulated_protocol_context", [("2.20", "OT-2")], indirect=True +) +def test_liquid_class_feature_flag(simulated_protocol_context: ProtocolContext) -> None: """It should raise a not implemented error without the allowLiquidClass flag set.""" - protocol_context = simulate.get_protocol_api(version="2.20", robot_type="OT-2") with pytest.raises(NotImplementedError): - protocol_context.define_liquid_class("fixture_glycerol50") + simulated_protocol_context.define_liquid_class("fixture_glycerol50") diff --git a/api/tests/opentrons/protocol_api_integration/test_modules.py b/api/tests/opentrons/protocol_api_integration/test_modules.py index e8a26112d88..72ee8ed8c52 100644 --- a/api/tests/opentrons/protocol_api_integration/test_modules.py +++ b/api/tests/opentrons/protocol_api_integration/test_modules.py @@ -3,13 +3,17 @@ import typing import pytest -from opentrons import simulate, protocol_api +from opentrons import protocol_api -def test_absorbance_reader_labware_load_conflict() -> None: +@pytest.mark.parametrize( + "simulated_protocol_context", [("2.21", "Flex")], indirect=True +) +def test_absorbance_reader_labware_load_conflict( + simulated_protocol_context: protocol_api.ProtocolContext, +) -> None: """It should prevent loading a labware onto a closed absorbance reader.""" - protocol = simulate.get_protocol_api(version="2.21", robot_type="Flex") - module = protocol.load_module("absorbanceReaderV1", "A3") + module = simulated_protocol_context.load_module("absorbanceReaderV1", "A3") # The lid should be treated as initially closed. with pytest.raises(Exception): @@ -19,7 +23,7 @@ def test_absorbance_reader_labware_load_conflict() -> None: # Should not raise after opening the lid. labware_1 = module.load_labware("opentrons_96_wellplate_200ul_pcr_full_skirt") - protocol.move_labware(labware_1, protocol_api.OFF_DECK) + simulated_protocol_context.move_labware(labware_1, protocol_api.OFF_DECK) # Should raise after closing the lid again. module.close_lid() # type: ignore[union-attr] @@ -27,34 +31,44 @@ def test_absorbance_reader_labware_load_conflict() -> None: module.load_labware("opentrons_96_wellplate_200ul_pcr_full_skirt") -def test_absorbance_reader_labware_move_conflict() -> None: +@pytest.mark.parametrize( + "simulated_protocol_context", [("2.21", "Flex")], indirect=True +) +def test_absorbance_reader_labware_move_conflict( + simulated_protocol_context: protocol_api.ProtocolContext, +) -> None: """It should prevent moving a labware onto a closed absorbance reader.""" - protocol = simulate.get_protocol_api(version="2.21", robot_type="Flex") - module = protocol.load_module("absorbanceReaderV1", "A3") - labware = protocol.load_labware("opentrons_96_wellplate_200ul_pcr_full_skirt", "A1") + module = simulated_protocol_context.load_module("absorbanceReaderV1", "A3") + labware = simulated_protocol_context.load_labware( + "opentrons_96_wellplate_200ul_pcr_full_skirt", "A1" + ) with pytest.raises(Exception): # The lid should be treated as initially closed. - protocol.move_labware(labware, module, use_gripper=True) + simulated_protocol_context.move_labware(labware, module, use_gripper=True) module.open_lid() # type: ignore[union-attr] # Should not raise after opening the lid. - protocol.move_labware(labware, module, use_gripper=True) + simulated_protocol_context.move_labware(labware, module, use_gripper=True) - protocol.move_labware(labware, "A1", use_gripper=True) + simulated_protocol_context.move_labware(labware, "A1", use_gripper=True) # Should raise after closing the lid again. module.close_lid() # type: ignore[union-attr] with pytest.raises(Exception): - protocol.move_labware(labware, module, use_gripper=True) + simulated_protocol_context.move_labware(labware, module, use_gripper=True) -def test_absorbance_reader_read_preconditions() -> None: +@pytest.mark.parametrize( + "simulated_protocol_context", [("2.21", "Flex")], indirect=True +) +def test_absorbance_reader_read_preconditions( + simulated_protocol_context: protocol_api.ProtocolContext, +) -> None: """Test the preconditions for triggering an absorbance reader read.""" - protocol = simulate.get_protocol_api(version="2.21", robot_type="Flex") module = typing.cast( protocol_api.AbsorbanceReaderContext, - protocol.load_module("absorbanceReaderV1", "A3"), + simulated_protocol_context.load_module("absorbanceReaderV1", "A3"), ) with pytest.raises(Exception, match="initialize"): diff --git a/api/tests/opentrons/protocol_api_integration/test_pipette_movement_deck_conflicts.py b/api/tests/opentrons/protocol_api_integration/test_pipette_movement_deck_conflicts.py index cad2bffddf9..2b7fc11ca91 100644 --- a/api/tests/opentrons/protocol_api_integration/test_pipette_movement_deck_conflicts.py +++ b/api/tests/opentrons/protocol_api_integration/test_pipette_movement_deck_conflicts.py @@ -2,54 +2,59 @@ import pytest -from opentrons import simulate -from opentrons.protocol_api import COLUMN, ALL, SINGLE, ROW +from opentrons.protocol_api import COLUMN, ALL, SINGLE, ROW, ProtocolContext from opentrons.protocol_api.core.engine.pipette_movement_conflict import ( PartialTipMovementNotAllowedError, ) @pytest.mark.ot3_only -def test_deck_conflicts_for_96_ch_a12_column_configuration() -> None: +@pytest.mark.parametrize( + "simulated_protocol_context", [("2.16", "Flex")], indirect=True +) +def test_deck_conflicts_for_96_ch_a12_column_configuration( + simulated_protocol_context: ProtocolContext, +) -> None: """It should raise errors for the expected deck conflicts.""" - protocol_context = simulate.get_protocol_api(version="2.16", robot_type="Flex") - trash_labware = protocol_context.load_labware( + trash_labware = simulated_protocol_context.load_labware( "opentrons_1_trash_3200ml_fixed", "A3" ) - badly_placed_tiprack = protocol_context.load_labware( + badly_placed_tiprack = simulated_protocol_context.load_labware( "opentrons_flex_96_tiprack_50ul", "C2" ) - well_placed_tiprack = protocol_context.load_labware( + well_placed_tiprack = simulated_protocol_context.load_labware( "opentrons_flex_96_tiprack_50ul", "C1" ) - tiprack_on_adapter = protocol_context.load_labware( + tiprack_on_adapter = simulated_protocol_context.load_labware( "opentrons_flex_96_tiprack_50ul", "C3", adapter="opentrons_flex_96_tiprack_adapter", ) - thermocycler = protocol_context.load_module("thermocyclerModuleV2") - tc_adjacent_plate = protocol_context.load_labware( + thermocycler = simulated_protocol_context.load_module("thermocyclerModuleV2") + tc_adjacent_plate = simulated_protocol_context.load_labware( "opentrons_96_wellplate_200ul_pcr_full_skirt", "A2" ) accessible_plate = thermocycler.load_labware( "opentrons_96_wellplate_200ul_pcr_full_skirt" ) - instrument = protocol_context.load_instrument("flex_96channel_1000", mount="left") + instrument = simulated_protocol_context.load_instrument( + "flex_96channel_1000", mount="left" + ) instrument.trash_container = trash_labware # ############ SHORT LABWARE ################ # These labware should be to the west of tall labware to avoid any partial tip deck conflicts - badly_placed_labware = protocol_context.load_labware( + badly_placed_labware = simulated_protocol_context.load_labware( "nest_96_wellplate_200ul_flat", "D2" ) - well_placed_labware = protocol_context.load_labware( + well_placed_labware = simulated_protocol_context.load_labware( "nest_96_wellplate_200ul_flat", "D3" ) # ############ TALL LABWARE ############## - protocol_context.load_labware( + simulated_protocol_context.load_labware( "opentrons_10_tuberack_falcon_4x50ml_6x15ml_conical", "D1" ) @@ -104,24 +109,30 @@ def test_deck_conflicts_for_96_ch_a12_column_configuration() -> None: @pytest.mark.ot3_only -def test_close_shave_deck_conflicts_for_96_ch_a12_column_configuration() -> None: +@pytest.mark.parametrize( + "simulated_protocol_context", [("2.20", "Flex")], indirect=True +) +def test_close_shave_deck_conflicts_for_96_ch_a12_column_configuration( + simulated_protocol_context: ProtocolContext, +) -> None: """Shouldn't raise errors for "almost collision"s.""" - protocol_context = simulate.get_protocol_api(version="2.20", robot_type="Flex") - res12 = protocol_context.load_labware("nest_12_reservoir_15ml", "C3") + res12 = simulated_protocol_context.load_labware("nest_12_reservoir_15ml", "C3") # Mag block and tiprack adapter are very close to the destination reservoir labware - protocol_context.load_module("magneticBlockV1", "D2") - protocol_context.load_labware( + simulated_protocol_context.load_module("magneticBlockV1", "D2") + simulated_protocol_context.load_labware( "opentrons_flex_96_tiprack_200ul", "B3", adapter="opentrons_flex_96_tiprack_adapter", ) - tiprack_8 = protocol_context.load_labware("opentrons_flex_96_tiprack_200ul", "B2") - hs = protocol_context.load_module("heaterShakerModuleV1", "C1") + tiprack_8 = simulated_protocol_context.load_labware( + "opentrons_flex_96_tiprack_200ul", "B2" + ) + hs = simulated_protocol_context.load_module("heaterShakerModuleV1", "C1") hs_adapter = hs.load_adapter("opentrons_96_deep_well_adapter") deepwell = hs_adapter.load_labware("nest_96_wellplate_2ml_deep") - protocol_context.load_trash_bin("A3") - p1000_96 = protocol_context.load_instrument("flex_96channel_1000") + simulated_protocol_context.load_trash_bin("A3") + p1000_96 = simulated_protocol_context.load_instrument("flex_96channel_1000") p1000_96.configure_nozzle_layout(style=SINGLE, start="A12", tip_racks=[tiprack_8]) hs.close_labware_latch() # type: ignore[union-attr] @@ -135,16 +146,28 @@ def test_close_shave_deck_conflicts_for_96_ch_a12_column_configuration() -> None @pytest.mark.ot3_only -def test_deck_conflicts_for_96_ch_a1_column_configuration() -> None: +@pytest.mark.parametrize( + "simulated_protocol_context", [("2.16", "Flex")], indirect=True +) +def test_deck_conflicts_for_96_ch_a1_column_configuration( + simulated_protocol_context: ProtocolContext, +) -> None: """It should raise errors for expected deck conflicts.""" - protocol = simulate.get_protocol_api(version="2.16", robot_type="Flex") - instrument = protocol.load_instrument("flex_96channel_1000", mount="left") - trash_labware = protocol.load_labware("opentrons_1_trash_3200ml_fixed", "A3") + instrument = simulated_protocol_context.load_instrument( + "flex_96channel_1000", mount="left" + ) + trash_labware = simulated_protocol_context.load_labware( + "opentrons_1_trash_3200ml_fixed", "A3" + ) instrument.trash_container = trash_labware - badly_placed_tiprack = protocol.load_labware("opentrons_flex_96_tiprack_50ul", "C2") - well_placed_tiprack = protocol.load_labware("opentrons_flex_96_tiprack_50ul", "A1") - tiprack_on_adapter = protocol.load_labware( + badly_placed_tiprack = simulated_protocol_context.load_labware( + "opentrons_flex_96_tiprack_50ul", "C2" + ) + well_placed_tiprack = simulated_protocol_context.load_labware( + "opentrons_flex_96_tiprack_50ul", "A1" + ) + tiprack_on_adapter = simulated_protocol_context.load_labware( "opentrons_flex_96_tiprack_50ul", "C3", adapter="opentrons_flex_96_tiprack_adapter", @@ -152,11 +175,15 @@ def test_deck_conflicts_for_96_ch_a1_column_configuration() -> None: # ############ SHORT LABWARE ################ # These labware should be to the east of tall labware to avoid any partial tip deck conflicts - badly_placed_plate = protocol.load_labware("nest_96_wellplate_200ul_flat", "B1") - well_placed_plate = protocol.load_labware("nest_96_wellplate_200ul_flat", "B3") + badly_placed_plate = simulated_protocol_context.load_labware( + "nest_96_wellplate_200ul_flat", "B1" + ) + well_placed_plate = simulated_protocol_context.load_labware( + "nest_96_wellplate_200ul_flat", "B3" + ) # ############ TALL LABWARE ############### - my_tuberack = protocol.load_labware( + my_tuberack = simulated_protocol_context.load_labware( "opentrons_10_tuberack_falcon_4x50ml_6x15ml_conical", "B2" ) @@ -208,7 +235,7 @@ def test_deck_conflicts_for_96_ch_a1_column_configuration() -> None: instrument.drop_tip() instrument.trash_container = None # type: ignore - protocol.load_trash_bin("C1") + simulated_protocol_context.load_trash_bin("C1") # This doesn't raise an error because it now treats the trash bin as an addressable area # and the bounds check doesn't yet check moves to addressable areas. @@ -229,28 +256,38 @@ def test_deck_conflicts_for_96_ch_a1_column_configuration() -> None: @pytest.mark.ot3_only -def test_deck_conflicts_for_96_ch_and_reservoirs() -> None: +@pytest.mark.parametrize( + "simulated_protocol_context", [("2.20", "Flex")], indirect=True +) +def test_deck_conflicts_for_96_ch_and_reservoirs( + simulated_protocol_context: ProtocolContext, +) -> None: """It should raise errors for expected deck conflicts when moving to reservoirs. This test checks that the critical point of the pipette is taken into account, specifically when it differs from the primary nozzle. """ - protocol = simulate.get_protocol_api(version="2.20", robot_type="Flex") - instrument = protocol.load_instrument("flex_96channel_1000", mount="left") + instrument = simulated_protocol_context.load_instrument( + "flex_96channel_1000", mount="left" + ) # trash_labware = protocol.load_labware("opentrons_1_trash_3200ml_fixed", "A3") # instrument.trash_container = trash_labware - protocol.load_trash_bin("A3") - right_tiprack = protocol.load_labware("opentrons_flex_96_tiprack_50ul", "C3") - front_tiprack = protocol.load_labware("opentrons_flex_96_tiprack_50ul", "D2") + simulated_protocol_context.load_trash_bin("A3") + right_tiprack = simulated_protocol_context.load_labware( + "opentrons_flex_96_tiprack_50ul", "C3" + ) + front_tiprack = simulated_protocol_context.load_labware( + "opentrons_flex_96_tiprack_50ul", "D2" + ) # Tall deck item in B3 - protocol.load_labware( + simulated_protocol_context.load_labware( "opentrons_flex_96_tiprack_50ul", "B3", adapter="opentrons_flex_96_tiprack_adapter", ) # Tall deck item in B1 - protocol.load_labware( + simulated_protocol_context.load_labware( "opentrons_flex_96_tiprack_50ul", "B1", adapter="opentrons_flex_96_tiprack_adapter", @@ -258,8 +295,12 @@ def test_deck_conflicts_for_96_ch_and_reservoirs() -> None: # ############ RESERVOIRS ################ # These labware should be to the east of tall labware to avoid any partial tip deck conflicts - reservoir_1_well = protocol.load_labware("nest_1_reservoir_195ml", "C2") - reservoir_12_well = protocol.load_labware("nest_12_reservoir_15ml", "B2") + reservoir_1_well = simulated_protocol_context.load_labware( + "nest_1_reservoir_195ml", "C2" + ) + reservoir_12_well = simulated_protocol_context.load_labware( + "nest_12_reservoir_15ml", "B2" + ) # ########### Use COLUMN A1 Config ############# instrument.configure_nozzle_layout(style=COLUMN, start="A1") diff --git a/api/tests/opentrons/protocol_api_integration/test_trashes.py b/api/tests/opentrons/protocol_api_integration/test_trashes.py index 18dfa62170d..1166ba01c70 100644 --- a/api/tests/opentrons/protocol_api_integration/test_trashes.py +++ b/api/tests/opentrons/protocol_api_integration/test_trashes.py @@ -1,46 +1,42 @@ """Tests for the APIs around waste chutes and trash bins.""" -from opentrons import protocol_api, simulate +from opentrons import protocol_api from opentrons.protocols.api_support.types import APIVersion from opentrons.protocols.api_support.util import UnsupportedAPIError import contextlib from typing import ContextManager, Optional, Type -from typing_extensions import Literal import re import pytest @pytest.mark.parametrize( - ("version", "robot_type", "expected_trash_class"), + ("simulated_protocol_context", "expected_trash_class"), [ - ("2.13", "OT-2", protocol_api.Labware), - ("2.14", "OT-2", protocol_api.Labware), - ("2.15", "OT-2", protocol_api.Labware), + (("2.13", "OT-2"), protocol_api.Labware), + (("2.14", "OT-2"), protocol_api.Labware), + (("2.15", "OT-2"), protocol_api.Labware), pytest.param( - "2.15", - "Flex", + ("2.15", "Flex"), protocol_api.Labware, marks=pytest.mark.ot3_only, # Simulating a Flex protocol requires a Flex hardware API. ), pytest.param( - "2.16", - "OT-2", + ("2.16", "OT-2"), protocol_api.TrashBin, ), pytest.param( - "2.16", - "Flex", + ("2.16", "Flex"), None, marks=pytest.mark.ot3_only, # Simulating a Flex protocol requires a Flex hardware API. ), ], + indirect=["simulated_protocol_context"], ) def test_fixed_trash_presence( - robot_type: Literal["OT-2", "Flex"], - version: str, + simulated_protocol_context: protocol_api.ProtocolContext, expected_trash_class: Optional[Type[object]], ) -> None: """Test the presence of the fixed trash. @@ -49,9 +45,10 @@ def test_fixed_trash_presence( For those that do, ProtocolContext.fixed_trash and InstrumentContext.trash_container should point to it. The type of the object depends on the API version. """ - protocol = simulate.get_protocol_api(version=version, robot_type=robot_type) - instrument = protocol.load_instrument( - "p300_single_gen2" if robot_type == "OT-2" else "flex_1channel_50", + instrument = simulated_protocol_context.load_instrument( + "p300_single_gen2" + if simulated_protocol_context._core.robot_type == "OT-2 Standard" + else "flex_1channel_50", mount="left", ) @@ -59,46 +56,53 @@ def test_fixed_trash_presence( with pytest.raises( UnsupportedAPIError, match=re.escape( - "Error 4002 API_REMOVED (UnsupportedAPIError): Fixed Trash is not available after API version 2.16. You are currently using API version 2.16. Fixed trash is no longer supported on Flex protocols." + "Error 4002 API_REMOVED (UnsupportedAPIError): Fixed Trash is not available after API version 2.16." + " You are currently using API version 2.16. Fixed trash is no longer supported on Flex protocols." ), ): - protocol.fixed_trash + simulated_protocol_context.fixed_trash with pytest.raises(Exception, match="No trash container has been defined"): instrument.trash_container else: - assert isinstance(protocol.fixed_trash, expected_trash_class) - assert instrument.trash_container is protocol.fixed_trash + assert isinstance(simulated_protocol_context.fixed_trash, expected_trash_class) + assert instrument.trash_container is simulated_protocol_context.fixed_trash @pytest.mark.ot3_only # Simulating a Flex protocol requires a Flex hardware API. -def test_trash_search() -> None: +@pytest.mark.parametrize( + "simulated_protocol_context", [("2.16", "Flex")], indirect=True +) +def test_trash_search(simulated_protocol_context: protocol_api.ProtocolContext) -> None: """Test the automatic trash search for protocols without a fixed trash.""" - protocol = simulate.get_protocol_api(version="2.16", robot_type="Flex") - instrument = protocol.load_instrument("flex_1channel_50", mount="left") + instrument = simulated_protocol_context.load_instrument( + "flex_1channel_50", mount="left" + ) # By default, there should be no trash. with pytest.raises( UnsupportedAPIError, match=re.escape( - "Error 4002 API_REMOVED (UnsupportedAPIError): Fixed Trash is not available after API version 2.16. You are currently using API version 2.16. Fixed trash is no longer supported on Flex protocols." + "Error 4002 API_REMOVED (UnsupportedAPIError): Fixed Trash is not available after API version 2.16." + " You are currently using API version 2.16. Fixed trash is no longer supported on Flex protocols." ), ): - protocol.fixed_trash + simulated_protocol_context.fixed_trash with pytest.raises(Exception, match="No trash container has been defined"): instrument.trash_container - loaded_first = protocol.load_trash_bin("A1") - loaded_second = protocol.load_trash_bin("B1") + loaded_first = simulated_protocol_context.load_trash_bin("A1") + loaded_second = simulated_protocol_context.load_trash_bin("B1") # After loading some trashes, there should still be no protocol.fixed_trash... with pytest.raises( UnsupportedAPIError, match=re.escape( - "Error 4002 API_REMOVED (UnsupportedAPIError): Fixed Trash is not available after API version 2.16. You are currently using API version 2.16. Fixed trash is no longer supported on Flex protocols." + "Error 4002 API_REMOVED (UnsupportedAPIError): Fixed Trash is not available after API version 2.16." + " You are currently using API version 2.16. Fixed trash is no longer supported on Flex protocols." ), ): - protocol.fixed_trash + simulated_protocol_context.fixed_trash # ...but instrument.trash_container should automatically update to point to # the first trash that we loaded. assert instrument.trash_container is loaded_first @@ -109,40 +113,36 @@ def test_trash_search() -> None: @pytest.mark.parametrize( - ("version", "robot_type", "expect_load_to_succeed"), + ("simulated_protocol_context", "expect_load_to_succeed"), [ pytest.param( - "2.13", - "OT-2", + ("2.13", "OT-2"), False, # This xfail (the system does let you load a labware onto slot 12, and does not raise) # is surprising to me. It may be be a bug in old PAPI versions. marks=pytest.mark.xfail(strict=True, raises=pytest.fail.Exception), ), - ("2.14", "OT-2", False), - ("2.15", "OT-2", False), + (("2.14", "OT-2"), False), + (("2.15", "OT-2"), False), pytest.param( - "2.15", - "Flex", + ("2.15", "Flex"), False, marks=pytest.mark.ot3_only, # Simulating a Flex protocol requires a Flex hardware API. ), pytest.param( - "2.16", - "OT-2", + ("2.16", "OT-2"), False, ), pytest.param( - "2.16", - "Flex", + ("2.16", "Flex"), True, marks=pytest.mark.ot3_only, # Simulating a Flex protocol requires a Flex hardware API. ), ], + indirect=["simulated_protocol_context"], ) def test_fixed_trash_load_conflicts( - robot_type: Literal["Flex", "OT-2"], - version: str, + simulated_protocol_context: protocol_api.ProtocolContext, expect_load_to_succeed: bool, ) -> None: """Test loading something onto the location historically used for the fixed trash. @@ -150,14 +150,12 @@ def test_fixed_trash_load_conflicts( In configurations where there is a fixed trash, this should be disallowed. In configurations without a fixed trash, this should be allowed. """ - protocol = simulate.get_protocol_api(version=version, robot_type=robot_type) - if expect_load_to_succeed: expected_error: ContextManager[object] = contextlib.nullcontext() else: # If we're expecting an error, it'll be a LocationIsOccupied for 2.15 and below, otherwise # it will fail with an IncompatibleAddressableAreaError, since slot 12 will not be in the deck config - if APIVersion.from_string(version) < APIVersion(2, 16): + if simulated_protocol_context.api_version < APIVersion(2, 16): error_name = "LocationIsOccupiedError" else: error_name = "IncompatibleAddressableAreaError" @@ -169,4 +167,6 @@ def test_fixed_trash_load_conflicts( ) with expected_error: - protocol.load_labware("opentrons_96_wellplate_200ul_pcr_full_skirt", 12) + simulated_protocol_context.load_labware( + "opentrons_96_wellplate_200ul_pcr_full_skirt", 12 + )