fix: bump ophyd version to 1.10, remove patch, fix corresponding test

ophyd_devices/__init__.py

@@ -1,7 +1,3 @@
-from .ophyd_patch import monkey_patch_ophyd
-
-monkey_patch_ophyd()
-
 from .devices.sls_devices import SLSInfo, SLSOperatorMessages
 from .sim.sim_camera import SimCamera
 from .sim.sim_monitor import SimMonitor, SimMonitorAsync

ophyd_devices/ophyd_patch.py

@@ -1,91 +0,0 @@
-import importlib
-import importlib.util
-import pathlib
-import sys
-from importlib.abc import Loader, MetaPathFinder
-from types import ModuleType
-
-_patched_status_base = """
-import threading
-from unittest.mock import Mock, patch
-
-_StatusBase = StatusBase
-
-class StatusBase(_StatusBase):
-    _bec_patched = True
-
-    def __init__(self, *args, **kwargs):
-        timeout = kwargs.get("timeout", None)
-        if not timeout:
-            with patch("threading.Thread", Mock(spec=threading.Thread)):
-                super().__init__(*args, **kwargs)
-        else:
-            super().__init__(*args, **kwargs)
-
-    def set_finished(self, *args, **kwargs):
-        super().set_finished(*args, **kwargs)
-        if isinstance(self._callback_thread, Mock):
-            if self.settle_time > 0:
-
-                def settle_done():
-                    self._settled_event.set()
-                    self._run_callbacks()
-
-                threading.Timer(self.settle_time, settle_done).start()
-            else:
-                self._run_callbacks()
-
-    def set_exception(self, *args, **kwargs):
-        super().set_exception(*args, **kwargs)
-        if isinstance(self._callback_thread, Mock):
-            self._run_callbacks()
-
-"""
-
-
-class _CustomLoader(Loader):
-    def __init__(self, patched_code):
-        self.patched_code = patched_code
-
-    def load_module(self, fullname):
-        """Load and execute ophyd.status"""
-        status_module = ModuleType("ophyd.status")
-        status_module.__loader__ = self
-        status_module.__file__ = None
-        status_module.__name__ = fullname
-
-        exec(self.patched_code, status_module.__dict__)
-        sys.modules[fullname] = status_module
-
-        return status_module, True
-
-
-class _CustomImporter(MetaPathFinder):
-    def __init__(self):
-        origin = pathlib.Path(importlib.util.find_spec("ophyd").origin)
-        module_file = str(origin.parent / "status.py")
-
-        with open(module_file, "r") as source:
-            src = source.read()
-            before, _, after = src.partition("class StatusBase")
-            orig_status_base, _, final = after.partition("\nclass ")
-
-        self.patched_source = (
-            f"{before}class StatusBase{orig_status_base}{_patched_status_base}class {final}"
-        )
-        self.patched_code = compile(self.patched_source, module_file, "exec")
-        self.loader = _CustomLoader(self.patched_code)
-
-    def find_spec(self, fullname, path, target=None):
-        # The new import classes are difficult to grasp;
-        # why the finder needs a .loader member, it could be returned
-        # from here. And also .name, which name has to correspond to
-        # the searched module ???
-        if fullname == "ophyd.status":
-            self.name = fullname
-            return self
-        return None
-
-
-def monkey_patch_ophyd():
-    sys.meta_path.insert(0, _CustomImporter())

pyproject.toml

@@ -14,7 +14,7 @@ classifiers = [
 ]
 
 dependencies = [
-    "ophyd ~= 1.9",
+    "ophyd ~= 1.10",
     "typeguard ~= 4.1, >=4.1.5",
     "prettytable ~= 3.9",
     "bec_lib >=2.12.2, <=4.0",

tests/test_ophyd_status_obj.py

@@ -6,11 +6,10 @@ import pytest
 from ophyd.status import DeviceStatus, StatusBase, StatusTimeoutError
 
 
-def test_ophyd_status_patch():
+def test_ophyd_status():
-    assert StatusBase._bec_patched
+    device = Mock()
-
+    device.name.return_value = "test"
-    st = DeviceStatus(device="test")
+    st = DeviceStatus(device)
-    assert st._bec_patched
     assert isinstance(st, StatusBase)
 
     cb = Mock()
@@ -25,7 +24,7 @@ def test_ophyd_status_patch():
     cb.reset_mock()
 
     st = StatusBase()
-    assert isinstance(st._callback_thread, Mock)
+    assert st._callback_thread is None
     st.add_callback(cb)
     st.set_finished()
     cb.assert_called_once()
@@ -34,7 +33,7 @@ def test_ophyd_status_patch():
 
     st = StatusBase(settle_time=1)
     st.add_callback(cb)
-    assert isinstance(st._callback_thread, Mock)
+    assert st._callback_thread is None
     st.set_finished()
     assert cb.call_count == 0
     time.sleep(0.5)