Changes to unit tests to work on python 3.12

pyproject.toml

@@ -148,6 +148,10 @@ filterwarnings = [
     # Our dependency is via opentelemetry-exporter-otlp==1.27 which depends opentelemetry-proto==1.27
     # which depends protobuf>=3.19, <5.0
     # https://github.com/open-telemetry/opentelemetry-python/pull/4206/commits addresses this but is not yet in a release
+    "ignore:.*custom tp_new.*:DeprecationWarning",
+    # Ignore warning about deprecated throw() call https://github.com/bluesky/bluesky/issues/1817
+    "ignore:.*signature of throw\\(\\) is deprecated.*:DeprecationWarning",
+
 ]
 # Doctest python code in docs, python code in src docstrings, test functions in tests
 testpaths = "docs src tests/unit_tests"

src/mx_bluesky/hyperion/external_interaction/nexus/nexus_utils.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import time
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import numpy as np
 from dodal.devices.detector import DetectorParams
@@ -93,7 +93,7 @@ def create_goniometer_axes(
 
 def get_start_and_predicted_end_time(time_expected: float) -> tuple[str, str]:
     time_format = r"%Y-%m-%dT%H:%M:%SZ"
-    start = datetime.utcfromtimestamp(time.time())
+    start = datetime.fromtimestamp(time.time(), tz=timezone.utc)
     end_est = start + timedelta(seconds=time_expected)
     return start.strftime(time_format), end_est.strftime(time_format)
 

tests/conftest.py

@@ -14,6 +14,7 @@
 import numpy
 import numpy as np
 import pytest
+from bluesky.protocols import Movable, Status
 from bluesky.run_engine import RunEngine
 from bluesky.simulators import RunEngineSimulator
 from bluesky.utils import Msg

tests/unit_tests/beamlines/i04/test_thawing.py

@@ -204,19 +204,21 @@ def test_thaw_and_stream_adds_murko_callback_and_produces_expected_messages(
     patch_murko_callback: MagicMock,
     smargon: Smargon,
     thawer: Thawer,
+    robot: BartRobot,
     oav_forwarder: OAVToRedisForwarder,
     oav: OAV,
     RE: RunEngine,
 ):
     patch_murko_instance = patch_murko_callback.return_value
+    from bluesky.protocols import Locatable
     RE(
         thaw_and_stream_to_redis(
             10,
             360,
             thawer=thawer,
             smargon=smargon,
             oav=oav,
-            robot=MagicMock(),
+            robot=robot,
             oav_to_redis_forwarder=oav_forwarder,
         )
     )
@@ -239,6 +241,7 @@ def test_thaw_and_stream_will_produce_events_that_call_murko(
     patch_murko_call: MagicMock,
     smargon: Smargon,
     thawer: Thawer,
+    robot: BartRobot,
     oav_forwarder: OAVToRedisForwarder,
     oav: OAV,
     RE: RunEngine,
@@ -250,7 +253,7 @@ def test_thaw_and_stream_will_produce_events_that_call_murko(
             thawer=thawer,
             smargon=smargon,
             oav=oav,
-            robot=MagicMock(),
+            robot=robot,
             oav_to_redis_forwarder=oav_forwarder,
         )
     )

tests/unit_tests/hyperion/device_setup_plans/test_setup_panda.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from ophyd_async.fastcs.panda import HDFPanda
 from unittest.mock import MagicMock, patch
 
 import numpy as np
@@ -24,11 +25,13 @@ def get_smargon_speed(x_step_size_mm: float, time_between_x_steps_ms: float) ->
 
 
 def run_simulating_setup_panda_functions(
-    plan: str, sim_run_engine: RunEngineSimulator, mock_load_device=MagicMock
+    plan: str,
+    panda: HDFPanda,
+    sim_run_engine: RunEngineSimulator,
+    mock_load_device=MagicMock
 ):
     num_of_sets = 0
     num_of_waits = 0
-    mock_panda = MagicMock()
 
     def count_commands(msg):
         nonlocal num_of_sets
@@ -45,7 +48,7 @@ def count_commands(msg):
         smargon_speed = get_smargon_speed(0.1, 1)
         sim.simulate_plan(
             setup_panda_for_flyscan(
-                mock_panda,
+                panda,
                 PandAGridScanParams(transmission_fraction=0.01),
                 1,
                 0.1,
@@ -54,15 +57,15 @@ def count_commands(msg):
             )
         )
     elif plan == "disarm":
-        sim.simulate_plan(disarm_panda_for_gridscan(mock_panda))
+        sim.simulate_plan(disarm_panda_for_gridscan(panda))
 
     return num_of_sets, num_of_waits
 
 
 @patch("mx_bluesky.hyperion.device_setup_plans.setup_panda.load_device")
-def test_setup_panda_performs_correct_plans(mock_load_device, sim_run_engine):
+def test_setup_panda_performs_correct_plans(mock_load_device, sim_run_engine, panda):
     num_of_sets, num_of_waits = run_simulating_setup_panda_functions(
-        "setup", sim_run_engine, mock_load_device
+        "setup", panda, sim_run_engine, mock_load_device
     )
     mock_load_device.assert_called_once()
     assert num_of_sets == 8
@@ -181,7 +184,8 @@ def test_setup_panda_correctly_configures_table(
         )
 
 
-def test_wait_between_setting_table_and_arming_panda(RE: RunEngine):
+def test_wait_between_setting_table_and_arming_panda(RE: RunEngine,
+                                                     panda):
     bps_wait_done = False
 
     def handle_wait(*args, **kwargs):
@@ -207,7 +211,7 @@ def assert_set_table_has_been_waited_on(*args, **kwargs):
     ):
         RE(
             setup_panda_for_flyscan(
-                MagicMock(),
+                panda,
                 PandAGridScanParams(transmission_fraction=0.01),
                 1,
                 0.1,
@@ -216,12 +220,14 @@ def assert_set_table_has_been_waited_on(*args, **kwargs):
             )
         )
 
+    assert bps_wait_done
+
 
 # It also would be useful to have some system tests which check that (at least)
 # all the blocks which were enabled on setup are also disabled on tidyup
-def test_disarm_panda_disables_correct_blocks(sim_run_engine):
+def test_disarm_panda_disables_correct_blocks(sim_run_engine, panda):
     num_of_sets, num_of_waits = run_simulating_setup_panda_functions(
-        "disarm", sim_run_engine
+        "disarm", panda, sim_run_engine
     )
     assert num_of_sets == 5
     assert num_of_waits == 1

tests/unit_tests/hyperion/device_setup_plans/test_utils.py

@@ -4,6 +4,7 @@
 from bluesky import plan_stubs as bps
 from bluesky.utils import FailedStatus
 from dodal.beamlines import i03
+from ophyd_async.core import get_mock_put
 from ophyd.status import Status
 
 from mx_bluesky.hyperion.device_setup_plans.utils import (
@@ -25,14 +26,13 @@ class MyTestException(Exception):
 
 
 def test_given_plan_raises_when_exception_raised_then_eiger_disarmed_and_correct_exception_returned(
-    mock_eiger, RE
+    mock_eiger, detector_motion, RE
 ):
     def my_plan():
         yield from bps.null()
         raise MyTestException()
 
     eiger = mock_eiger
-    detector_motion = MagicMock()
 
     with pytest.raises(MyTestException):
         RE(
@@ -74,9 +74,8 @@ def test_given_shutter_open_fails_then_eiger_disarmed_and_correct_exception_retu
 
 
 def test_given_detector_move_fails_then_eiger_disarmed_and_correct_exception_returned(
-    mock_eiger, null_plan, RE
+    mock_eiger, detector_motion, null_plan, RE
 ):
-    detector_motion = MagicMock()
     status = Status()
     status.set_exception(MyTestException())
     detector_motion.shutter.set = MagicMock(return_value=status)
@@ -90,5 +89,5 @@ def test_given_detector_move_fails_then_eiger_disarmed_and_correct_exception_ret
     assert e.value.args[0] is status
 
     mock_eiger.async_stage.assert_called_once()
-    detector_motion.z.set.assert_called_once()
+    get_mock_put(detector_motion.z.user_setpoint).assert_called_once()
     mock_eiger.disarm_detector.assert_called_once()

tests/unit_tests/hyperion/device_setup_plans/test_zebra_setup.py

@@ -2,6 +2,7 @@
 
 import pytest
 from bluesky import plan_stubs as bps
+from bluesky.protocols import Movable
 from dodal.beamlines import i03
 from dodal.devices.zebra import (
     AUTO_SHUTTER_GATE,
@@ -94,7 +95,7 @@ class MyException(Exception):
 
 
 def test_when_first_try_fails_then_bluesky_retry_tries_again(RE, done_status):
-    mock_device = MagicMock()
+    mock_device = MagicMock(spec=Movable)
 
     @bluesky_retry
     def my_plan(value):
@@ -108,7 +109,7 @@ def my_plan(value):
 
 
 def test_when_all_tries_fail_then_bluesky_retry_throws_error(RE, done_status):
-    mock_device = MagicMock()
+    mock_device = MagicMock(spec=Movable)
 
     @bluesky_retry
     def my_plan(value):

tests/unit_tests/hyperion/experiment_plans/conftest.py

@@ -4,15 +4,20 @@
 
 import pytest
 from bluesky.utils import Msg
-from dodal.devices.aperturescatterguard import ApertureValue
+
+from dodal.devices.aperturescatterguard import ApertureValue, ApertureScatterguard
+from dodal.devices.backlight import Backlight
+from dodal.devices.detector.detector_motion import DetectorMotion
 from dodal.devices.fast_grid_scan import ZebraFastGridScan
 from dodal.devices.oav.oav_detector import OAVConfigParams
+from dodal.devices.smargon import Smargon
 from dodal.devices.synchrotron import SynchrotronMode
 from dodal.devices.zocalo import ZocaloResults, ZocaloTrigger
 from event_model import Event
 from ophyd.sim import NullStatus
 from ophyd_async.core import AsyncStatus, DeviceCollector, set_mock_value
 
+from mx_bluesky.hyperion.experiment_plans.grid_detect_then_xray_centre_plan import GridDetectThenXRayCentreComposite
 from mx_bluesky.hyperion.experiment_plans.robot_load_and_change_energy import (
     RobotLoadAndEnergyChangeComposite,
 )
@@ -68,6 +73,38 @@ def make_event_doc(data, descriptor="abc123") -> Event:
 }
 
 
+@pytest.fixture
+def grid_detect_devices(
+    aperture_scatterguard: ApertureScatterguard,
+    backlight: Backlight,
+    detector_motion: DetectorMotion,
+    smargon: Smargon,
+) -> GridDetectThenXRayCentreComposite:
+    return GridDetectThenXRayCentreComposite(
+        aperture_scatterguard=aperture_scatterguard,
+        attenuator=MagicMock(),
+        backlight=backlight,
+        detector_motion=detector_motion,
+        eiger=MagicMock(),
+        zebra_fast_grid_scan=MagicMock(),
+        flux=MagicMock(),
+        oav=MagicMock(),
+        pin_tip_detection=MagicMock(),
+        smargon=smargon,
+        synchrotron=MagicMock(),
+        s4_slit_gaps=MagicMock(),
+        undulator=MagicMock(),
+        xbpm_feedback=MagicMock(),
+        zebra=MagicMock(),
+        zocalo=MagicMock(),
+        panda=MagicMock(),
+        panda_fast_grid_scan=MagicMock(),
+        dcm=MagicMock(),
+        robot=MagicMock(),
+        sample_shutter=MagicMock(),
+    )
+
+
 @pytest.fixture
 def sim_run_engine_for_rotation(sim_run_engine):
     sim_run_engine.add_handler(
@@ -187,7 +224,7 @@ def fake_read(obj, initial_positions, _):
 
 
 @pytest.fixture
-def simple_beamline(detector_motion, oav, smargon, synchrotron, test_config_files, dcm):
+def simple_beamline(detector_motion, eiger, oav, smargon, synchrotron, test_config_files, dcm):
     magic_mock = MagicMock(autospec=True)
 
     with DeviceCollector(mock=True):
@@ -199,6 +236,7 @@ def simple_beamline(detector_motion, oav, smargon, synchrotron, test_config_file
     magic_mock.detector_motion = detector_motion
     magic_mock.dcm = dcm
     magic_mock.synchrotron = synchrotron
+    magic_mock.eiger = eiger
     oav.zoom_controller.frst.set("7.5x")
     oav.parameters = OAVConfigParams(
         test_config_files["zoom_params_file"], test_config_files["display_config"]

tests/unit_tests/hyperion/experiment_plans/test_grid_detect_then_xray_centre_plan.py

@@ -60,38 +60,6 @@ def _fake_grid_detection(
     yield from bps.save()
 
 
-@pytest.fixture
-def grid_detect_devices(
-    aperture_scatterguard: i03.ApertureScatterguard,
-    backlight: i03.Backlight,
-    detector_motion: i03.DetectorMotion,
-    smargon: Smargon,
-):
-    return GridDetectThenXRayCentreComposite(
-        aperture_scatterguard=aperture_scatterguard,
-        attenuator=MagicMock(),
-        backlight=backlight,
-        detector_motion=detector_motion,
-        eiger=MagicMock(),
-        zebra_fast_grid_scan=MagicMock(),
-        flux=MagicMock(),
-        oav=MagicMock(),
-        pin_tip_detection=MagicMock(),
-        smargon=smargon,
-        synchrotron=MagicMock(),
-        s4_slit_gaps=MagicMock(),
-        undulator=MagicMock(),
-        xbpm_feedback=MagicMock(),
-        zebra=MagicMock(),
-        zocalo=MagicMock(),
-        panda=MagicMock(),
-        panda_fast_grid_scan=MagicMock(),
-        dcm=MagicMock(),
-        robot=MagicMock(),
-        sample_shutter=MagicMock(),
-    )
-
-
 def test_full_grid_scan(
     test_fgs_params: ThreeDGridScan, test_config_files: dict[str, str]
 ):