wip feat: handle initial mock pv value as argument

ophyd_devices/tests/utils.py

@@ -10,6 +10,7 @@ import ophyd
 from bec_lib.devicemanager import ScanInfo
 from bec_lib.logger import bec_logger
 from bec_lib.utils.import_utils import lazy_import_from
+from copier import partial
 from ophyd import Device
 from ophyd.utils.epics_pvs import AlarmSeverity, AlarmStatus
 
@@ -25,7 +26,9 @@ T = TypeVar("T", bound=Device)
 
 
 @contextmanager
-def patched_device(device_type: type[T], *args, **kwargs) -> Generator[T, None, None]:
+def patched_device(
+    device_type: type[T], *args, _mock_pv_initial_value=0, **kwargs
+) -> Generator[T, None, None]:
     """Context manager to yield a patched ophyd device with certain initialisation args.
     *args and **kwargs are passed directly through to the device constructor.
 
@@ -37,7 +40,7 @@ def patched_device(device_type: type[T], *args, **kwargs) -> Generator[T, None,
 
     """
     with mock.patch.object(ophyd, "cl") as mock_cl:
-        mock_cl.get_pv = MockPV
+        mock_cl.get_pv = partial(MockPV, _mock_pv_initial_value=_mock_pv_initial_value)
         mock_cl.thread_class = threading.Thread
         device = device_type(*args, **kwargs)
         patch_dual_pvs(device)
@@ -137,8 +140,6 @@ class MockPV:
 
     """
 
-    DEFAULT_VALUE = 0
-
     _fmtsca = "<PV '%(pvname)s', count=%(count)i, type=%(typefull)s, access=%(access)s>"
     _fmtarr = "<PV '%(pvname)s', count=%(count)i/%(nelm)i, type=%(typefull)s, access=%(access)s>"
     _fields = (
@@ -173,6 +174,8 @@ class MockPV:
     def __init__(
         self,
         pvname,
+        *,
+        _mock_pv_initial_value=0,
         callback=None,
         form="time",
         verbose=False,
@@ -202,7 +205,7 @@ class MockPV:
         self._args["access"] = "unknown"
         self._args["status"] = 0
         self.connection_callbacks = []
-        self._mock_data = self.DEFAULT_VALUE
+        self._mock_data = _mock_pv_initial_value
 
         if connection_callback is not None:
             self.connection_callbacks = [connection_callback]

tests/test_utils.py

@@ -20,7 +20,8 @@ from ophyd_devices import (
     StatusBase,
     SubscriptionStatus,
 )
-from ophyd_devices.tests.utils import MockPV
+from ophyd_devices.devices.psi_motor import EpicsMotor
+from ophyd_devices.tests.utils import MockPV, patched_device
 from ophyd_devices.utils.bec_signals import (
     AsyncMultiSignal,
     AsyncSignal,
@@ -77,6 +78,17 @@ def task_handler(device):
     yield TaskHandler(parent=device)
 
 
+@pytest.fixture(scope="function")
+def mock_device_with_initial_value():
+    with patched_device(EpicsMotor, _mock_pv_initial_value=2, name="motor") as mtr:
+        yield mtr
+
+
+def test_mock_device_initial_value(mock_device_with_initial_value: EpicsMotor):
+    mtr = mock_device_with_initial_value
+    assert mtr.velocity.get() == 2
+
+
 def test_utils_file_handler_has_full_path(file_handler):
     """Ensure that file_handler has a get_full_path method"""
     assert hasattr(file_handler, "get_full_path")