test(api): Configure ctx fixture as PAPIv2.14 when it's configured as an OT-3

api/pytest.ini

@@ -1,11 +1,6 @@
 [pytest]
 markers =
-        api1_only: Test only functions using API version 1 (legacy_api)
-        api2_only: Test only functions using API version 2 (protocol API and hardware control)
-        model1: Marks for functions using gen1 pipettes in deck cal cli tests
-        model2: Marks for functions using gen2 pipettes in deck cal cli tests
-        apiv1: This test invocation requires apiv1
-        apiv2: This test invocation requires apiv2
+        apiv2_non_pe_only: This test invocation requires a legacy PAPI context, not backed by Protocol Engine
         ot2_only: Test only functions using the OT2 hardware
         ot3_only: Test only functions using the OT3 hardware
 addopts = --color=yes

api/tests/opentrons/async_context_manager_in_thread.py

@@ -0,0 +1,98 @@
+"""A test helper to enter an async context manager in a worker thread."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import queue
+import typing
+
+from concurrent.futures import ThreadPoolExecutor
+
+
+_T = typing.TypeVar("_T")
+
+
+@contextlib.contextmanager
+def async_context_manager_in_thread(
+    async_context_manager: typing.AsyncContextManager[_T],
+) -> typing.Generator[typing.Tuple[_T, asyncio.AbstractEventLoop], None, None]:
+    """Enter an async context manager in a worker thread.
+
+    When you enter this context manager, it:
+
+    1. Spawns a worker thread.
+    2. In that thread, starts an asyncio event loop.
+    3. In that event loop, enters the context manager that you passed in.
+    4. Returns: the result of entering that context manager, and the running event loop.
+       Use functions like `asyncio.run_coroutine_threadsafe()` to safely interact
+       with the returned object from your thread.
+
+    When you exit this context manager, it:
+
+    1. In the worker thread's event loop, exits the context manager that you passed in.
+    2. Stops and cleans up the worker thread's event loop.
+    3. Joins the worker thread.
+    """
+    with _run_loop_in_thread() as loop_in_thread:
+        async_object = asyncio.run_coroutine_threadsafe(
+            async_context_manager.__aenter__(),
+            loop=loop_in_thread,
+        ).result()
+
+        try:
+            yield async_object, loop_in_thread
+
+        finally:
+            exit = asyncio.run_coroutine_threadsafe(
+                async_context_manager.__aexit__(None, None, None),
+                loop=loop_in_thread,
+            )
+            exit.result()
+
+
+@contextlib.contextmanager
+def _run_loop_in_thread() -> typing.Generator[asyncio.AbstractEventLoop, None, None]:
+    """Run an event loop in a worker thread.
+
+    Entering this context manager spawns a thread, starts an asyncio event loop in it,
+    and returns that loop.
+
+    Exiting this context manager stops and cleans up the event loop, and then joins the thread.
+    """
+    loop_queue: "queue.SimpleQueue[asyncio.AbstractEventLoop]" = queue.SimpleQueue()
+
+    def _in_thread() -> None:
+        loop = asyncio.new_event_loop()
+
+        # We assume that the lines above this will never fail,
+        # so we will always reach this point to unblock the parent thread.
+        loop_queue.put(loop)
+
+        loop.run_forever()
+
+        # If we've reached here, the loop has been stopped from outside this thread. Clean it up.
+        #
+        # This cleanup is naive because asyncio makes it difficult and confusing to get it right.
+        # Compare this with asyncio.run()'s cleanup, which:
+        #
+        # * Cancels and awaits any remaining tasks
+        #   (according to the source code--this seems undocumented)
+        # * Shuts down asynchronous generators
+        #   (see asyncio.shutdown_asyncgens())
+        # * Shuts down the default thread pool executor
+        #   (see https://bugs.python.org/issue34037 and asyncio.shutdown_default_executor())
+        #
+        # In Python >=3.11, we should rewrite this to use asyncio.Runner,
+        # which can take care of these nuances for us.
+        loop.close()
+
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        executor.submit(_in_thread)
+
+        loop_in_thread = loop_queue.get()
+
+        try:
+            yield loop_in_thread
+        finally:
+            loop_in_thread.call_soon_threadsafe(loop_in_thread.stop)

api/tests/opentrons/conftest.py

@@ -53,6 +53,8 @@
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.types import Location, Point
 
+from .protocol_engine_in_thread import protocol_engine_in_thread
+
 
 if TYPE_CHECKING:
     from opentrons.drivers.smoothie_drivers import SmoothieDriver as SmoothieDriverType
@@ -255,11 +257,42 @@ async def hardware(
         yield hw
 
 
-@pytest.fixture()
-def ctx(hardware: ThreadManagedHardware) -> ProtocolContext:
+def _make_ot2_non_pe_ctx(hardware: ThreadManagedHardware) -> ProtocolContext:
+    """Return a ProtocolContext configured for an OT-2 and not backed by Protocol Engine."""
     return create_protocol_context(api_version=APIVersion(2, 13), hardware_api=hardware)
 
 
+@contextlib.contextmanager
+def _make_ot3_pe_ctx(
+    hardware: ThreadManagedHardware,
+) -> Generator[ProtocolContext, None, None]:
+    """Return a ProtocolContext configured for an OT-3 and backed by Protocol Engine."""
+    with protocol_engine_in_thread(hardware=hardware) as (engine, loop):
+        yield create_protocol_context(
+            api_version=APIVersion(2, 14),
+            hardware_api=hardware,
+            protocol_engine=engine,
+            # TODO will this deadlock?
+            protocol_engine_loop=loop,
+        )
+
+
+@pytest.fixture()
+def ctx(
+    request: pytest.FixtureRequest,
+    robot_model: RobotModel,
+    hardware: ThreadManagedHardware,
+) -> Generator[ProtocolContext, None, None]:
+    if robot_model == "OT-2 Standard":
+        yield _make_ot2_non_pe_ctx(hardware=hardware)
+    elif robot_model == "OT-3 Standard":
+        if request.node.get_closest_marker("apiv2_non_pe_only"):
+            pytest.skip("Test requests only non-Protocol-Engine ProtocolContexts")
+        else:
+            with _make_ot3_pe_ctx(hardware=hardware) as ctx:
+                yield ctx
+
+
 @pytest.fixture()
 async def smoothie(
     virtual_smoothie_env: None,

api/tests/opentrons/protocol_api_old/test_labware_load.py

@@ -6,6 +6,10 @@
 labware_name = "corning_96_wellplate_360ul_flat"
 
 
+# labware._core.get_geometry() is an implementation detail and is not present in ProtocolContexts
+# backed by Protocol Engine.
+# TODO(mm, 2022-04-28): Make sure this logic is tested elsewhere, then delete this test.
+@pytest.mark.apiv2_non_pe_only
 def test_load_to_slot(
     ctx: papi.ProtocolContext, deck_definition: DeckDefinitionV3
 ) -> None:

api/tests/opentrons/protocol_api_old/test_module_context.py

@@ -145,6 +145,10 @@ def test_incorrect_module_error(ctx_with_tempdeck):
         assert ctx_with_tempdeck.load_module("the cool module", 1)
 
 
+# TODO(mm, 2023-04-28): This test uses mod.geometry, which was always a quasi-implementation-detail
+# and was removed in PAPIv2.14. We should make sure this logic is adequately covered elsewhere
+# in a way that doesn't depend on mod.geometry, and then delete this test.
+@pytest.mark.apiv2_non_pe_only
 @pytest.mark.parametrize(
     "loadname,klass,model",
     [
@@ -243,6 +247,7 @@ def test_thermocycler_profile_no_hold(ctx_with_thermocycler, mock_module_control
         )
 
 
+@pytest.mark.apiv2_non_pe_only  # Semi plate configuration removed in PAPIv2.14.
 def test_thermocycler_semi_plate_configuration(ctx):
     labware_name = "nest_96_wellplate_100ul_pcr_full_skirt"
     mod = ctx.load_module("thermocycler", configuration="semi")
@@ -411,6 +416,7 @@ def test_deprecated_module_load_labware_by_name(ctx_with_tempdeck):
     )
 
 
+@pytest.mark.apiv2_non_pe_only  # engage(height=...) param was removed in PAPIv2.14.
 async def test_magdeck_gen1_labware_props(ctx):
     # TODO Ian 2019-05-29 load fixtures, not real defs
     labware_name = "biorad_96_wellplate_200ul_pcr"
@@ -454,6 +460,7 @@ async def test_magdeck_gen1_labware_props(ctx):
     )
 
 
+@pytest.mark.apiv2_non_pe_only  # engage(height=...) param was removed in PAPIv2.14.
 def test_magdeck_gen2_labware_props(ctx):
     mod = ctx.load_module("magnetic module gen2", 1)
     mod.engage(height=25)

api/tests/opentrons/protocol_engine_in_thread.py

@@ -0,0 +1,62 @@
+"""Run a `ProtocolEngine` in a worker thread."""
+
+import asyncio
+import contextlib
+import typing
+
+from opentrons.hardware_control import ThreadManagedHardware
+from opentrons.protocol_engine import create_protocol_engine, ProtocolEngine, Config
+
+from .async_context_manager_in_thread import async_context_manager_in_thread
+
+
+@contextlib.contextmanager
+def protocol_engine_in_thread(
+    hardware: ThreadManagedHardware,
+) -> typing.Generator[
+    typing.Tuple[ProtocolEngine, asyncio.AbstractEventLoop], None, None
+]:
+    """Run a `ProtocolEngine` in a worker thread.
+
+    When this context manager is entered, it:
+
+    1. Starts a worker thread.
+    2. Starts an asyncio event loop in that worker thread.
+    3. Creates and `.play()`s a `ProtocolEngine` in that event loop.
+    4. Returns the `ProtocolEngine` and the event loop.
+       Use functions like `asyncio.run_coroutine_threadsafe()` to safely interact with
+       the `ProtocolEngine` from your thread.
+
+    When this context manager is exited, it:
+
+    1. Cleans up the `ProtocolEngine`.
+    2. Stops and cleans up the event loop.
+    3. Joins the thread.
+    """
+    with async_context_manager_in_thread(_protocol_engine(hardware)) as (
+        protocol_engine,
+        loop,
+    ):
+        yield protocol_engine, loop
+
+
+@contextlib.asynccontextmanager
+async def _protocol_engine(
+    hardware: ThreadManagedHardware,
+) -> typing.AsyncGenerator[ProtocolEngine, None]:
+    protocol_engine = await create_protocol_engine(
+        hardware_api=hardware.wrapped(),
+        config=Config(
+            robot_type="OT-3 Standard",
+            ignore_pause=True,
+            use_virtual_pipettes=True,
+            use_virtual_modules=True,
+            use_virtual_gripper=True,
+            block_on_door_open=False,
+        ),
+    )
+    try:
+        protocol_engine.play()
+        yield protocol_engine
+    finally:
+        await protocol_engine.finish()

api/tests/opentrons/protocols/api_support/test_instrument.py

@@ -45,6 +45,13 @@ def test_validate_takes_liquid(ctx: ProtocolContext, reject_module: bool) -> Non
         )
 
 
+# TODO(mm, 2023-04-28): The validate_takes_liquid() function is used both by ProtocolContexts
+# that are backed by Protocol Engine, and those that aren't. But this test is only runnable
+# with a non-Protocol-Engine ProtocolContext because it relies on the internal module.geometry
+# property.
+#
+# Find a different way to test this so that both paths are covered.
+@pytest.mark.apiv2_non_pe_only
 def test_validate_takes_liquid_module_location(ctx):
     module = ctx.load_module("magdeck", 1)
 

api/tests/opentrons/protocols/api_support/test_util.py

@@ -85,6 +85,13 @@ def test_build_edges():
     assert res2 == new_correct_edges
 
 
+# TODO(mm, 2023-04-28): The build_edges() function is used both by ProtocolContexts
+# that are backed by Protocol Engine, and those that aren't. But this test is only runnable
+# with a non-Protocol-Engine ProtocolContext because it relies on the internal ctx._core.get_deck()
+# property.
+#
+# Find a different way to test this so that both paths are covered.
+@pytest.mark.apiv2_non_pe_only
 def test_build_edges_left_pipette(ctx):
     test_lw = ctx.load_labware("corning_96_wellplate_360ul_flat", "2")
     test_lw2 = ctx.load_labware("corning_96_wellplate_360ul_flat", "6")
@@ -124,6 +131,8 @@ def test_build_edges_left_pipette(ctx):
     assert res2 == left_pip_edges
 
 
+# TODO(mm, 2023-04-28): See note on test_build_edges_left_pipette().
+@pytest.mark.apiv2_non_pe_only
 def test_build_edges_right_pipette(ctx):
     test_lw = ctx.load_labware("corning_96_wellplate_360ul_flat", "2")
     test_lw2 = ctx.load_labware("corning_96_wellplate_360ul_flat", "6")
@@ -141,7 +150,7 @@ def test_build_edges_right_pipette(ctx):
         test_lw["A1"],
         1.0,
         Mount.RIGHT,
-        ctx._core._deck_layout,
+        ctx._core.get_deck(),
         version=APIVersion(2, 4),
     )
     assert res == right_pip_edges