From 7262291818fdfff32a99a1e14eae142db56707f6 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Wed, 23 Sep 2026 17:41:53 +0200 Subject: [PATCH] feat(tests): ship widget test fixtures importable by plugin repositories --- AGENTS.md | 24 +- bec_widgets/tests/fixtures.py | 351 +++++++++++++++++++++++++ bec_widgets/tests/utils.py | 185 +++++++------ tests/conftest.py | 20 -- tests/unit_tests/conftest.py | 345 +----------------------- tests/unit_tests/test_testing_utils.py | 15 ++ 6 files changed, 482 insertions(+), 458 deletions(-) create mode 100644 bec_widgets/tests/fixtures.py delete mode 100644 tests/conftest.py diff --git a/AGENTS.md b/AGENTS.md index 0fce8287..45dc580f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -35,7 +35,7 @@ Start here when orienting yourself: - `bec_widgets/utils/error_popups.py` — `SafeSlot` and user-visible exception handling - `bec_widgets/utils/generate_cli.py` — generated RPC and Designer code entry point - `bec_widgets/utils/bec_plugin_helper.py` — plugin discovery and entry points -- `tests/unit_tests/conftest.py` — shared widget fixtures and `create_widget(...)` +- `bec_widgets/tests/fixtures.py` — shared widget test fixtures, also imported by plugin repos - `pyproject.toml` — scripts, tooling, and dependency source of truth ## Repo Layout @@ -84,8 +84,8 @@ If you change: when the visual change is intentional - `bec_widgets/utils/bec_dispatcher.py`, `bec_widget.py`, or shared plumbing used by many widgets: run the relevant focused tests plus the broader affected package test scope before finishing -- test behavior or a flaky widget test: check `tests/unit_tests/conftest.py` and `bec_widgets/tests/` - for reusable fixtures and helpers before adding new ones +- test behavior or a flaky widget test: check `bec_widgets/tests/` for reusable fixtures and helpers + before adding new ones - docs, examples, or commands only: no broad GUI or e2e run is required unless commands or runnable examples changed @@ -155,13 +155,17 @@ Run the smallest relevant test target first. For substantial UI plumbing changes changes, or work that affects many widgets, run the broader affected package suite before finishing. Unit tests are the default. CI runs them with `--random-order`, so local validation should do the same -when practical. Create widgets with `create_widget(...)` from `tests/unit_tests/conftest.py`. It -registers the widget with `qtbot` so it is closed at test end; the autouse conftest fixtures handle the -rest of the teardown (dispatcher disconnect, singleton resets) and fail the test if any top-level -widget is left open. Before adding a new fixture, check for reusable fixtures in -`tests/unit_tests/conftest.py` and the packaged helpers in `bec_widgets/tests/`: `client_mocks.py` -(`mocked_client`, `mocked_client_with_dap`), `fake_devices.py`, and `utils.py`, which holds the -bodies of the autouse fixtures so plugin repositories can reuse them. +when practical. Create widgets with `create_widget(...)` from `bec_widgets/tests/utils.py`. It +registers the widget with `qtbot` so it is closed at test end; the autouse fixtures handle the rest of +the teardown (dispatcher disconnect, singleton resets) and fail the test if any top-level widget is +left open or a timer is still running. + +The fixtures live in the package so plugin repositories get the same test bed: +`bec_widgets/tests/fixtures.py` holds all of them (autouse fixtures, `mocked_client`, +`mocked_client_with_dap`, scan history fixtures), `client_mocks.py` and `fake_devices.py` the mocked +client and fake devices, `utils.py` plain helpers. `tests/unit_tests/conftest.py` only star-imports +`bec_widgets.tests.fixtures`, exactly as a plugin's widget-test conftest does, so add new shared +fixtures there rather than to the conftest. Mock BEC, Redis, and hardware in unit tests. Reuse existing helpers such as `FakeDevice`, `FakePositioner`, `DMMock`, and the shared autouse fixtures rather than rolling your own. diff --git a/bec_widgets/tests/fixtures.py b/bec_widgets/tests/fixtures.py new file mode 100644 index 00000000..7078d1db --- /dev/null +++ b/bec_widgets/tests/fixtures.py @@ -0,0 +1,351 @@ +""" +Pytest fixtures for testing BEC widgets against a mocked BEC client, shared with plugin repos. + +Import all of them into the ``conftest.py`` next to the widget tests:: + + from bec_widgets.tests.fixtures import * # noqa: F401,F403 + +The autouse fixtures reset the QApplication and the BEC singletons around every test, back the +dispatcher with a BECClient on fakeredis that knows the fake devices of +:mod:`bec_widgets.tests.fake_devices`, and fail a passing test that leaves a top-level widget open +or a timer running. ``mocked_client``, ``mocked_client_with_dap`` and the scan history fixtures are +available on request, :func:`bec_widgets.tests.utils.create_widget` creates widgets under test. + +The fixtures are deliberately not registered as a pytest plugin, so that they only apply below the +conftest that imports them and leave the non-GUI tests of a repository alone. Importing this module +replaces ``qtpy.QtCore.QTimer`` with :class:`TestableQTimer`, which only affects widget modules +imported afterwards: import it in a conftest that is loaded before the widgets under test. + +Requires ``bec_widgets[dev]`` (pytest-qt, fakeredis) and bec_lib's ``threads_check`` fixture. +""" + +# pylint: disable=redefined-outer-name,unused-argument,wrong-import-position +import os + +# Keep ophyd on its dummy control layer, so importing it does not create a real EPICS CA context +os.environ.setdefault("OPHYD_CONTROL_LAYER", "dummy") + +import qtpy.QtCore + +from bec_widgets.tests.testable_qtimer import TestableQTimer + +# To support 'from qtpy.QtCore import QTimer' syntax, QTimer is replaced for the whole session +# see: https://docs.python.org/3/library/unittest.mock.html#where-to-patch +qtpy.QtCore.QTimer = TestableQTimer + +# isort: split + +from unittest import mock + +import numpy as np +import pytest +from bec_lib.messages import _StoredDataInfo +from bec_qthemes import apply_theme +from bec_qthemes._theme import Theme +from ophyd._dummy_shim import _dispatcher +from pytestqt.exceptions import TimeoutError as QtBotTimeoutError +from qtpy.QtWidgets import QApplication, QMessageBox + +from bec_widgets.tests.client_mocks import dap_plugin_message, mocked_client, mocked_client_with_dap +from bec_widgets.tests.utils import create_history_file, mock_client, process_all_deferred_deletes +from bec_widgets.utils import bec_dispatcher as bec_dispatcher_module +from bec_widgets.utils import error_popups +from bec_widgets.utils.rpc_register import RPCRegister +from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_banner import ( + BECNotificationBroker, +) + +__all__ = [ + "pytest_runtest_makereport", + "_capture_test_name_in_qtimer", + "_register_worker_thread_dummies", + "testable_qtimer_class", + "qapplication", + "rpc_register", + "bec_dispatcher", + "clean_singleton", + "suppress_message_box", + "mocked_client", + "mocked_client_with_dap", + "dap_plugin_message", + "grid_scan_history_msg", + "scan_history_factory", +] + + +@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 + + +@pytest.fixture(autouse=True) +def _capture_test_name_in_qtimer(request): + TestableQTimer._current_test_name = request.node.name + yield + TestableQTimer._current_test_name = "" + + +@pytest.fixture +def testable_qtimer_class(): + return TestableQTimer + + +@pytest.fixture(autouse=True) +def qapplication(qtbot, request, testable_qtimer_class): + 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() + from bec_widgets.widgets.editors.bec_console.bec_console import _bec_console_registry + + _bec_console_registry.clear() + process_all_deferred_deletes(qapp) + 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() + + +@pytest.fixture(autouse=True) +def bec_dispatcher(threads_check): + 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 notification_broker_reset(bec_dispatcher): + """Tear down the app-wide notification broker before the dispatcher it is bound to.""" + yield + BECNotificationBroker.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) + + +@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", + "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, + "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 + + +@pytest.fixture(scope="session", autouse=True) +def _register_worker_thread_dummies(): + """Qt thread-pool threads register a permanent ``threading._DummyThread`` the first + time Python inspects them (e.g. a loguru call inside a background Worker). Saturate + the global pool once up front - with thread expiry disabled so the threads persist - + so bec_lib's threads_check fixture never sees these registrations appear mid-test.""" + import threading + + from qtpy.QtCore import QRunnable, QThreadPool + + pool = QThreadPool.globalInstance() + pool.setExpiryTimeout(-1) + barrier = threading.Barrier(pool.maxThreadCount() + 1, timeout=10) + + class _Warmup(QRunnable): + def run(self): + threading.current_thread() + try: + barrier.wait() + except threading.BrokenBarrierError: + pass + + for _ in range(pool.maxThreadCount()): + pool.start(_Warmup()) + try: + barrier.wait() + except threading.BrokenBarrierError: + pass diff --git a/bec_widgets/tests/utils.py b/bec_widgets/tests/utils.py index 43d5316b..69e3a517 100644 --- a/bec_widgets/tests/utils.py +++ b/bec_widgets/tests/utils.py @@ -1,31 +1,21 @@ """ -Shared bodies of the autouse fixtures of the bec_widgets unit tests. - -The ``*_fixture`` generators are meant to be wrapped in local fixtures, so that plugin repositories -can run their widget tests against the same mocked BEC client and leak checks, e.g.:: - - @pytest.fixture(autouse=True) - def bec_dispatcher(threads_check): - yield from bec_widgets.tests.utils.bec_dispatcher_fixture() +Helpers for testing BEC widgets against a mocked BEC client. The fixtures built on top of them are +in :mod:`bec_widgets.tests.fixtures`. """ -from unittest import mock +import json +import time from unittest.mock import patch import fakeredis -from bec_lib import service_config +import h5py +from bec_lib import messages, service_config from bec_lib.client import BECClient -from bec_qthemes import apply_theme -from bec_qthemes._theme import Theme -from pytestqt.exceptions import TimeoutError as QtBotTimeoutError +from bec_lib.messages import _StoredDataInfo from qtpy.QtCore import QEvent, QEventLoop -from qtpy.QtWidgets import QApplication from bec_widgets.tests.fake_devices 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 -from bec_widgets.utils.rpc_register import RPCRegister def process_all_deferred_deletes(qapp): @@ -33,58 +23,6 @@ def process_all_deferred_deletes(qapp): qapp.processEvents(QEventLoop.AllEvents) -def qapplication_fixture(qtbot, request, testable_qtimer_class): - """ - Reset the application to the light theme, then check after the test that all timers are - stopped and all top-level widgets are closed. The checks are skipped for failed tests, which - requires a ``pytest_runtest_makereport`` hook that stores ``item.stash["failed"]``. - """ - 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 - - # Imported here so that the test session can select the ophyd control layer before ophyd loads - from ophyd._dummy_shim import _dispatcher - - from bec_widgets.widgets.editors.bec_console.bec_console import _bec_console_registry - - # stop pyepics dispatcher for leaking tests - _dispatcher.stop() - _bec_console_registry.clear() - process_all_deferred_deletes(qapp) - # if the test failed, we don't want to check for open widgets as - # it simply pollutes the output - 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 - - -def rpc_register_fixture(): - """Provide the RPCRegister singleton and reset it after the test.""" - yield RPCRegister() - RPCRegister.reset_singleton() - - _REDIS_CONN: QtRedisConnector | None = None @@ -113,25 +51,98 @@ def mock_client(*_, **__): return client -def bec_dispatcher_fixture(): +def create_widget(qtbot, widget, *args, **kwargs): """ - Provide a BECDispatcher backed by :func:`mock_client` and tear it down after the test. The - wrapping fixture should depend on bec_lib's ``threads_check`` so that the thread check runs - after the client is shut down. + 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 """ - 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() + widget = widget(*args, **kwargs) + qtbot.addWidget(widget) + qtbot.waitExposed(widget) + return widget -def clean_singleton_fixture(): - """Drop the error popup singleton before the test.""" - error_popups._popup_utility_instance = None - yield +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["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 diff --git a/tests/conftest.py b/tests/conftest.py deleted file mode 100644 index d2cbd9d0..00000000 --- a/tests/conftest.py +++ /dev/null @@ -1,20 +0,0 @@ -import pytest -import qtpy.QtCore - -from bec_widgets.tests.testable_qtimer import TestableQTimer - -# To support 'from qtpy.QtCore import QTimer' syntax we just replace this completely for the test session -# see: https://docs.python.org/3/library/unittest.mock.html#where-to-patch -qtpy.QtCore.QTimer = TestableQTimer - - -@pytest.fixture(autouse=True) -def _capture_test_name_in_qtimer(request): - TestableQTimer._current_test_name = request.node.name - yield - TestableQTimer._current_test_name = "" - - -@pytest.fixture -def testable_qtimer_class(): - return TestableQTimer diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index be64f8f8..43bd3f37 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -1,341 +1,4 @@ -# Force ophyd onto its dummy control layer in tests so importing it does not -# try to create a real EPICS CA context. -import os - -os.environ.setdefault("OPHYD_CONTROL_LAYER", "dummy") -import json -import time - -import h5py -import numpy as np -import pytest -from bec_lib import messages -from bec_lib.messages import _StoredDataInfo -from qtpy.QtWidgets import QMessageBox - -from bec_widgets.tests import utils as test_utils -from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_banner import ( - BECNotificationBroker, -) - -# 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 - - -@pytest.fixture(autouse=True) -def qapplication(qtbot, request, testable_qtimer_class): # pylint: disable=unused-argument - yield from test_utils.qapplication_fixture(qtbot, request, testable_qtimer_class) - - -@pytest.fixture(autouse=True) -def rpc_register(): - yield from test_utils.rpc_register_fixture() - - -@pytest.fixture(autouse=True) -def bec_dispatcher(threads_check): # pylint: disable=unused-argument - yield from test_utils.bec_dispatcher_fixture() - - -@pytest.fixture(autouse=True) -def notification_broker_reset(bec_dispatcher): # pylint: disable=unused-argument - """Tear down the app-wide notification broker before the dispatcher it is bound to.""" - yield - BECNotificationBroker.reset_singleton() - - -@pytest.fixture(autouse=True) -def clean_singleton(): - yield from test_utils.clean_singleton_fixture() - - -@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["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", - "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, - "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 - - -@pytest.fixture(scope="session", autouse=True) -def _register_worker_thread_dummies(): - """Qt thread-pool threads register a permanent ``threading._DummyThread`` the first - time Python inspects them (e.g. a loguru call inside a background Worker). Saturate - the global pool once up front - with thread expiry disabled so the threads persist - - so bec_lib's threads_check fixture never sees these registrations appear mid-test.""" - import threading - - from qtpy.QtCore import QRunnable, QThreadPool - - pool = QThreadPool.globalInstance() - pool.setExpiryTimeout(-1) - barrier = threading.Barrier(pool.maxThreadCount() + 1, timeout=10) - - class _Warmup(QRunnable): - def run(self): - threading.current_thread() - try: - barrier.wait() - except threading.BrokenBarrierError: - pass - - for _ in range(pool.maxThreadCount()): - pool.start(_Warmup()) - try: - barrier.wait() - except threading.BrokenBarrierError: - pass +# The shared widget test fixtures, imported the same way plugin repositories use them. +# pylint: disable=wildcard-import,unused-wildcard-import,unused-import +from bec_widgets.tests.fixtures import * # noqa: F401,F403 +from bec_widgets.tests.utils import create_widget # noqa: F401 diff --git a/tests/unit_tests/test_testing_utils.py b/tests/unit_tests/test_testing_utils.py index 9ec88370..26bf5c04 100644 --- a/tests/unit_tests/test_testing_utils.py +++ b/tests/unit_tests/test_testing_utils.py @@ -1,5 +1,7 @@ +import sys from unittest.mock import MagicMock +from bec_widgets.tests.testable_qtimer import TestableQTimer from bec_widgets.widgets.services.bec_status_box.bec_status_box import BECServiceStatusMixin @@ -8,3 +10,16 @@ def test_qtimer_uses_testable_qtimer(): assert service_status._service_update_timer.__class__.__name__ != "QTimer" assert service_status._service_update_timer.__class__.__name__ == "TestableQTimer" service_status.cleanup() + + +def test_no_module_binds_the_unpatched_qtimer(): + """bec_widgets.tests.fixtures must patch QTimer before anything imports it by name.""" + unpatched_qtimer = TestableQTimer.__bases__[0] + bound = [ + name + for name, module in list(sys.modules.items()) + if name.startswith(("bec_widgets", "bec_qthemes")) + and name != "bec_widgets.tests.testable_qtimer" + and getattr(module, "QTimer", None) is unpatched_qtimer + ] + assert bound == []