fixup: revert demo-changes to Protocol API, add task runner

api/src/opentrons/file_runner/abstract_file_runner.py

@@ -12,7 +12,7 @@ def load(self) -> None:
 
     @abstractmethod
     async def run(self) -> None:
-        """Start running the protocol file."""
+        """Run the protocol file to completion."""
         ...
 
     @abstractmethod

api/src/opentrons/file_runner/json_file_runner.py

@@ -43,7 +43,8 @@ def load(self) -> None:
 
     async def run(self) -> None:
         """Run the protocol to completion."""
-        # TODO(mc, 2021-06-30): do not merge
+        # TODO(mc, 2021-06-30): this will not work with pause/resume. Rework
+        # so that queue worker has `wait_for_done` method, instead.
         self.play()
         await self._command_queue_worker.wait_to_be_idle()
 

api/src/opentrons/file_runner/python_file_runner.py

@@ -33,7 +33,7 @@ async def run(self) -> None:
         await self._executor.execute()
 
     def play(self) -> None:
-        """Resumt running the Python protocol file."""
+        """Resume running the Python protocol file after a pause."""
         raise NotImplementedError()
 
     def pause(self) -> None:

api/src/opentrons/protocol_api_experimental/labware.py

@@ -23,19 +23,8 @@ def __init__(
             engine_client: A client to access protocol state.
             labware_id: The labware's identifier in commands and protocol state.
         """
-        definition = engine_client.state.labware.get_labware_definition(labware_id)
-
         self._engine_client = engine_client
         self._labware_id = labware_id
-        self._wells_by_name: Dict[str, Well] = {
-            well_name: Well(
-                well_name=well_name,
-                engine_client=engine_client,
-                labware=self,
-            )
-            for row in definition.ordering
-            for well_name in row
-        }
 
     # TODO(mc, 2021-04-22): remove this property; it's redundant and
     # unlikely to be used by PAPI users
@@ -119,7 +108,7 @@ def quirks(self) -> List[str]:  # noqa: D102
     # operational logic, and its presence in this interface is no longer
     # necessary with Protocol Engine controlling execution. Can we get rid of it?
     @property
-    def magdeck_engage_height(self) -> Optional[float]:  # noqa: D102
+    def magdeck_engage_height(self) -> Optional[float]:    # noqa: D102
         definition = self._engine_client.state.labware.get_labware_definition(
             labware_id=self._labware_id
         )
@@ -179,13 +168,13 @@ def tip_length(self) -> float:
         return definition.parameters.tipLength
 
     def well(self, idx: int) -> Well:  # noqa: D102
-        return self.wells()[idx]
+        raise NotImplementedError()
 
     def wells(self) -> List[Well]:  # noqa: D102
-        return list(self._wells_by_name.values())
+        raise NotImplementedError()
 
     def wells_by_name(self) -> Dict[str, Well]:  # noqa: D102
-        return self._wells_by_name
+        raise NotImplementedError()
 
     def rows(self) -> List[List[Well]]:  # noqa: D102
         raise NotImplementedError()

api/tests/opentrons/data/testosaur_v2.py

@@ -1,7 +1,7 @@
 from opentrons import types
 
 metadata = {
-    "protocolName": "Testosaur Version 2",
+    "protocolName": "Testosaur",
     "author": "Opentrons <[email protected]>",
     "description": 'A variant on "Dinosaur" for testing',
     "source": "Opentrons Repository",

api/tests/opentrons/data/testosaur_v3.py

@@ -17,6 +17,7 @@ def run(ctx):
     pipette = ctx.load_instrument("p300_single_gen2", types.Mount.RIGHT, [])
 
     for i in range(4):
+        pipette.pick_up_tip(tip_rack.well(i))
         pipette.pick_up_tip(tip_rack.well(i))
         pipette.aspirate(50, source.wells_by_name()["A1"])
         pipette.dispense(50, dest.well(i))

robot-server/robot_server/service/task_runner.py

@@ -0,0 +1,66 @@
+"""Background task management.
+
+This mosule is mostly a thin wrapper around fastapi.BackgroundTasks
+that adds logging. It should be tested primarly through integration
+and end-to-end tests.
+"""
+from asyncio import iscoroutinefunction
+from fastapi import BackgroundTasks
+from logging import getLogger
+from typing import Awaitable, Callable, Union
+
+log = getLogger(__name__)
+
+
+TaskFunc = Union[
+    Callable[[], Awaitable[None]],
+    Callable[[], None],
+]
+
+
+class TaskRunner:
+    def __init__(self, background_tasks: BackgroundTasks) -> None:
+        """Initialize the TaskRunner.
+
+        Add to any route handler with `FastAPI.Depends`/
+
+        Arguments:
+            background_tasks: FastAPI's background task system, fed in
+                automatically by FastAPI's dependency injection system.
+        """
+        self._background_tasks = background_tasks
+
+    def run(self, func: TaskFunc) -> None:
+        """Run a function in the background.
+
+        Will log when the function completes, including any error
+        that may occur.
+
+        Arguments:
+            func: A argumentless, None-returing function to run the in
+                the background. Use functools.partial to add arguments,
+                if required.
+        """
+        func_name = func.__name__
+
+        async def _run_async_task() -> None:
+            try:
+                await func()  # type: ignore[misc]
+                log.debug(f"Backgound task {func_name} succeeded")
+            except Exception as e:
+                log.warning(f"Backgound task {func_name} failed", exc_info=e)
+
+        def _run_sync_task() -> None:
+            try:
+                func()
+                log.debug(f"Backgound task {func_name} succeeded")
+            except Exception as e:
+                log.warning(f"Backgound task {func_name} failed", exc_info=e)
+
+        # NOTE: FastAPI will run async background tasks differently than
+        # sync background tasks (a threadpool is involved). Ensure we
+        # maintain the asynchronicity of the original function
+        if iscoroutinefunction(func):
+            self._background_tasks.add_task(_run_async_task)
+        else:
+            self._background_tasks.add_task(_run_sync_task)

robot-server/robot_server/sessions/router.py

@@ -1,14 +1,14 @@
 """Router for /sessions endpoints."""
-from fastapi import APIRouter, BackgroundTasks, Depends, status
+from fastapi import APIRouter, Depends, status
 from datetime import datetime
 from typing import Optional, Union
 from typing_extensions import Literal
 
 from opentrons.protocol_engine import commands as pe_commands, errors as pe_errors
 
 from robot_server.errors import ErrorDetails, ErrorResponse
-
 from robot_server.service.dependencies import get_current_time, get_unique_id
+from robot_server.service.task_runner import TaskRunner
 from robot_server.service.json_api import (
     RequestModel,
     ResponseModel,
@@ -232,25 +232,25 @@ async def remove_session_by_id(
 )
 async def create_session_action(
     sessionId: str,
-    background_tasks: BackgroundTasks,
     request_body: RequestModel[SessionActionCreateData],
     session_view: SessionView = Depends(SessionView),
     session_store: SessionStore = Depends(get_session_store),
     engine_store: EngineStore = Depends(get_engine_store),
     action_id: str = Depends(get_unique_id),
     created_at: datetime = Depends(get_current_time),
+    task_runner: TaskRunner = Depends(TaskRunner),
 ) -> ResponseModel[SessionAction]:
     """Create a session control action.
 
     Arguments:
         sessionId: Session ID pulled from the URL.
-        background_tasks: FastAPI background task manager.
         request_body: Input payload from the request body.
         session_view: Resource model builder.
         session_store: Session storage interface.
         engine_store: Protocol engine and runner storage.
         action_id: Generated ID to assign to the control action.
         created_at: Timestamp to attach to the control action.
+        task_runner: Background task runner.
     """
     try:
         prev_session = session_store.get(session_id=sessionId)
@@ -267,7 +267,7 @@ async def create_session_action(
         # before starting the protocol run
         # TODO(mc, 2021-06-30): capture errors (e.g. uncaught Python raise)
         # and place them in the session response
-        background_tasks.add_task(engine_store.runner.run)
+        task_runner.run(engine_store.runner.run)
 
     except SessionNotFoundError as e:
         raise SessionNotFound(detail=str(e)).as_error(status.HTTP_404_NOT_FOUND)

robot-server/tests/sessions/conftest.py

@@ -5,12 +5,19 @@
 from pathlib import Path
 from decoy import Decoy
 
+from robot_server.service.task_runner import TaskRunner
 from robot_server.protocols import ProtocolStore
 from robot_server.sessions.session_view import SessionView
 from robot_server.sessions.session_store import SessionStore
 from robot_server.sessions.engine_store import EngineStore
 
 
+@pytest.fixture
+def task_runner(decoy: Decoy) -> TaskRunner:
+    """Get a mock background TaskRunner."""
+    return decoy.create_decoy(spec=TaskRunner)
+
+
 @pytest.fixture
 def protocol_store(decoy: Decoy) -> ProtocolStore:
     """Get a mock ProtocolStore interface."""

robot-server/tests/sessions/test_engine_store.py

@@ -59,7 +59,6 @@ async def test_create_engine_for_json_protocol(
     assert isinstance(subject.runner, JsonFileRunner)
 
 
-@pytest.mark.xfail(raises=NotImplementedError, strict=True)
 async def test_create_engine_for_python_protocol(
     subject: EngineStore,
     python_protocol_file: Path,

robot-server/tests/sessions/test_sessions_router.py

@@ -15,6 +15,9 @@
 )
 
 from robot_server.errors import exception_handlers
+
+from robot_server.service.task_runner import TaskRunner
+
 from robot_server.protocols import (
     ProtocolStore,
     ProtocolResource,
@@ -34,11 +37,7 @@
     SessionCommandSummary,
 )
 
-from robot_server.sessions.engine_store import (
-    EngineStore,
-    EngineConflictError,
-    EngineMissingError,
-)
+from robot_server.sessions.engine_store import EngineStore, EngineConflictError
 
 from robot_server.sessions.session_store import (
     SessionStore,
@@ -70,6 +69,7 @@
 
 @pytest.fixture
 def app(
+    task_runner: TaskRunner,
     session_store: SessionStore,
     session_view: SessionView,
     engine_store: EngineStore,
@@ -79,6 +79,7 @@ def app(
 ) -> FastAPI:
     """Get a FastAPI app with /sessions routes and mocked-out dependencies."""
     app = FastAPI(exception_handlers=exception_handlers)
+    app.dependency_overrides[TaskRunner] = lambda: task_runner
     app.dependency_overrides[SessionView] = lambda: session_view
     app.dependency_overrides[get_session_store] = lambda: session_store
     app.dependency_overrides[get_engine_store] = lambda: engine_store
@@ -442,6 +443,7 @@ def test_delete_session_with_bad_id(
 
 def test_create_session_action(
     decoy: Decoy,
+    task_runner: TaskRunner,
     session_view: SessionView,
     session_store: SessionStore,
     engine_store: EngineStore,
@@ -489,7 +491,7 @@ def test_create_session_action(
     )
 
     verify_response(response, expected_status=201, expected_data=actions)
-    decoy.verify(engine_store.runner.play())
+    decoy.verify(task_runner.run(engine_store.runner.run))
 
 
 def test_create_session_action_with_missing_id(
@@ -516,6 +518,7 @@ def test_create_session_action_with_missing_id(
     )
 
 
+@pytest.mark.xfail(strict=True)
 def test_create_session_action_without_runner(
     decoy: Decoy,
     session_view: SessionView,
@@ -559,7 +562,12 @@ def test_create_session_action_without_runner(
         ),
     ).then_return((actions, next_session))
 
-    decoy.when(engine_store.runner.play()).then_raise(EngineMissingError("oh no"))
+    # TODO(mc, 2021-07-06): in reality, it will be the engine_store.runner
+    # property access that triggers this raise. Explore adding property access
+    # rehearsals to decoy
+    # decoy.when(
+    #     await engine_store.runner.run()
+    # ).then_raise(EngineMissingError("oh no"))
 
     response = client.post(
         "/sessions/session-id/actions",