api: fix testst

api/src/opentrons/protocol_engine/execution/command_executor.py

@@ -182,7 +182,7 @@ async def execute(self, command_id: str) -> None:
                     "completedAt": self._model_utils.get_timestamp(),
                     "notes": note_tracker.get_notes(),
                 }
-                succeeded_command = running_command.copy(update=update)
+                succeeded_command = running_command.model_copy(update=update)
                 self._action_dispatcher.dispatch(
                     SucceedCommandAction(
                         command=succeeded_command, private_result=result.private

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

@@ -314,7 +314,7 @@ def _handle_queue_command_action(self, action: QueueCommandAction) -> None:
     def _handle_run_command_action(self, action: RunCommandAction) -> None:
         prev_entry = self._state.command_history.get(action.command_id)
 
-        running_command = prev_entry.command.copy(
+        running_command = prev_entry.command.model_copy(
             update={
                 "status": CommandStatus.RUNNING,
                 "startedAt": action.started_at,
@@ -503,7 +503,7 @@ def _update_to_failed(
         notes: Optional[List[CommandNote]],
     ) -> None:
         prev_entry = self._state.command_history.get(command_id)
-        failed_command = prev_entry.command.copy(
+        failed_command = prev_entry.command.model_copy(
             update={
                 "completedAt": failed_at,
                 "status": CommandStatus.FAILED,

api/src/opentrons/protocol_engine/state/geometry.py

@@ -425,9 +425,9 @@ def get_well_position(
         if well_location is not None:
             offset = well_location.offset
             if well_location.origin == WellOrigin.TOP:
-                offset = offset.copy(update={"z": offset.z + well_depth})
+                offset = offset.model_copy(update={"z": offset.z + well_depth})
             elif well_location.origin == WellOrigin.CENTER:
-                offset = offset.copy(update={"z": offset.z + well_depth / 2.0})
+                offset = offset.model_copy(update={"z": offset.z + well_depth / 2.0})
 
         return Point(
             x=labware_pos.x + offset.x + well_def.x,

api/src/opentrons/protocol_runner/legacy_command_mapper.py

@@ -174,7 +174,7 @@ def map_command(  # noqa: C901
             completed_command: pe_commands.Command
             if command_error is None:
                 if isinstance(running_command, pe_commands.PickUpTip):
-                    completed_command = running_command.copy(
+                    completed_command = running_command.model_copy(
                         update={
                             "result": pe_commands.PickUpTipResult(
                                 tipVolume=command["payload"]["location"].max_volume,  # type: ignore[typeddict-item]
@@ -187,7 +187,7 @@ def map_command(  # noqa: C901
                         }
                     )
                 elif isinstance(running_command, pe_commands.DropTip):
-                    completed_command = running_command.copy(
+                    completed_command = running_command.model_copy(
                         update={
                             "result": pe_commands.DropTipResult(
                                 position=pe_types.DeckPoint(x=0, y=0, z=0)
@@ -198,7 +198,7 @@ def map_command(  # noqa: C901
                         }
                     )
                 elif isinstance(running_command, pe_commands.Aspirate):
-                    completed_command = running_command.copy(
+                    completed_command = running_command.model_copy(
                         update={
                             # Don't .construct() result, because we want to validate
                             # volume.
@@ -212,7 +212,7 @@ def map_command(  # noqa: C901
                         }
                     )
                 elif isinstance(running_command, pe_commands.Dispense):
-                    completed_command = running_command.copy(
+                    completed_command = running_command.model_copy(
                         update={
                             # Don't .construct() result, because we want to validate
                             # volume.
@@ -226,7 +226,7 @@ def map_command(  # noqa: C901
                         }
                     )
                 elif isinstance(running_command, pe_commands.BlowOut):
-                    completed_command = running_command.copy(
+                    completed_command = running_command.model_copy(
                         update={
                             "result": pe_commands.BlowOutResult(
                                 position=pe_types.DeckPoint(x=0, y=0, z=0)
@@ -237,7 +237,7 @@ def map_command(  # noqa: C901
                         }
                     )
                 elif isinstance(running_command, pe_commands.Comment):
-                    completed_command = running_command.copy(
+                    completed_command = running_command.model_copy(
                         update={
                             "result": pe_commands.CommentResult.construct(),
                             "status": pe_commands.CommandStatus.SUCCEEDED,
@@ -246,7 +246,7 @@ def map_command(  # noqa: C901
                         }
                     )
                 elif isinstance(running_command, pe_commands.Custom):
-                    completed_command = running_command.copy(
+                    completed_command = running_command.model_copy(
                         update={
                             "result": pe_commands.CustomResult(),
                             "status": pe_commands.CommandStatus.SUCCEEDED,
@@ -258,7 +258,7 @@ def map_command(  # noqa: C901
                     # TODO(mm, 2024-06-13): This looks potentially wrong.
                     # We're creating a `SUCCEEDED` command that does not have a `result`,
                     # which is not normally possible.
-                    completed_command = running_command.copy(
+                    completed_command = running_command.model_copy(
                         update={
                             "status": pe_commands.CommandStatus.SUCCEEDED,
                             "completedAt": now,

api/tests/opentrons/protocol_engine/conftest.py

@@ -101,7 +101,7 @@ def ot3_fixed_trash_def() -> LabwareDefinition:
 @pytest.fixture(scope="session")
 def ot3_absorbance_reader_lid() -> LabwareDefinition:
     """Get the definition of the OT-3 plate reader lid."""
-    return LabwareDefinition.parse_obj(
+    return LabwareDefinition.model_validate(
         load_definition("opentrons_flex_lid_absorbance_plate_reader_module", 1)
     )
 

api/tests/opentrons/protocol_engine/execution/test_command_executor.py

@@ -6,14 +6,15 @@
 
 import pytest
 from decoy import Decoy, matchers
-from pydantic import BaseModel
+from pydantic import BaseModel, PrivateAttr
 
 from opentrons.hardware_control import HardwareControlAPI, OT2HardwareControlAPI
 
 from opentrons.protocol_engine import errors
 from opentrons.protocol_engine.error_recovery_policy import (
     ErrorRecoveryPolicy,
     ErrorRecoveryType,
+    never_recover,
 )
 from opentrons.protocol_engine.errors.error_occurrence import ErrorOccurrence
 from opentrons.protocol_engine.errors.exceptions import (
@@ -251,21 +252,28 @@ async def test_execute(
     TestCommandImplCls = decoy.mock(func=_TestCommandImpl)
     command_impl = decoy.mock(cls=_TestCommandImpl)
 
+    # Note: private attrs (which are attrs that start with _) are instantiated via deep
+    # copy from a provided default in the model, so if
+    # _TestCommand()._ImplementationCls != _TestCommand._ImplementationCls.default if
+    # we provide a default. Therefore, provide a default factory, so we can always have
+    # the same object.
+
     class _TestCommand(
         BaseCommand[_TestCommandParams, _TestCommandResult, ErrorOccurrence]
     ):
         commandType: str = "testCommand"
         params: _TestCommandParams
         result: Optional[_TestCommandResult]
 
-        _ImplementationCls: Type[_TestCommandImpl] = TestCommandImplCls
+        _ImplementationCls: Type[_TestCommandImpl] = PrivateAttr(
+            default_factory=lambda: TestCommandImplCls
+        )
 
     command_params = _TestCommandParams()
     command_result = SuccessData(public=_TestCommandResult(), private=None)
-
     queued_command = cast(
         Command,
-        _TestCommand(  # type: ignore[call-arg]
+        _TestCommand.model_construct(  # type: ignore[call-arg]
             id="command-id",
             key="command-key",
             createdAt=datetime(year=2021, month=1, day=1),
@@ -276,7 +284,7 @@ class _TestCommand(
 
     running_command = cast(
         Command,
-        _TestCommand(  # type: ignore[call-arg]
+        _TestCommand.model_construct(  # type: ignore[call-arg]
             id="command-id",
             key="command-key",
             createdAt=datetime(year=2021, month=1, day=1),
@@ -297,7 +305,7 @@ class _TestCommand(
 
     expected_completed_command = cast(
         Command,
-        _TestCommand(
+        _TestCommand.model_construct(
             id="command-id",
             key="command-key",
             createdAt=datetime(year=2021, month=1, day=1),
@@ -325,7 +333,6 @@ class _TestCommand(
             state_store.commands.get(command_id="command-id")
         ).then_return(running_command)
     )
-
     decoy.when(
         queued_command._ImplementationCls(
             state_view=state_store,
@@ -355,6 +362,10 @@ class _TestCommand(
         datetime(year=2023, month=3, day=3),
     )
 
+    decoy.when(state_store.commands.get_error_recovery_policy()).then_return(
+        never_recover
+    )
+
     await subject.execute("command-id")
 
     decoy.verify(
@@ -419,13 +430,15 @@ class _TestCommand(
         params: _TestCommandParams
         result: Optional[_TestCommandResult]
 
-        _ImplementationCls: Type[_TestCommandImpl] = TestCommandImplCls
+        _ImplementationCls: Type[_TestCommandImpl] = PrivateAttr(
+            default_factory=lambda: TestCommandImplCls
+        )
 
     command_params = _TestCommandParams()
 
     queued_command = cast(
         Command,
-        _TestCommand(  # type: ignore[call-arg]
+        _TestCommand.model_construct(  # type: ignore[call-arg]
             id="command-id",
             key="command-key",
             createdAt=datetime(year=2021, month=1, day=1),
@@ -436,7 +449,7 @@ class _TestCommand(
 
     running_command = cast(
         Command,
-        _TestCommand(  # type: ignore[call-arg]
+        _TestCommand.model_construct(  # type: ignore[call-arg]
             id="command-id",
             key="command-key",
             createdAt=datetime(year=2021, month=1, day=1),
@@ -552,7 +565,9 @@ class _TestCommand(
         params: _TestCommandParams
         result: Optional[_TestCommandResult]
 
-        _ImplementationCls: Type[_TestCommandImpl] = TestCommandImplCls
+        _ImplementationCls: Type[_TestCommandImpl] = PrivateAttr(
+            default_factory=lambda: TestCommandImplCls
+        )
 
     command_params = _TestCommandParams()
     command_id = "command-id"
@@ -566,7 +581,7 @@ class _TestCommand(
     )
     queued_command = cast(
         Command,
-        _TestCommand(  # type: ignore[call-arg]
+        _TestCommand.model_construct(  # type: ignore[call-arg]
             id=command_id,
             key="command-key",
             createdAt=created_at,
@@ -576,7 +591,7 @@ class _TestCommand(
     )
     running_command = cast(
         Command,
-        _TestCommand(  # type: ignore[call-arg]
+        _TestCommand.model_construct(  # type: ignore[call-arg]
             id=command_id,
             key="command-key",
             createdAt=created_at,

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

@@ -198,13 +198,13 @@ def test_set_fixit_running_command_id(command_history: CommandHistory) -> None:
     """It should set the ID of the currently running fixit command."""
     command_entry = create_queued_command()
     command_history.append_queued_command(command_entry)
-    running_command = command_entry.copy(
+    running_command = command_entry.model_copy(
         update={
             "status": CommandStatus.RUNNING,
         }
     )
     command_history.set_command_running(running_command)
-    finished_command = command_entry.copy(
+    finished_command = command_entry.model_copy(
         update={
             "status": CommandStatus.SUCCEEDED,
         }
@@ -214,7 +214,7 @@ def test_set_fixit_running_command_id(command_history: CommandHistory) -> None:
         command_id="fixit-id", intent=CommandIntent.FIXIT
     )
     command_history.append_queued_command(fixit_command_entry)
-    fixit_running_command = fixit_command_entry.copy(
+    fixit_running_command = fixit_command_entry.model_copy(
         update={
             "status": CommandStatus.RUNNING,
         }

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

@@ -210,7 +210,7 @@ def addressable_area_view(
 @pytest.fixture
 def nice_labware_definition() -> LabwareDefinition:
     """Load a nice labware def that won't blow up your terminal."""
-    return LabwareDefinition.parse_obj(
+    return LabwareDefinition.model_validate(
         json.loads(
             load_shared_data("labware/fixtures/2/fixture_12_trough_v2.json").decode(
                 "utf-8"
@@ -222,7 +222,7 @@ def nice_labware_definition() -> LabwareDefinition:
 @pytest.fixture
 def nice_adapter_definition() -> LabwareDefinition:
     """Load a friendly adapter definition."""
-    return LabwareDefinition.parse_obj(
+    return LabwareDefinition.model_validate(
         json.loads(
             load_shared_data(
                 "labware/definitions/2/opentrons_aluminum_flat_bottom_plate/1.json"

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

@@ -533,7 +533,7 @@ def test_blow_out_clears_volume(
         ),
         (
             FailCommandAction(
-                running_command=cmd.LiquidProbe(  # type: ignore[call-arg]
+                running_command=cmd.LiquidProbe.model_construct(  # type: ignore[call-arg]
                     id="command-id",
                     createdAt=datetime.now(),
                     startedAt=datetime.now(),