diff --git a/tests/client_mocks.py b/tests/client_mocks.py index 29a9f766..6764e03e 100644 --- a/tests/client_mocks.py +++ b/tests/client_mocks.py @@ -1,8 +1,9 @@ # pylint: disable = no-name-in-module,missing-class-docstring, missing-module-docstring - -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import pytest +from bec_lib.device import Positioner +from bec_lib.devicemanager import DeviceContainer class FakeDevice: @@ -12,7 +13,7 @@ class FakeDevice: self.name = name self.enabled = enabled self.signals = {self.name: {"value": 1.0}} - self.description = {self.name: {"source": self.name}} + self.description = {self.name: {"source": self.name, "dtype": "number", "shape": []}} def __contains__(self, item): return item == self.name @@ -38,41 +39,85 @@ class FakeDevice: return self.description -def get_mocked_device(device_name: str): - """ - Helper function to mock the devices - Args: - device_name(str): Name of the device to mock - """ - return FakeDevice(name=device_name, enabled=True) +class FakePositioner(FakeDevice): + def __init__(self, name, enabled=True, limits=None, read_value=1.0): + super().__init__(name, enabled) + self.limits = limits if limits is not None else [0, 0] + self.read_value = read_value + + def set_read_value(self, value): + self.read_value = value + + def read(self): + return {self.name: {"value": self.read_value}} + + def set_limits(self, limits): + self.limits = limits + + def move(self, value, relative=False): + """Simulates moving the device to a new position.""" + if relative: + self.read_value += value + else: + self.read_value = value + # Respect the limits + self.read_value = max(min(self.read_value, self.limits[1]), self.limits[0]) + + @property + def readback(self): + return MagicMock(get=MagicMock(return_value=self.read_value)) + + +class DMMock: + def __init__(self): + self.devices = DeviceContainer() + + def add_devives(self, devices: list): + for device in devices: + self.devices[device.name] = device + + +DEVICES = [ + FakePositioner("samx", limits=[-10, 10], read_value=2.0), + FakePositioner("samy", limits=[-5, 5], read_value=3.0), + FakePositioner("aptrx", limits=None, read_value=4.0), + FakePositioner("aptry", limits=None, read_value=5.0), + FakeDevice("gauss_bpm"), + FakeDevice("gauss_adc1"), + FakeDevice("gauss_adc2"), + FakeDevice("gauss_adc3"), + FakeDevice("bpm4i"), + FakeDevice("bpm3a"), + FakeDevice("bpm3i"), +] @pytest.fixture(scope="function") def mocked_client(): - # Create a dictionary of mocked devices - device_names = [ - "samx", - "samy", - "gauss_bpm", - "gauss_adc1", - "gauss_adc2", - "gauss_adc3", - "bpm4i", - "bpm3a", - "bpm3i", - ] - mocked_devices = {name: get_mocked_device(name) for name in device_names} - # Create a MagicMock object client = MagicMock() # Mock the device_manager.devices attribute - client.device_manager.devices = MagicMock() - client.device_manager.devices.__getitem__.side_effect = lambda x: mocked_devices.get(x) - client.device_manager.devices.__contains__.side_effect = lambda x: x in mocked_devices + client.device_manager = DMMock() + client.device_manager.add_devives(DEVICES) - # Set each device as an attribute of the mock - for name, device in mocked_devices.items(): - setattr(client.device_manager.devices, name, device) + def mock_mv(*args, relative=False): + # Extracting motor and value pairs + for i in range(0, len(args), 2): + motor = args[i] + value = args[i + 1] + motor.move(value, relative=relative) + return MagicMock(wait=MagicMock()) - return client + client.scans = MagicMock(mv=mock_mv) + + # Ensure isinstance check for Positioner passes + original_isinstance = isinstance + + def isinstance_mock(obj, class_info): + if class_info == Positioner and isinstance(obj, FakePositioner): + return True + return original_isinstance(obj, class_info) + + with patch("builtins.isinstance", new=isinstance_mock): + yield client diff --git a/tests/test_bec_monitor.py b/tests/test_bec_monitor.py index 7f0d0434..5feac4e7 100644 --- a/tests/test_bec_monitor.py +++ b/tests/test_bec_monitor.py @@ -7,6 +7,8 @@ import yaml from bec_widgets.widgets import BECMonitor +from .client_mocks import mocked_client + def load_test_config(config_name): """Helper function to load config from yaml file.""" @@ -16,69 +18,6 @@ def load_test_config(config_name): return config -class FakeDevice: - """Fake minimal positioner class for testing.""" - - def __init__(self, name, enabled=True): - self.name = name - self.enabled = enabled - self.signals = {self.name: {"value": 1.0}} - self.description = {self.name: {"source": self.name}} - - def __contains__(self, item): - return item == self.name - - @property - def _hints(self): - return [self.name] - - def set_value(self, fake_value: float = 1.0) -> None: - """ - Setup fake value for device readout - Args: - fake_value(float): Desired fake value - """ - self.signals[self.name]["value"] = fake_value - - def describe(self) -> dict: - """ - Get the description of the device - Returns: - dict: Description of the device - """ - return self.description - - -def get_mocked_device(device_name: str): - """ - Helper function to mock the devices - Args: - device_name(str): Name of the device to mock - """ - return FakeDevice(name=device_name, enabled=True) - - -@pytest.fixture(scope="function") -def mocked_client(): - # Create a dictionary of mocked devices - device_names = ["samx", "gauss_bpm", "gauss_adc1", "gauss_adc2", "gauss_adc3", "bpm4i"] - mocked_devices = {name: get_mocked_device(name) for name in device_names} - - # Create a MagicMock object - client = MagicMock() - - # Mock the device_manager.devices attribute - client.device_manager.devices = MagicMock() - client.device_manager.devices.__getitem__.side_effect = lambda x: mocked_devices.get(x) - client.device_manager.devices.__contains__.side_effect = lambda x: x in mocked_devices - - # Set each device as an attribute of the mock - for name, device in mocked_devices.items(): - setattr(client.device_manager.devices, name, device) - - return client - - @pytest.fixture(scope="function") def monitor(bec_dispatcher, qtbot, mocked_client): # client = MagicMock() @@ -266,9 +205,6 @@ def test_on_scan_segment(monitor, config_name, msg, metadata, expected_data): config = load_test_config(config_name) monitor.on_config_update(config) - # Get hints - monitor.dev.__getitem__.side_effect = mock_getitem - # Mock scan_storage.find_scan_by_ID mock_scan_data = MagicMock() mock_scan_data.data = { diff --git a/tests/test_config_dialog.py b/tests/test_config_dialog.py index 5405a64d..5b7306eb 100644 --- a/tests/test_config_dialog.py +++ b/tests/test_config_dialog.py @@ -8,6 +8,8 @@ from qtpy.QtWidgets import QTableWidgetItem, QTabWidget from bec_widgets.widgets.monitor.config_dialog import ConfigDialog +from .client_mocks import mocked_client + def load_test_config(config_name): """Helper function to load config from yaml file.""" @@ -17,69 +19,6 @@ def load_test_config(config_name): return config -class FakeDevice: - """Fake minimal positioner class for testing.""" - - def __init__(self, name, enabled=True): - self.name = name - self.enabled = enabled - self.signals = {self.name: {"value": 1.0}} - self.description = {self.name: {"source": self.name}} - - def __contains__(self, item): - return item == self.name - - @property - def _hints(self): - return [self.name] - - def set_value(self, fake_value: float = 1.0) -> None: - """ - Setup fake value for device readout - Args: - fake_value(float): Desired fake value - """ - self.signals[self.name]["value"] = fake_value - - def describe(self) -> dict: - """ - Get the description of the device - Returns: - dict: Description of the device - """ - return self.description - - -def get_mocked_device(device_name: str): - """ - Helper function to mock the devices - Args: - device_name(str): Name of the device to mock - """ - return FakeDevice(name=device_name, enabled=True) - - -@pytest.fixture(scope="function") -def mocked_client(): - # Create a dictionary of mocked devices - device_names = ["samx", "gauss_bpm", "gauss_adc1", "gauss_adc2", "gauss_adc3", "bpm4i"] - mocked_devices = {name: get_mocked_device(name) for name in device_names} - - # Create a MagicMock object - client = MagicMock() - - # Mock the device_manager.devices attribute - client.device_manager.devices = MagicMock() - client.device_manager.devices.__getitem__.side_effect = lambda x: mocked_devices.get(x) - client.device_manager.devices.__contains__.side_effect = lambda x: x in mocked_devices - - # Set each device as an attribute of the mock - for name, device in mocked_devices.items(): - setattr(client.device_manager.devices, name, device) - - return client - - @pytest.fixture(scope="function") def config_dialog(qtbot, mocked_client): client = mocked_client diff --git a/tests/test_motor_control.py b/tests/test_motor_control.py index 137efbe6..23cc952a 100644 --- a/tests/test_motor_control.py +++ b/tests/test_motor_control.py @@ -2,7 +2,7 @@ from unittest.mock import MagicMock, patch import pytest -from bec_lib.device import Positioner +from bec_lib.devicemanager import DeviceContainer from bec_widgets.examples import ( MotorControlApp, @@ -20,6 +20,8 @@ from bec_widgets.widgets import ( ) from bec_widgets.widgets.motor_control.motor_control import MotorActions +from .client_mocks import mocked_client + CONFIG_DEFAULT = { "motor_control": { "motor_x": "samx", @@ -52,81 +54,6 @@ CONFIG_DEFAULT = { ], } -####################################################### -# Client and devices fixture -####################################################### - - -class FakeDevice: - """Fake minimal positioner class for testing.""" - - def __init__(self, name, enabled=True, limits=None, read_value=1.0): - super().__init__() - self.name = name - self.enabled = enabled - self.read_value = read_value - self.limits = limits or (-100, 100) # Default limits if not provided - - def read(self): - """Simulates reading the current position of the device.""" - return {self.name: {"value": self.read_value}} - - def move(self, value, relative=False): - """Simulates moving the device to a new position.""" - if relative: - self.read_value += value - else: - self.read_value = value - # Respect the limits - self.read_value = max(min(self.read_value, self.limits[1]), self.limits[0]) - - @property - def readback(self): - return MagicMock(get=MagicMock(return_value=self.read_value)) - - def describe(self): - """Describes the device.""" - return {self.name: {"source": self.name, "dtype": "number", "shape": []}} - - -@pytest.fixture -def mocked_client(): - client = MagicMock() - - # Setup the fake devices - motors = { - "samx": FakeDevice("samx", limits=[-10, 10], read_value=2.0), - "samy": FakeDevice("samy", limits=[-5, 5], read_value=3.0), - "aptrx": FakeDevice("aptrx", read_value=4.0), - "aptry": FakeDevice("aptry", read_value=5.0), - } - - client.device_manager.devices = MagicMock() - client.device_manager.devices.__getitem__.side_effect = lambda x: motors.get(x, FakeDevice(x)) - client.device_manager.devices.enabled_devices = list(motors.values()) - - # Mock the scans.mv method - def mock_mv(*args, relative=False): - # Extracting motor and value pairs - for i in range(0, len(args), 2): - motor = args[i] - value = args[i + 1] - motor.move(value, relative=relative) - return MagicMock(wait=MagicMock()) # Simulate wait method of the move status object - - client.scans = MagicMock(mv=mock_mv) - - # Ensure isinstance check for Positioner passes - original_isinstance = isinstance - - def isinstance_mock(obj, class_info): - if class_info == Positioner: - return True - return original_isinstance(obj, class_info) - - with patch("builtins.isinstance", new=isinstance_mock): - yield client - ####################################################### # Motor Thread @@ -140,7 +67,7 @@ def motor_thread(mocked_client): def test_motor_thread_initialization(mocked_client): motor_thread = MotorThread(client=mocked_client) assert motor_thread.client == mocked_client - assert isinstance(motor_thread.dev, MagicMock) + assert isinstance(motor_thread.dev, DeviceContainer) def test_get_all_motors_names(mocked_client): @@ -175,12 +102,16 @@ def test_move_motor_absolute_by_run(mocked_client): def test_move_motor_relative_by_run(mocked_client): motor_thread = MotorThread(client=mocked_client) + + initial_value = motor_thread.dev["samx"].read()["samx"]["value"] + move_value = 2.0 + expected_value = initial_value + move_value motor_thread.motor = "samx" - motor_thread.value = 2.0 + motor_thread.value = move_value motor_thread.action = MotorActions.MOVE_RELATIVE motor_thread.run() - assert mocked_client.device_manager.devices["samx"].read_value == 4.0 + assert mocked_client.device_manager.devices["samx"].read_value == expected_value def test_motor_thread_move_absolute(motor_thread): @@ -291,8 +222,12 @@ def test_absolute_initialization(motor_absolute_widget): def test_absolute_save_current_coordinates(motor_absolute_widget): - motor_absolute_widget.client.device_manager["samx"].set_value(2.0) - motor_absolute_widget.client.device_manager["samy"].set_value(3.0) + motor_x_value = motor_absolute_widget.client.device_manager.devices["samx"].read()["samx"][ + "value" + ] + motor_y_value = motor_absolute_widget.client.device_manager.devices["samy"].read()["samy"][ + "value" + ] motor_absolute_widget.change_motors("samx", "samy") emitted_coordinates = [] @@ -305,8 +240,7 @@ def test_absolute_save_current_coordinates(motor_absolute_widget): # Trigger saving current coordinates motor_absolute_widget.pushButton_save.click() - # Default position of samx and samy are 2.0 and 3.0 respectively - assert emitted_coordinates == [(2.0, 3.0)] + assert emitted_coordinates == [(motor_x_value, motor_y_value)] def test_absolute_set_absolute_coordinates(motor_absolute_widget): diff --git a/tests/test_motor_map.py b/tests/test_motor_map.py index 26bde3d8..447d7e58 100644 --- a/tests/test_motor_map.py +++ b/tests/test_motor_map.py @@ -5,6 +5,8 @@ import pytest from bec_widgets.widgets import MotorMap +from .client_mocks import mocked_client + CONFIG_DEFAULT = { "plot_settings": { "colormap": "Greys", @@ -61,68 +63,6 @@ CONFIG_ONE_DEVICE = { } -class FakeDevice: - """Fake minimal positioner class for testing.""" - - def __init__(self, name, enabled=True, limits=None, read_value=1.0): - self.name = name - self.enabled = enabled - self.signals = {self.name: {"value": 1.0}} - self.description = {self.name: {"source": self.name}} - self.limits = limits if limits is not None else [0, 0] - self.read_value = read_value - - def set_read_value(self, value): - self.read_value = value - - def read(self): - return {self.name: {"value": self.read_value}} - - def set_limits(self, limits): - self.limits = limits - - def __contains__(self, item): - return item == self.name - - @property - def _hints(self): - return [self.name] - - def set_value(self, fake_value: float = 1.0) -> None: - """ - Setup fake value for device readout - Args: - fake_value(float): Desired fake value - """ - self.signals[self.name]["value"] = fake_value - - def describe(self) -> dict: - """ - Get the description of the device - Returns: - dict: Description of the device - """ - return self.description - - -@pytest.fixture -def mocked_client(): - client = MagicMock() - - # Mocking specific motors with their limits - motors = { - "samx": FakeDevice("samx", limits=[-10, 10], read_value=2.0), - "samy": FakeDevice("samy", limits=[-5, 5], read_value=3.0), - "aptrx": FakeDevice("aptrx", read_value=4.0), - "aptry": FakeDevice("aptry", read_value=5.0), - } - - client.device_manager.devices = MagicMock() - client.device_manager.devices.__getitem__.side_effect = lambda x: motors.get(x, FakeDevice(x)) - - return client - - @pytest.fixture(scope="function") def motor_map(qtbot, mocked_client): widget = MotorMap(client=mocked_client) @@ -144,12 +84,15 @@ def test_motor_limits_initialization(motor_map): def test_motor_initial_position(motor_map): motor_map.precision = 2 + + motor_map_dev = motor_map.client.device_manager.devices + # Example test to check if motor initial positions are correctly initialized expected_positions = { - ("samx", "samx"): 2.0, - ("samy", "samy"): 3.0, - ("aptrx", "aptrx"): 4.0, - ("aptry", "aptry"): 5.0, + ("samx", "samx"): motor_map_dev["samx"].read()["samx"]["value"], + ("samy", "samy"): motor_map_dev["samy"].read()["samy"]["value"], + ("aptrx", "aptrx"): motor_map_dev["aptrx"].read()["aptrx"]["value"], + ("aptry", "aptry"): motor_map_dev["aptry"].read()["aptry"]["value"], } for (motor_name, entry), expected_position in expected_positions.items(): actual_position = motor_map._get_motor_init_position(motor_name, entry)