tests

Author: caila-marashaj

api/src/opentrons/protocol_engine/commands/aspirate_while_tracking.py

@@ -114,7 +114,6 @@ async def execute(self, params: AspirateWhileTrackingParams) -> _ExecuteReturn:
             labware_id=params.labwareId,
             well_name=params.wellName,
         )
-
         move_result = await move_to_well(
             movement=self._movement,
             model_utils=self._model_utils,

api/tests/opentrons/protocol_engine/commands/test_aspirate_while_tracking.py

@@ -4,6 +4,8 @@
 
 import pytest
 from decoy import Decoy, matchers
+from mock import AsyncMock, patch
+from typing import Iterator
 
 from opentrons_shared_data.errors.exceptions import PipetteOverpressureError
 
@@ -15,6 +17,7 @@
     GantryMover,
     MovementHandler,
 )
+from opentrons.protocol_engine.commands.movement_common import move_to_well
 from opentrons.protocol_engine.commands.aspirate_while_tracking import (
     AspirateWhileTrackingParams,
     AspirateWhileTrackingResult,
@@ -36,6 +39,7 @@
     WellOrigin,
     WellOffset,
     DeckPoint,
+    WellLocation,
 )
 from opentrons.protocol_engine.state import update_types
 
@@ -58,6 +62,12 @@ def pipetting(decoy: Decoy) -> PipettingHandler:
     return decoy.mock(cls=PipettingHandler)
 
 
+@pytest.fixture
+def movement(decoy: Decoy) -> MovementHandler:
+    """Get a mock in the shape of a MovementHandler."""
+    return decoy.mock(cls=MovementHandler)
+
+
 @pytest.fixture
 def subject(
     pipetting: PipettingHandler,
@@ -87,11 +97,11 @@ def subject(
         (
             CurrentWell(
                 pipette_id="pipette-id-abc",
-                labware_id="labware-id-1",
-                well_name="well-name-1",
+                labware_id="funky-labware",
+                well_name="funky-well",
             ),
-            "labware-id-1",
-            "well-name-1",
+            "funky-labware",
+            "funky-well",
         ),
         (
             CurrentAddressableArea("pipette-id-abc", "addressable-area-1"),
@@ -161,6 +171,26 @@ async def test_aspirate_while_tracking_implementation(
 
     decoy.when(state_store.pipettes.get_current_location()).then_return(location)
 
+    _well_location = LiquidHandlingWellLocation(
+        origin=WellOrigin.MENISCUS, offset=WellOffset(x=0.0, y=0.0, z=1.0)
+    )
+    _current_well = CurrentWell(
+        pipette_id="pipette-id-abc", labware_id="funky-labware", well_name="funky-well"
+    )
+    decoy.when(
+        await subject._movement.move_to_well(
+            pipette_id="pipette-id-abc",
+            labware_id="funky-labware",
+            well_name="funky-well",
+            well_location=_well_location,
+            current_well=_current_well,
+            force_direct=False,
+            minimum_z_height=None,
+            speed=None,
+            operation_volume=-123.0,
+        ),
+    ).then_return(Point(x=4, y=5, z=6))
+
     result = await subject.execute(params=data)
 
     if isinstance(location, CurrentWell):
@@ -272,6 +302,26 @@ async def test_aspirate_raises_volume_error(
         )
     ).then_raise(AssertionError("blah blah"))
 
+    _well_location = LiquidHandlingWellLocation(
+        origin=WellOrigin.MENISCUS, offset=WellOffset(x=0.0, y=0.0, z=1.0)
+    )
+    _current_well = CurrentWell(
+        pipette_id="pipette-id-abc", labware_id="funky-labware", well_name="funky-well"
+    )
+    decoy.when(
+        await subject._movement.move_to_well(
+            pipette_id="pipette-id-abc",
+            labware_id="funky-labware",
+            well_name="funky-well",
+            well_location=_well_location,
+            current_well=_current_well,
+            force_direct=False,
+            minimum_z_height=None,
+            speed=None,
+            operation_volume=-50.0,
+        ),
+    ).then_return(Point(x=4, y=5, z=6))
+
     with pytest.raises(AssertionError):
         await subject.execute(data)
 
@@ -281,15 +331,15 @@ async def test_aspirate_raises_volume_error(
     [
         (
             CurrentWell(
-                pipette_id="pipette-id",
-                labware_id="labware-id-1",
-                well_name="well-name-1",
+                pipette_id="pipette-id-abc",
+                labware_id="funky-labware",
+                well_name="funky-well",
             ),
-            "labware-id-1",
-            "well-name-1",
+            "funky-labware",
+            "funky-well",
         ),
         (None, None, None),
-        (CurrentAddressableArea("pipette-id", "addressable-area-1"), None, None),
+        (CurrentAddressableArea("pipette-id-abc", "addressable-area-1"), None, None),
     ],
 )
 async def test_overpressure_error(
@@ -305,7 +355,7 @@ async def test_overpressure_error(
     stateupdateWell: str,
 ) -> None:
     """It should return an overpressure error if the hardware API indicates that."""
-    pipette_id = "pipette-id"
+    pipette_id = "pipette-id-abc"
 
     position = Point(x=1, y=2, z=3)
 
@@ -315,13 +365,13 @@ async def test_overpressure_error(
         state_store.geometry.get_nozzles_per_well(
             labware_id=stateupdateLabware,
             target_well_name=stateupdateWell,
-            pipette_id="pipette-id",
+            pipette_id="pipette-id-abc",
         )
     ).then_return(2)
 
     decoy.when(
         state_store.geometry.get_wells_covered_by_pipette_with_active_well(
-            stateupdateLabware, stateupdateWell, "pipette-id"
+            stateupdateLabware, stateupdateWell, "pipette-id-abc"
         )
     ).then_return(["A3", "A4"])
     well_location = LiquidHandlingWellLocation(
@@ -357,6 +407,26 @@ async def test_overpressure_error(
     decoy.when(await gantry_mover.get_position(pipette_id)).then_return(position)
     decoy.when(state_store.pipettes.get_current_location()).then_return(location)
 
+    _well_location = LiquidHandlingWellLocation(
+        origin=WellOrigin.MENISCUS, offset=WellOffset(x=0.0, y=0.0, z=1.0)
+    )
+    _current_well = CurrentWell(
+        pipette_id="pipette-id-abc", labware_id="funky-labware", well_name="funky-well"
+    )
+    decoy.when(
+        await subject._movement.move_to_well(
+            pipette_id=pipette_id,
+            labware_id="funky-labware",
+            well_name="funky-well",
+            well_location=_well_location,
+            current_well=_current_well,
+            force_direct=False,
+            minimum_z_height=None,
+            speed=None,
+            operation_volume=-50.0,
+        ),
+    ).then_return(Point(x=4, y=5, z=6))
+
     result = await subject.execute(data)
 
     if isinstance(location, CurrentWell):
@@ -374,7 +444,7 @@ async def test_overpressure_error(
                     volume_added=update_types.CLEAR,
                 ),
                 pipette_aspirated_fluid=update_types.PipetteUnknownFluidUpdate(
-                    pipette_id="pipette-id"
+                    pipette_id="pipette-id-abc"
                 ),
             ),
         )
@@ -388,7 +458,7 @@ async def test_overpressure_error(
             ),
             state_update=update_types.StateUpdate(
                 pipette_aspirated_fluid=update_types.PipetteUnknownFluidUpdate(
-                    pipette_id="pipette-id"
+                    pipette_id="pipette-id-abc"
                 )
             ),
         )

api/tests/opentrons/protocol_engine/state/test_geometry_view.py

@@ -4299,3 +4299,34 @@ def test_virtual_find_height_and_volume(
     #  SimulatedProbeResult
     if isinstance(target_height_volume, SimulatedProbeResult):
         assert height_estimate == volume_estimate == target_height_volume
+
+
+def test_get_liquid_handling_z_change(
+    decoy: Decoy,
+    subject: GeometryView,
+    mock_labware_view: LabwareView,
+    mock_well_view: WellView,
+) -> None:
+    decoy.when(mock_labware_view.get_well_definition("labware-id", "A1")).then_return(
+        RectangularWellDefinition3.model_construct(totalLiquidVolume=1100000)  # type: ignore[call-arg]
+    )
+    decoy.when(mock_labware_view.get_well_geometry("labware-id", "A1")).then_return(
+        _TEST_INNER_WELL_GEOMETRY
+    )
+    probe_time = datetime.now()
+    decoy.when(mock_well_view.get_last_liquid_update("labware-id", "A1")).then_return(
+        probe_time
+    )
+    decoy.when(mock_well_view.get_well_liquid_info("labware-id", "A1")).then_return(
+        WellLiquidInfo(
+            loaded_volume=None,
+            probed_height=ProbedHeightInfo(height=40.0, last_probed=probe_time),
+            probed_volume=None,
+        )
+    )
+    # make sure that liquid handling z change math stays the same
+    change = subject.get_liquid_handling_z_change(
+        labware_id="labware-id", well_name="A1", operation_volume=199.0
+    )
+    expected_change = 3.2968
+    assert isclose(change, expected_change)