unit test and error raise for blowout location

api/src/opentrons/protocol_api/core/engine/instrument.py

@@ -38,6 +38,7 @@
 from opentrons_shared_data.errors.exceptions import (
     UnsupportedHardwareCommand,
 )
+from opentrons_shared_data.liquid_classes.liquid_class_definition import BlowoutLocation
 from opentrons.protocol_api._nozzle_layout import NozzleLayout
 from . import overlap_versions, pipette_movement_conflict
 from . import transfer_components_executor as tx_comps_executor
@@ -1246,6 +1247,16 @@ def consolidate_liquid(  # noqa: C901
         transfer_props = liquid_class.get_for(
             pipette=self.get_pipette_name(), tip_rack=tiprack_uri_for_transfer_props
         )
+        blow_out_properties = transfer_props.dispense.retract.blowout
+        if (
+            blow_out_properties.enabled
+            and blow_out_properties.location == BlowoutLocation.SOURCE
+        ):
+            raise RuntimeError(
+                'Blowout location "source" incompatible with consolidate liquid.'
+                ' Please choose "destination" or "trash".'
+            )
+
         # TODO: use the ID returned by load_liquid_class in command annotations
         self.load_liquid_class(
             name=liquid_class.name,

api/tests/opentrons/protocol_api/test_instrument_context.py

@@ -2350,16 +2350,6 @@ def test_consolidate_liquid_raises_if_tip_has_liquid(
     decoy.when(mock_validation.ensure_new_tip_policy("never")).then_return(
         TransferTipPolicyV2.ONCE
     )
-    decoy.when(mock_instrument_core.get_nozzle_map()).then_return(MOCK_MAP)
-    decoy.when(mock_instrument_core.get_active_channels()).then_return(2)
-    decoy.when(
-        labware.next_available_tip(
-            starting_tip=None,
-            tip_racks=tip_racks,
-            channels=2,
-            nozzle_map=MOCK_MAP,
-        )
-    ).then_return((decoy.mock(cls=Labware), decoy.mock(cls=Well)))
     decoy.when(mock_instrument_core.get_current_volume()).then_return(1000)
     with pytest.raises(RuntimeError, match="liquid already in the tip"):
         subject.consolidate_liquid(
@@ -2371,6 +2361,37 @@ def test_consolidate_liquid_raises_if_tip_has_liquid(
         )
 
 
+@pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"])
+def test_consolidate_liquid_raises_if_tip_policy_per_source(
+    decoy: Decoy,
+    mock_protocol_core: ProtocolCore,
+    mock_instrument_core: InstrumentCore,
+    subject: InstrumentContext,
+    robot_type: RobotType,
+    minimal_liquid_class_def2: LiquidClassSchemaV1,
+) -> None:
+    """It should raise errors if the tip policy is "per source"."""
+    test_liq_class = LiquidClass.create(minimal_liquid_class_def2)
+    mock_well = decoy.mock(cls=Well)
+
+    decoy.when(
+        mock_validation.ensure_valid_flat_wells_list_for_transfer_v2([mock_well])
+    ).then_return([mock_well])
+    decoy.when(mock_validation.ensure_new_tip_policy("per source")).then_return(
+        TransferTipPolicyV2.PER_SOURCE
+    )
+    with pytest.raises(
+        RuntimeError, match='"per source" incompatible with consolidate.'
+    ):
+        subject.consolidate_liquid(
+            liquid_class=test_liq_class,
+            volume=10,
+            source=[mock_well],
+            dest=mock_well,
+            new_tip="per source",
+        )
+
+
 @pytest.mark.parametrize("robot_type", ["OT-2 Standard", "OT-3 Standard"])
 def test_consolidate_liquid_delegates_to_engine_core(
     decoy: Decoy,