mirror of
https://github.com/bec-project/bec_widgets.git
synced 2025-07-14 03:31:50 +02:00
test: mock_client unified for all tests
This commit is contained in:
@ -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
|
||||
|
@ -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 = {
|
||||
|
@ -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
|
||||
|
@ -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):
|
||||
|
@ -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)
|
||||
|
Reference in New Issue
Block a user