Testing BEC Widgets#
Importance of Writing Tests for Widgets#
Writing tests for widgets, even for simple ones, is crucial in maintaining the reliability and stability of a software project. Tests help ensure that new contributions don’t unintentionally break existing functionality, and they provide a safety net for developers making changes in the future. In the context of the BEC Widgets, testing is particularly important due to the complexity of interactions with external systems like the BEC server.
Testing with Pytest and QtBot#
For testing Qt-based applications, we use pytest along with
the pytest-qt plugin, which provides the qtbot fixture. qtbot is
specifically designed to simplify interactions with Qt applications during testing. It handles the creation, management,
and cleanup of widgets, ensuring that your tests are robust and do not leave behind any lingering resources or open
windows.
Fixtures for Testing BEC Widgets#
Let’s break down the key fixtures used in testing BEC Widgets:
qapplicationFixture: This fixture ensures that all Qt applications and widgets are properly closed after each test. It usesqtbotto wait until all top-level widgets are closed, raising an error if any remain open.rpc_registerFixture: This fixture manages theRPCRegistersingleton, ensuring that it is reset after each test. This prevents state from leaking between tests, which could cause unexpected behavior.bec_dispatcherFixture: This fixture initializes theBECDispatcherand ensures that all connections are properly disconnected and the BEC client is shut down after each test. Likerpc_register, it resets the singleton after each test.clean_singletonFixture: This fixture cleans up any singleton instances used in error popups, preventing interference between tests.create_widgetHelper Function: This function is a helper that should be used in all tests requiring widget creation. It ensures that widgets are properly added toqtbot, which manages their lifecycle during tests. We highly recommend using this function to create widgets in your tests to ensure proper cleanup and compatibility with otherautousefixtures.
Note
These fixtures are automatically applied to all tests within the tests/unit_tests directory, ensuring consistency and proper cleanup between tests. You can find all unit test fixtures in the conftest.py file located in the tests/unit_tests directory of the BEC Widgets repository.
View code: Conftest with Fixtures
import json
import time
from unittest import mock
from unittest.mock import patch
import fakeredis
import h5py
import numpy as np
import pytest
from bec_lib import messages, service_config
from bec_lib.bec_service import messages
from bec_lib.client import BECClient
from bec_lib.messages import _StoredDataInfo
from bec_qthemes import apply_theme
from bec_qthemes._theme import Theme
from ophyd._pyepics_shim import _dispatcher
from pytestqt.exceptions import TimeoutError as QtBotTimeoutError
from qtpy.QtCore import QEvent, QEventLoop
from qtpy.QtWidgets import QApplication, QMessageBox
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.tests.utils import DEVICES, DMMock
from bec_widgets.utils import bec_dispatcher as bec_dispatcher_module
from bec_widgets.utils import error_popups
from bec_widgets.utils.bec_dispatcher import QtRedisConnector
# Patch to set default RAISE_ERROR_DEFAULT to True for tests
# This means that by default, error popups will raise exceptions during tests
# error_popups.RAISE_ERROR_DEFAULT = True
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
# execute all other hooks to obtain the report object
outcome = yield
rep = outcome.get_result()
item.stash["failed"] = rep.failed
def process_all_deferred_deletes(qapp):
qapp.sendPostedEvents(None, QEvent.DeferredDelete)
qapp.processEvents(QEventLoop.AllEvents)
@pytest.fixture(autouse=True)
def qapplication(qtbot, request, testable_qtimer_class): # pylint: disable=unused-argument
qapp = QApplication.instance()
process_all_deferred_deletes(qapp)
if (
not hasattr(qapp, "theme")
or not isinstance(qapp.theme, Theme)
or qapp.theme.theme != "light"
):
apply_theme("light")
qapp.processEvents()
yield
# if the test failed, we don't want to check for open widgets as
# it simply pollutes the output
# stop pyepics dispatcher for leaking tests
_dispatcher.stop()
if request.node.stash._storage.get("failed"):
print("Test failed, skipping cleanup checks")
return
bec_dispatcher = bec_dispatcher_module.BECDispatcher()
bec_dispatcher.stop_cli_server()
testable_qtimer_class.check_all_stopped(qtbot)
qapp.processEvents()
if hasattr(qapp, "os_listener") and qapp.os_listener:
qapp.removeEventFilter(qapp.os_listener)
try:
qtbot.waitUntil(lambda: qapp.topLevelWidgets() == [])
except QtBotTimeoutError as exc:
raise TimeoutError(f"Failed to close all widgets: {qapp.topLevelWidgets()}") from exc
@pytest.fixture(autouse=True)
def rpc_register():
yield RPCRegister()
RPCRegister.reset_singleton()
_REDIS_CONN: QtRedisConnector | None = None
def global_mock_qt_redis_connector(*_, **__):
global _REDIS_CONN
if _REDIS_CONN is None:
_REDIS_CONN = QtRedisConnector(bootstrap="localhost:1", redis_cls=fakeredis.FakeRedis)
return _REDIS_CONN
def mock_client(*_, **__):
with (
patch("bec_lib.client.DeviceManagerBase", DMMock),
patch("bec_lib.client.DAPPlugins"),
patch("bec_lib.client.Scans"),
patch("bec_lib.client.ScanManager"),
patch("bec_lib.bec_service.BECAccess"),
):
client = BECClient(
config=service_config.ServiceConfig(config={"redis": {"host": "localhost", "port": 1}}),
connector_cls=global_mock_qt_redis_connector,
)
client.start()
client.device_manager.add_devices(DEVICES)
return client
@pytest.fixture(autouse=True)
def bec_dispatcher(threads_check): # pylint: disable=unused-argument
with mock.patch.object(bec_dispatcher_module, "BECClient", mock_client):
bec_dispatcher = bec_dispatcher_module.BECDispatcher()
yield bec_dispatcher
bec_dispatcher.disconnect_all()
# clean BEC client
bec_dispatcher.client.shutdown()
# stop the cli server
bec_dispatcher.stop_cli_server()
# reinitialize singleton for next test
bec_dispatcher_module.BECDispatcher.reset_singleton()
@pytest.fixture(autouse=True)
def clean_singleton():
error_popups._popup_utility_instance = None
@pytest.fixture(autouse=True)
def suppress_message_box(monkeypatch):
"""
Auto-suppress any QMessageBox.exec_ calls by returning Ok immediately.
"""
monkeypatch.setattr(QMessageBox, "exec_", lambda *args, **kwargs: QMessageBox.Ok)
def create_widget(qtbot, widget, *args, **kwargs):
"""
Create a widget and add it to the qtbot for testing. This is a helper function that
should be used in all tests that require a widget to be created.
Args:
qtbot (fixture): pytest-qt fixture
widget (QWidget): widget class to be created
*args: positional arguments for the widget
**kwargs: keyword arguments for the widget
Returns:
QWidget: the created widget
"""
widget = widget(*args, **kwargs)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
return widget
def create_history_file(file_path, data: dict, metadata: dict) -> messages.ScanHistoryMessage:
"""
Helper to create a history file with the given data.
The data should contain readout groups, e.g.
{
"baseline": {"samx": {"samx": {"value": [1, 2, 3], "timestamp": [100, 200, 300]}},
"monitored": {"bpm4i": {"bpm4i": {"value": [5, 6, 7], "timestamp": [101, 201, 301]}}},
"async": {"async_device": {"async_device": {"value": [1, 2, 3], "timestamp": [11, 21, 31]}}},
}
"""
with h5py.File(file_path, "w") as f:
_metadata = f.create_group("entry/collection/metadata")
_metadata.create_dataset("sample_name", data="test_sample")
metadata_bec = f.create_group("entry/collection/metadata/bec")
for key, value in metadata.items():
if isinstance(value, dict):
metadata_bec.create_group(key)
for sub_key, sub_value in value.items():
if isinstance(sub_value, list):
sub_value = json.dumps(sub_value)
metadata_bec[key].create_dataset(sub_key, data=sub_value)
elif isinstance(sub_value, dict):
for sub_sub_key, sub_sub_value in sub_value.items():
sub_sub_group = metadata_bec[key].create_group(sub_key)
# Handle _StoredDataInfo objects
if isinstance(sub_sub_value, _StoredDataInfo):
# Store the numeric shape
sub_sub_group.create_dataset("shape", data=sub_sub_value.shape)
# Store the dtype as a UTF-8 string
dt = sub_sub_value.dtype or ""
sub_sub_group.create_dataset(
"dtype", data=dt, dtype=h5py.string_dtype(encoding="utf-8")
)
continue
if isinstance(sub_sub_value, list):
json_val = json.dumps(sub_sub_value)
sub_sub_group.create_dataset(sub_sub_key, data=json_val)
elif isinstance(sub_sub_value, dict):
for k2, v2 in sub_sub_value.items():
val = json.dumps(v2) if isinstance(v2, list) else v2
sub_sub_group.create_dataset(k2, data=val)
else:
sub_sub_group.create_dataset(sub_sub_key, data=sub_sub_value)
else:
metadata_bec[key].create_dataset(sub_key, data=sub_value)
else:
metadata_bec.create_dataset(key, data=value)
for group, devices in data.items():
readout_group = f.create_group(f"entry/collection/readout_groups/{group}")
for device, device_data in devices.items():
dev_group = f.create_group(f"entry/collection/devices/{device}")
for signal, signal_data in device_data.items():
signal_group = dev_group.create_group(signal)
for signal_key, signal_values in signal_data.items():
signal_group.create_dataset(signal_key, data=signal_values)
readout_group[device] = h5py.SoftLink(f"/entry/collection/devices/{device}")
msg = messages.ScanHistoryMessage(
scan_id=metadata["scan_id"],
scan_name=metadata["scan_name"],
exit_status=metadata["exit_status"],
file_path=file_path,
scan_number=metadata["scan_number"],
dataset_number=metadata["dataset_number"],
start_time=time.time(),
end_time=time.time(),
num_points=metadata["num_points"],
request_inputs=metadata["request_inputs"],
stored_data_info=metadata.get("stored_data_info"),
metadata={"scan_report_devices": metadata.get("scan_report_devices")},
)
return msg
@pytest.fixture
def grid_scan_history_msg(tmpdir):
x_grid, y_grid = np.meshgrid(np.linspace(-5, 5, 10), np.linspace(-5, 5, 10))
x_flat = x_grid.T.ravel()
y_flat = y_grid.T.ravel()
positions = np.vstack((x_flat, y_flat)).T
num_points = len(positions)
data = {
"baseline": {"bpm1a": {"bpm1a": {"value": [1], "timestamp": [100]}}},
"monitored": {
"bpm4i": {
"bpm4i": {
"value": np.random.rand(num_points),
"timestamp": np.random.rand(num_points),
}
},
"samx": {"samx": {"value": x_flat, "timestamp": np.random.rand(num_points)}},
"samy": {"samy": {"value": y_flat, "timestamp": np.random.rand(num_points)}},
},
"async": {
"async_device": {
"async_device": {
"value": np.random.rand(num_points * 10),
"timestamp": np.random.rand(num_points * 10),
}
}
},
}
metadata = {
"scan_id": "test_scan",
"scan_name": "grid_scan",
"scan_type": "step",
"exit_status": "closed",
"scan_number": 1,
"dataset_number": 1,
"request_inputs": {
"arg_bundle": ["samx", -5, 5, 10, "samy", -5, 5, 10],
"kwargs": {"relative": True},
},
"positions": positions.tolist(),
"num_points": num_points,
}
file_path = str(tmpdir.join("scan_1.h5"))
return create_history_file(file_path, data, metadata)
@pytest.fixture
def scan_history_factory(tmpdir):
"""
Factory to create scan history messages with custom parameters.
Usage:
msg1 = scan_history_factory(scan_id="id1", scan_number=1, num_points=10)
msg2 = scan_history_factory(scan_id="id2", scan_number=2, scan_name="grid_scan", num_points=16)
"""
def _factory(
scan_id: str = "test_scan",
scan_number: int = 1,
dataset_number: int = 1,
scan_name: str = "line_scan",
scan_type: str = "step",
num_points: int = 10,
x_range: tuple = (-5, 5),
y_range: tuple = (-5, 5),
):
# Generate positions based on scan type
if scan_name == "grid_scan":
grid_size = int(np.sqrt(num_points))
x_grid, y_grid = np.meshgrid(
np.linspace(x_range[0], x_range[1], grid_size),
np.linspace(y_range[0], y_range[1], grid_size),
)
x_flat = x_grid.T.ravel()
y_flat = y_grid.T.ravel()
else:
x_flat = np.linspace(x_range[0], x_range[1], num_points)
y_flat = np.linspace(y_range[0], y_range[1], num_points)
positions = np.vstack((x_flat, y_flat)).T
num_pts = len(positions)
# Create dummy data
data = {
"baseline": {"bpm1a": {"bpm1a": {"value": [1], "timestamp": [100]}}},
"monitored": {
"bpm4i": {
"bpm4i": {
"value": np.random.rand(num_points),
"timestamp": np.random.rand(num_points),
}
},
"bpm3a": {
"bpm3a": {
"value": np.random.rand(num_points),
"timestamp": np.random.rand(num_points),
}
},
"samx": {"samx": {"value": x_flat, "timestamp": np.arange(num_pts)}},
"samy": {"samy": {"value": y_flat, "timestamp": np.arange(num_pts)}},
},
"async": {
"async_device": {
"async_device": {
"value": np.random.rand(num_pts * 10),
"timestamp": np.random.rand(num_pts * 10),
}
}
},
}
metadata = {
"scan_id": scan_id,
"scan_name": scan_name,
"scan_type": scan_type,
"exit_status": "closed",
"scan_number": scan_number,
"dataset_number": dataset_number,
"request_inputs": {
"arg_bundle": [
"samx",
x_range[0],
x_range[1],
num_pts,
"samy",
y_range[0],
y_range[1],
num_pts,
],
"kwargs": {"relative": True},
},
"positions": positions.tolist(),
"num_points": num_pts,
"stored_data_info": {
"samx": {"samx": _StoredDataInfo(shape=(num_points,), dtype="float64")},
"samy": {"samy": _StoredDataInfo(shape=(num_points,), dtype="float64")},
"bpm4i": {"bpm4i": _StoredDataInfo(shape=(10,), dtype="float64")},
"async_device": {
"async_device": _StoredDataInfo(shape=(num_points * 10,), dtype="float64")
},
},
"scan_report_devices": [b"samx"],
}
file_path = str(tmpdir.join(f"{scan_id}.h5"))
return create_history_file(file_path, data, metadata)
return _factory
Example Test for PositionerBox#
Below is an example of how to write a simple test for the PositionerBox widget,
utilizing the fixtures mentioned above:
View code: PositionerBox Widget Unit Tests
from unittest import mock
import pytest
from bec_lib.endpoints import MessageEndpoints
from bec_lib.messages import ScanQueueMessage
from qtpy.QtCore import Qt, QTimer
from qtpy.QtGui import QValidator
from qtpy.QtWidgets import QPushButton
from bec_widgets.tests.utils import Positioner
from bec_widgets.widgets.control.device_control.positioner_box import (
PositionerBox,
PositionerControlLine,
)
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import (
DeviceLineEdit,
)
from .client_mocks import mocked_client
from .conftest import create_widget
class PositionerWithoutPrecision(Positioner):
"""just placeholder for testing embedded isinstance check in DeviceCombobox"""
def __init__(self, precision, name="test", limits=None, read_value=1.0, enabled=True):
super().__init__(name, limits=limits, read_value=read_value, enabled=enabled)
self._precision = precision
@property
def precision(self):
return self._precision
@pytest.fixture
def positioner_box(qtbot, mocked_client):
"""Fixture for PositionerBox widget"""
with mock.patch(
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base.uuid.uuid4"
) as mock_uuid:
mock_uuid.return_value = "fake_uuid"
with mock.patch(
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base.PositionerBoxBase._check_device_is_valid",
return_value=True,
):
db = create_widget(qtbot, PositionerBox, device="samx", client=mocked_client)
yield db
def test_positioner_box(positioner_box):
"""Test init of positioner box"""
assert positioner_box.device == "samx"
data = positioner_box.dev["samx"].read(cached=True)
# Avoid check for Positioner class from BEC in _init_device
setpoint_text = positioner_box.ui.setpoint.text()
# check that the setpoint is taken correctly after init
assert float(setpoint_text) == data["samx_setpoint"]["value"]
# check that the precision is taken correctly after isnit
precision = positioner_box.dev["samx"].precision
assert setpoint_text == f"{data['samx_setpoint']['value']:.{precision}f}"
# check that the step size is set according to the device precision
assert positioner_box.ui.step_size.value() == 10**-precision * 10
def test_positioner_box_update_limits(positioner_box):
"""Test update of limits"""
positioner_box._limits = None
positioner_box.update_limits([0, 10])
assert positioner_box._limits == [0, 10]
assert positioner_box.setpoint_validator.bottom() == 0
assert positioner_box.setpoint_validator.top() == 10
assert positioner_box.setpoint_validator.validate("100", 0) == (
QValidator.State.Intermediate,
"100",
0,
)
positioner_box.update_limits(None)
assert positioner_box._limits is None
assert positioner_box.setpoint_validator.validate("100", 0) == (
QValidator.State.Acceptable,
"100",
0,
)
def test_positioner_box_on_stop(positioner_box):
"""Test on stop button"""
with mock.patch.object(positioner_box.client.connector, "send") as mock_send:
positioner_box.on_stop()
params = {"device": "samx", "rpc_id": "fake_uuid", "func": "stop", "args": [], "kwargs": {}}
msg = ScanQueueMessage(
scan_type="device_rpc",
parameter=params,
queue="emergency",
metadata={"RID": "fake_uuid", "response": False},
)
mock_send.assert_called_once_with(
MessageEndpoints.scan_queue_request(positioner_box.client.username), msg
)
def test_positioner_box_setpoint_change(positioner_box):
"""Test positioner box setpoint change"""
with mock.patch.object(positioner_box.dev["samx"], "move") as mock_move:
positioner_box.ui.setpoint.setText("100")
positioner_box.on_setpoint_change()
mock_move.assert_called_once_with(100, relative=False)
def test_positioner_box_on_tweak_right(positioner_box):
"""Test tweak right button"""
with mock.patch.object(positioner_box.dev["samx"], "move") as mock_move:
positioner_box.ui.step_size.setValue(0.1)
positioner_box.on_tweak_right()
mock_move.assert_called_once_with(0.1, relative=True)
def test_positioner_box_on_tweak_left(positioner_box):
"""Test tweak left button"""
with mock.patch.object(positioner_box.dev["samx"], "move") as mock_move:
positioner_box.ui.step_size.setValue(0.1)
positioner_box.on_tweak_left()
mock_move.assert_called_once_with(-0.1, relative=True)
def test_positioner_box_setpoint_out_of_range(positioner_box):
"""Test setpoint out of range"""
positioner_box.update_limits([0, 10])
positioner_box.ui.setpoint.setText("100")
positioner_box.on_setpoint_change()
assert positioner_box.ui.setpoint.text() == "100"
assert positioner_box.ui.setpoint.hasAcceptableInput() == False
def test_positioner_control_line(qtbot, mocked_client):
"""Test PositionerControlLine.
Inherits from PositionerBox, but the layout is changed. Check dimensions only
"""
with mock.patch(
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box_base.uuid.uuid4"
) as mock_uuid:
mock_uuid.return_value = "fake_uuid"
with mock.patch(
"bec_widgets.widgets.control.device_control.positioner_box.positioner_box.positioner_box.PositionerBox._check_device_is_valid",
return_value=True,
):
db = PositionerControlLine(device="samx", client=mocked_client)
qtbot.addWidget(db)
assert db.ui.device_box.height() == db.height()
assert db.ui.device_box.height() >= db.dimensions[0]
assert db.ui.device_box.width() == 600
def test_positioner_box_open_dialog_selection(qtbot, positioner_box):
"""Test open positioner edit"""
# Use a timer to close the dialog after it opens
def close_dialog():
# pylint: disable=protected-access
assert positioner_box._dialog is not None
qtbot.waitUntil(lambda: positioner_box._dialog.isVisible() is True, timeout=1000)
line_edit = positioner_box._dialog.findChild(DeviceLineEdit)
line_edit.setText("samy")
close_button = positioner_box._dialog.findChild(QPushButton)
assert close_button.text() == "Close"
qtbot.mouseClick(close_button, Qt.LeftButton)
# Execute the timer after the dialog opens to close it
QTimer.singleShot(100, close_dialog)
qtbot.mouseClick(positioner_box.ui.tool_button, Qt.LeftButton)
assert positioner_box.device == "samy"
def test_device_validity_check_rejects_non_positioner():
# isinstance checks for PositionerBox are mocked out in the mock client
positioner_box = mock.MagicMock(spec=PositionerBox)
positioner_box.dev = {"test": 5.123}
assert not PositionerBox._check_device_is_valid(positioner_box, "test")
def test_positioner_box_device_without_precision(qtbot, positioner_box):
"""Test positioner box with device without precision"""
for ii, mock_return in enumerate([None, 2, 2.0, True, "tmp"]):
dev_name = f"samy_{ii}"
device = PositionerWithoutPrecision(
precision=mock_return, name=dev_name, limits=[-5, 5], read_value=3.0
)
positioner_box.bec_dispatcher.client.device_manager.add_devices(devices=[device])
positioner_box.device = dev_name
def check_title():
return positioner_box.ui.device_box.title() == dev_name
qtbot.waitUntil(check_title, timeout=3000)
if isinstance(mock_return, (int, float)):
mock_return = int(mock_return)
assert positioner_box.ui.step_size.value() == 10**-mock_return * 10
else:
assert positioner_box.ui.step_size.value() == 10**-8 * 10
Key Points in the Test:#
Fixture Use: The
positioner_boxfixture handles widget creation and mocking of external dependencies. This ensures the test runs in isolation and doesn’t rely on actual hardware or network connections.Assertion Checks: The test includes several assertions to verify that the widget initializes correctly, including checking the setpoint, precision, and step size.
Conclusion#
By writing tests like the one shown above, you help ensure that your widget behaves as expected. Tests also provide a way to automatically verify that new changes do not introduce regressions. This is particularly important in a collaborative environment where multiple developers are contributing to the same codebase. Your tests not only safeguard your code but also provide confidence to others that their contributions won’t break existing functionality.