1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-04-09 18:20:55 +02:00

Compare commits

...

2 Commits

52 changed files with 410 additions and 424 deletions

View File

@@ -9,7 +9,7 @@ from bec_lib.endpoints import MessageEndpoints
from bec_lib.redis_connector import RedisConnector
from bec_lib.scan_history import ScanHistory
from bec_widgets.tests.utils import DEVICES, DMMock, FakePositioner, Positioner
from bec_widgets.tests.fake_devices import DEVICES, DMMock, FakePositioner, Positioner
def fake_redis_server(host, port):

View File

@@ -0,0 +1,285 @@
from unittest.mock import MagicMock
from bec_lib.device import Device as BECDevice
from bec_lib.device import Positioner as BECPositioner
from bec_lib.device import ReadoutPriority
from bec_lib.devicemanager import DeviceContainer
class FakeDevice(BECDevice):
"""Fake minimal positioner class for testing."""
def __init__(self, name, enabled=True, readout_priority=ReadoutPriority.MONITORED):
super().__init__(name=name)
self._enabled = enabled
self.signals = {self.name: {"value": 1.0}}
self.description = {self.name: {"source": self.name, "dtype": "number", "shape": []}}
self._readout_priority = readout_priority
self._config = {
"readoutPriority": "baseline",
"deviceClass": "ophyd.Device",
"deviceConfig": {},
"deviceTags": {"user device"},
"enabled": enabled,
"readOnly": False,
"name": self.name,
}
@property
def readout_priority(self):
return self._readout_priority
@readout_priority.setter
def readout_priority(self, value):
self._readout_priority = value
@property
def limits(self) -> tuple[float, float]:
return self._limits
@limits.setter
def limits(self, value: tuple[float, float]):
self._limits = value
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
class FakePositioner(BECPositioner):
def __init__(
self,
name,
enabled=True,
limits=None,
read_value=1.0,
readout_priority=ReadoutPriority.MONITORED,
):
super().__init__(name=name)
# self.limits = limits if limits is not None else [0.0, 0.0]
self.read_value = read_value
self.setpoint_value = read_value
self.motor_is_moving_value = 0
self._enabled = enabled
self._limits = limits
self._readout_priority = readout_priority
self.signals = {self.name: {"value": 1.0}}
self.description = {self.name: {"source": self.name, "dtype": "number", "shape": []}}
self._config = {
"readoutPriority": "baseline",
"deviceClass": "ophyd_devices.SimPositioner",
"deviceConfig": {"delay": 1, "tolerance": 0.01, "update_frequency": 400},
"deviceTags": {"user motors"},
"enabled": enabled,
"readOnly": False,
"name": self.name,
}
self._info = {
"signals": {
"readback": {
"kind_str": "hinted",
"component_name": "readback",
"obj_name": self.name,
}, # hinted
"setpoint": {
"kind_str": "normal",
"component_name": "setpoint",
"obj_name": f"{self.name}_setpoint",
}, # normal
"velocity": {
"kind_str": "config",
"component_name": "velocity",
"obj_name": f"{self.name}_velocity",
}, # config
}
}
self.signals = {
self.name: {"value": self.read_value},
f"{self.name}_setpoint": {"value": self.setpoint_value},
f"{self.name}_motor_is_moving": {"value": self.motor_is_moving_value},
}
@property
def readout_priority(self):
return self._readout_priority
@readout_priority.setter
def readout_priority(self, value):
self._readout_priority = value
@property
def enabled(self) -> bool:
return self._enabled
@enabled.setter
def enabled(self, value: bool):
self._enabled = value
@property
def limits(self) -> tuple[float, float]:
return self._limits
@limits.setter
def limits(self, value: tuple[float, float]):
self._limits = value
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.read_value = fake_value
def describe(self) -> dict:
"""
Get the description of the device
Returns:
dict: Description of the device
"""
return self.description
@property
def precision(self):
return 3
def set_read_value(self, value):
self.read_value = value
def read(self, cached=False):
return self.signals
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 Positioner(FakePositioner):
"""just placeholder for testing embedded isinstance check in DeviceCombobox"""
def __init__(self, name="test", limits=None, read_value=1.0, enabled=True):
super().__init__(name, limits=limits, read_value=read_value, enabled=enabled)
class Device(FakeDevice):
"""just placeholder for testing embedded isinstance check in DeviceCombobox"""
def __init__(self, name, enabled=True):
super().__init__(name, enabled)
class DMMock:
def __init__(self):
self.devices = DeviceContainer()
self.enabled_devices = [device for device in self.devices if device.enabled]
def add_devices(self, devices: list):
"""
Add devices to the DeviceContainer.
Args:
devices (list): List of device instances to add.
"""
for device in devices:
self.devices[device.name] = device
def get_bec_signals(self, signal_class_name: str):
"""
Emulate DeviceManager.get_bec_signals for unit-tests.
For “AsyncSignal” we list every device whose readout_priority is
ReadoutPriority.ASYNC and build a minimal tuple
(device_name, signal_name, signal_info_dict) that matches the real
API shape used by Waveform._check_async_signal_found.
"""
signals: list[tuple[str, str, dict]] = []
if signal_class_name != "AsyncSignal":
return signals
for device in self.devices.values():
if getattr(device, "readout_priority", None) == ReadoutPriority.ASYNC:
device_name = device.name
signal_name = device.name # primary signal in our mocks
signal_info = {
"component_name": signal_name,
"obj_name": signal_name,
"kind_str": "hinted",
"signal_class": signal_class_name,
"metadata": {
"connected": True,
"precision": None,
"read_access": True,
"timestamp": 0.0,
"write_access": True,
},
}
signals.append((device_name, signal_name, signal_info))
return signals
DEVICES = [
FakePositioner("samx", limits=[-10, 10], read_value=2.0),
FakePositioner("samy", limits=[-5, 5], read_value=3.0),
FakePositioner("samz", limits=[-8, 8], read_value=4.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"),
FakeDevice("eiger", readout_priority=ReadoutPriority.ASYNC),
FakeDevice("waveform1d"),
FakeDevice("async_device", readout_priority=ReadoutPriority.ASYNC),
Positioner("test", limits=[-10, 10], read_value=2.0),
Device("test_device"),
]
def check_remote_data_size(widget, plot_name, num_elements):
"""
Check if the remote data has the correct number of elements.
Used in the qtbot.waitUntil function.
"""
return len(widget.get_all_data()[plot_name]["x"]) == num_elements

View File

@@ -1,285 +1,76 @@
from unittest.mock import MagicMock
from pytestqt.exceptions import TimeoutError as QtBotTimeoutError
from qtpy.QtCore import QTimer
from qtpy.QtWidgets import QApplication
from bec_lib.device import Device as BECDevice
from bec_lib.device import Positioner as BECPositioner
from bec_lib.device import ReadoutPriority
from bec_lib.devicemanager import DeviceContainer
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils import bec_dispatcher as bec_dispatcher_module
from bec_widgets.utils import error_popups
class FakeDevice(BECDevice):
"""Fake minimal positioner class for testing."""
class TestableQTimer(QTimer):
_instances: list[tuple[QTimer, str]] = []
_current_test_name: str = ""
def __init__(self, name, enabled=True, readout_priority=ReadoutPriority.MONITORED):
super().__init__(name=name)
self._enabled = enabled
self.signals = {self.name: {"value": 1.0}}
self.description = {self.name: {"source": self.name, "dtype": "number", "shape": []}}
self._readout_priority = readout_priority
self._config = {
"readoutPriority": "baseline",
"deviceClass": "ophyd.Device",
"deviceConfig": {},
"deviceTags": {"user device"},
"enabled": enabled,
"readOnly": False,
"name": self.name,
}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
TestableQTimer._instances.append((self, TestableQTimer._current_test_name))
@property
def readout_priority(self):
return self._readout_priority
@classmethod
def check_all_stopped(cls, qtbot):
def _is_done_or_deleted(t: QTimer):
try:
return not t.isActive()
except RuntimeError as e:
return "already deleted" in e.args[0]
@readout_priority.setter
def readout_priority(self, value):
self._readout_priority = value
@property
def limits(self) -> tuple[float, float]:
return self._limits
@limits.setter
def limits(self, value: tuple[float, float]):
self._limits = value
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
try:
qtbot.waitUntil(lambda: all(_is_done_or_deleted(timer) for timer, _ in cls._instances))
except QtBotTimeoutError as exc:
active_timers = list(filter(lambda t: t[0].isActive(), cls._instances))
(t.stop() for t, _ in cls._instances)
raise TimeoutError(f"Failed to stop all timers: {active_timers}") from exc
cls._instances = []
class FakePositioner(BECPositioner):
def qapplication_fixture(qtbot, request, testable_qtimer_class):
yield
def __init__(
self,
name,
enabled=True,
limits=None,
read_value=1.0,
readout_priority=ReadoutPriority.MONITORED,
):
super().__init__(name=name)
# self.limits = limits if limits is not None else [0.0, 0.0]
self.read_value = read_value
self.setpoint_value = read_value
self.motor_is_moving_value = 0
self._enabled = enabled
self._limits = limits
self._readout_priority = readout_priority
self.signals = {self.name: {"value": 1.0}}
self.description = {self.name: {"source": self.name, "dtype": "number", "shape": []}}
self._config = {
"readoutPriority": "baseline",
"deviceClass": "ophyd_devices.SimPositioner",
"deviceConfig": {"delay": 1, "tolerance": 0.01, "update_frequency": 400},
"deviceTags": {"user motors"},
"enabled": enabled,
"readOnly": False,
"name": self.name,
}
self._info = {
"signals": {
"readback": {
"kind_str": "hinted",
"component_name": "readback",
"obj_name": self.name,
}, # hinted
"setpoint": {
"kind_str": "normal",
"component_name": "setpoint",
"obj_name": f"{self.name}_setpoint",
}, # normal
"velocity": {
"kind_str": "config",
"component_name": "velocity",
"obj_name": f"{self.name}_velocity",
}, # config
}
}
self.signals = {
self.name: {"value": self.read_value},
f"{self.name}_setpoint": {"value": self.setpoint_value},
f"{self.name}_motor_is_moving": {"value": self.motor_is_moving_value},
}
if request.node.stash._storage.get("failed"):
print("Test failed, skipping cleanup checks")
return
@property
def readout_priority(self):
return self._readout_priority
bec_dispatcher = bec_dispatcher_module.BECDispatcher()
bec_dispatcher.stop_cli_server()
@readout_priority.setter
def readout_priority(self, value):
self._readout_priority = value
@property
def enabled(self) -> bool:
return self._enabled
@enabled.setter
def enabled(self, value: bool):
self._enabled = value
@property
def limits(self) -> tuple[float, float]:
return self._limits
@limits.setter
def limits(self, value: tuple[float, float]):
self._limits = value
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.read_value = fake_value
def describe(self) -> dict:
"""
Get the description of the device
Returns:
dict: Description of the device
"""
return self.description
@property
def precision(self):
return 3
def set_read_value(self, value):
self.read_value = value
def read(self, cached=False):
return self.signals
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))
testable_qtimer_class.check_all_stopped(qtbot)
qapp = QApplication.instance()
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
class Positioner(FakePositioner):
"""just placeholder for testing embedded isinstance check in DeviceCombobox"""
def __init__(self, name="test", limits=None, read_value=1.0, enabled=True):
super().__init__(name, limits=limits, read_value=read_value, enabled=enabled)
def rpc_register_fixture():
try:
yield RPCRegister()
finally:
RPCRegister.reset_singleton()
class Device(FakeDevice):
"""just placeholder for testing embedded isinstance check in DeviceCombobox"""
def __init__(self, name, enabled=True):
super().__init__(name, enabled)
def bec_dispatcher_fixture(threads_check):
bec_dispatcher = bec_dispatcher_module.BECDispatcher()
try:
yield bec_dispatcher
finally:
bec_dispatcher.disconnect_all()
bec_dispatcher.client.shutdown()
bec_dispatcher.stop_cli_server()
bec_dispatcher_module.BECDispatcher.reset_singleton()
class DMMock:
def __init__(self):
self.devices = DeviceContainer()
self.enabled_devices = [device for device in self.devices if device.enabled]
def add_devices(self, devices: list):
"""
Add devices to the DeviceContainer.
Args:
devices (list): List of device instances to add.
"""
for device in devices:
self.devices[device.name] = device
def get_bec_signals(self, signal_class_name: str):
"""
Emulate DeviceManager.get_bec_signals for unit-tests.
For “AsyncSignal” we list every device whose readout_priority is
ReadoutPriority.ASYNC and build a minimal tuple
(device_name, signal_name, signal_info_dict) that matches the real
API shape used by Waveform._check_async_signal_found.
"""
signals: list[tuple[str, str, dict]] = []
if signal_class_name != "AsyncSignal":
return signals
for device in self.devices.values():
if getattr(device, "readout_priority", None) == ReadoutPriority.ASYNC:
device_name = device.name
signal_name = device.name # primary signal in our mocks
signal_info = {
"component_name": signal_name,
"obj_name": signal_name,
"kind_str": "hinted",
"signal_class": signal_class_name,
"metadata": {
"connected": True,
"precision": None,
"read_access": True,
"timestamp": 0.0,
"write_access": True,
},
}
signals.append((device_name, signal_name, signal_info))
return signals
DEVICES = [
FakePositioner("samx", limits=[-10, 10], read_value=2.0),
FakePositioner("samy", limits=[-5, 5], read_value=3.0),
FakePositioner("samz", limits=[-8, 8], read_value=4.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"),
FakeDevice("eiger", readout_priority=ReadoutPriority.ASYNC),
FakeDevice("waveform1d"),
FakeDevice("async_device", readout_priority=ReadoutPriority.ASYNC),
Positioner("test", limits=[-10, 10], read_value=2.0),
Device("test_device"),
]
def check_remote_data_size(widget, plot_name, num_elements):
"""
Check if the remote data has the correct number of elements.
Used in the qtbot.waitUntil function.
"""
return len(widget.get_all_data()[plot_name]["x"]) == num_elements
def clean_singleton_fixture():
error_popups._popup_utility_instance = None
yield

View File

@@ -1,33 +1,7 @@
import pytest
import qtpy.QtCore
from pytestqt.exceptions import TimeoutError as QtBotTimeoutError
from qtpy.QtCore import QTimer
class TestableQTimer(QTimer):
_instances: list[tuple[QTimer, str]] = []
_current_test_name: str = ""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
TestableQTimer._instances.append((self, TestableQTimer._current_test_name))
@classmethod
def check_all_stopped(cls, qtbot):
def _is_done_or_deleted(t: QTimer):
try:
return not t.isActive()
except RuntimeError as e:
return "already deleted" in e.args[0]
try:
qtbot.waitUntil(lambda: all(_is_done_or_deleted(timer) for timer, _ in cls._instances))
except QtBotTimeoutError as exc:
active_timers = list(filter(lambda t: t[0].isActive(), cls._instances))
(t.stop() for t, _ in cls._instances)
raise TimeoutError(f"Failed to stop all timers: {active_timers}") from exc
cls._instances = []
from bec_widgets.tests.utils 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

View File

@@ -6,7 +6,7 @@ from bec_lib.endpoints import MessageEndpoints
from bec_widgets.cli.client import Image, MotorMap, MultiWaveform, ScatterWaveform, Waveform
from bec_widgets.cli.rpc.rpc_base import RPCReference
from bec_widgets.tests.utils import check_remote_data_size
from bec_widgets.tests.fake_devices import check_remote_data_size
def test_rpc_waveform1d_custom_curve(qtbot, connected_client_gui_obj):

View File

@@ -6,12 +6,9 @@ import numpy as np
import pytest
from bec_lib import messages
from bec_lib.messages import _StoredDataInfo
from pytestqt.exceptions import TimeoutError as QtBotTimeoutError
from qtpy.QtWidgets import QApplication, QMessageBox
from qtpy.QtWidgets import QMessageBox
from bec_widgets.cli.rpc.rpc_register import RPCRegister
from bec_widgets.utils import bec_dispatcher as bec_dispatcher_module
from bec_widgets.utils import error_popups
from bec_widgets.tests import utils as test_utils
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
@@ -25,49 +22,22 @@ def pytest_runtest_makereport(item, call):
@pytest.fixture(autouse=True)
def qapplication(qtbot, request, testable_qtimer_class): # pylint: disable=unused-argument
yield
# 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 = QApplication.instance()
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
yield from test_utils.qapplication_fixture(qtbot, request, testable_qtimer_class)
@pytest.fixture(autouse=True)
def rpc_register():
yield RPCRegister()
RPCRegister.reset_singleton()
yield from test_utils.rpc_register_fixture()
@pytest.fixture(autouse=True)
def bec_dispatcher(threads_check): # pylint: disable=unused-argument
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()
yield from test_utils.bec_dispatcher_fixture(threads_check)
@pytest.fixture(autouse=True)
def clean_singleton():
error_popups._popup_utility_instance = None
yield from test_utils.clean_singleton_fixture()
@pytest.fixture(autouse=True)

View File

@@ -2,10 +2,9 @@
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.control.buttons.button_abort.button_abort import AbortButton
from .client_mocks import mocked_client
@pytest.fixture
def abort_button(qtbot, mocked_client):

View File

@@ -1,9 +1,9 @@
import pytest
from qtpy.QtWidgets import QDoubleSpinBox, QLineEdit
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.plots.plot_base import PlotBase
from bec_widgets.widgets.plots.setting_menus.axis_settings import AxisSettings
from tests.unit_tests.client_mocks import mocked_client
from tests.unit_tests.conftest import create_widget

View File

@@ -5,11 +5,10 @@ import pytest
from qtpy.QtCore import QObject
from qtpy.QtWidgets import QApplication
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.utils import BECConnector
from bec_widgets.utils.error_popups import SafeSlot as Slot
from .client_mocks import mocked_client
class BECConnectorQObject(BECConnector, QObject): ...

View File

@@ -5,9 +5,9 @@ from unittest import mock
import pytest
from bec_lib.endpoints import MessageEndpoints
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.containers.dock import BECDockArea
from .client_mocks import mocked_client
from .test_bec_queue import bec_queue_msg_full

View File

@@ -1,10 +1,9 @@
import pytest
from bec_lib import messages
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.services.bec_queue.bec_queue import BECQueue
from .client_mocks import mocked_client
@pytest.fixture
def bec_queue_msg_full():

View File

@@ -4,13 +4,12 @@ from unittest import mock
import pytest
from bec_lib.messages import BECStatus, ServiceMetricMessage, StatusMessage
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.services.bec_status_box.bec_status_box import (
BECServiceInfoContainer,
BECStatusBox,
)
from .client_mocks import mocked_client
@pytest.fixture
def service_status_fixture():

View File

@@ -4,11 +4,11 @@ from pydantic import ValidationError
from qtpy.QtGui import QColor
from qtpy.QtWidgets import QVBoxLayout, QWidget
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.utils import Colors, ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.widgets.plots.waveform.curve import CurveConfig
from tests.unit_tests.client_mocks import mocked_client
from tests.unit_tests.conftest import create_widget

View File

@@ -4,10 +4,10 @@ import pytest
from qtpy.QtCore import QPointF, Qt
from qtpy.QtGui import QTransform
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.utils import Crosshair
from bec_widgets.widgets.plots.image.image_item import ImageItem
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from tests.unit_tests.client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -6,13 +6,13 @@ from bec_lib.scan_history import ScanHistory
from qtpy.QtGui import QValidator
from qtpy.QtWidgets import QComboBox, QVBoxLayout
from bec_widgets.tests.client_mocks import dap_plugin_message, mocked_client, mocked_client_with_dap
from bec_widgets.widgets.plots.waveform.settings.curve_settings.curve_setting import CurveSetting
from bec_widgets.widgets.plots.waveform.settings.curve_settings.curve_tree import (
CurveTree,
ScanIndexValidator,
)
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from tests.unit_tests.client_mocks import dap_plugin_message, mocked_client, mocked_client_with_dap
from tests.unit_tests.conftest import create_widget
##################################################

View File

@@ -1,8 +1,8 @@
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.dap.dap_combo_box.dap_combo_box import DapComboBox
from .client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -4,12 +4,11 @@ import pytest
from qtpy.QtCore import Qt
from qtpy.QtWidgets import QApplication
# pylint: disable=unused-import
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.utils.colors import set_theme
from bec_widgets.widgets.utility.visual.dark_mode_button.dark_mode_button import DarkModeButton
# pylint: disable=unused-import
from .client_mocks import mocked_client
# pylint: disable=redefined-outer-name

View File

@@ -6,6 +6,7 @@ from bec_lib.device import Device
from qtpy.QtCore import QPoint, Qt
from qtpy.QtWidgets import QTabWidget
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.services.device_browser.device_browser import DeviceBrowser
from bec_widgets.widgets.services.device_browser.device_item.device_config_form import (
DeviceConfigForm,
@@ -14,8 +15,6 @@ from bec_widgets.widgets.services.device_browser.device_item.device_signal_displ
SignalDisplay,
)
from .client_mocks import mocked_client
if TYPE_CHECKING: # pragma: no cover
from qtpy.QtWidgets import QListWidgetItem

View File

@@ -4,13 +4,13 @@ import pytest
from bec_lib.device import ReadoutPriority
from qtpy.QtWidgets import QWidget
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import (
BECDeviceFilter,
DeviceInputBase,
DeviceInputConfig,
)
from .client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -1,14 +1,13 @@
import pytest
from bec_lib.device import ReadoutPriority
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter
from bec_widgets.widgets.control.device_input.device_combobox.device_combobox import DeviceComboBox
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import (
DeviceLineEdit,
)
from .client_mocks import mocked_client
@pytest.fixture
def device_input_combobox(qtbot, mocked_client):

View File

@@ -4,7 +4,7 @@ import pytest
from bec_lib.device import Signal
from qtpy.QtWidgets import QWidget
from bec_widgets.tests.utils import FakeDevice
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.utils.ophyd_kind_util import Kind
from bec_widgets.widgets.control.device_input.base_classes.device_input_base import BECDeviceFilter
from bec_widgets.widgets.control.device_input.base_classes.device_signal_input_base import (
@@ -16,7 +16,6 @@ from bec_widgets.widgets.control.device_input.signal_line_edit.signal_line_edit
SignalLineEdit,
)
from .client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -1,12 +1,12 @@
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.utils.filter_io import FilterIO
from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit import (
DeviceLineEdit,
)
from bec_widgets.widgets.dap.dap_combo_box.dap_combo_box import DapComboBox
from .client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -5,12 +5,9 @@ import pytest
from bec_lib import messages
from bec_lib.scan_history import ScanHistory
from bec_widgets.widgets.plots.heatmap.heatmap import Heatmap, HeatmapConfig, HeatmapDeviceSignal
# pytest: disable=unused-import
from tests.unit_tests.client_mocks import mocked_client
from .client_mocks import create_dummy_scan_item
from bec_widgets.tests.client_mocks import create_dummy_scan_item, mocked_client
from bec_widgets.widgets.plots.heatmap.heatmap import Heatmap, HeatmapConfig, HeatmapDeviceSignal
@pytest.fixture

View File

@@ -4,10 +4,10 @@ import numpy as np
import pytest
from qtpy.QtCore import QPointF, Qt
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.plots.image.image import Image
from bec_widgets.widgets.plots.image.setting_widgets.image_roi_tree import ROIPropertyTree
from bec_widgets.widgets.plots.roi.image_roi import CircularROI, RectangularROI
from tests.unit_tests.client_mocks import mocked_client
from tests.unit_tests.conftest import create_widget

View File

@@ -5,6 +5,7 @@ from typing import Literal
import numpy as np
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.plots.image.image import Image
from bec_widgets.widgets.plots.roi.image_roi import (
CircularROI,
@@ -12,7 +13,6 @@ from bec_widgets.widgets.plots.roi.image_roi import (
RectangularROI,
ROIController,
)
from tests.unit_tests.client_mocks import mocked_client
from tests.unit_tests.conftest import create_widget

View File

@@ -3,8 +3,8 @@ import pyqtgraph as pg
import pytest
from qtpy.QtCore import QPointF
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.plots.image.image import Image
from tests.unit_tests.client_mocks import mocked_client
from tests.unit_tests.conftest import create_widget
##################################################

View File

@@ -8,11 +8,10 @@ from qtpy.QtGui import QFontMetrics
import bec_widgets
from bec_widgets.applications.launch_window import LaunchWindow
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.containers.auto_update.auto_updates import AutoUpdates
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow
from .client_mocks import mocked_client
base_path = os.path.dirname(bec_widgets.__file__)

View File

@@ -3,9 +3,9 @@ from unittest import mock
import numpy as np
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.dap.lmfit_dialog.lmfit_dialog import LMFitDialog
from .client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -8,18 +8,12 @@ from unittest.mock import MagicMock, patch
import pytest
from bec_lib.messages import LogMessage
from bec_lib.redis_connector import StreamMessage
from qtpy.QtCore import QDateTime
from bec_widgets.widgets.utility.logpanel._util import (
log_time,
replace_escapes,
simple_color_format,
)
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.utility.logpanel._util import replace_escapes, simple_color_format
from bec_widgets.widgets.utility.logpanel.logpanel import DEFAULT_LOG_COLORS, LogPanel
from .client_mocks import mocked_client
TEST_TABLE_STRING = "2025-01-15 15:57:18 | bec_server.scan_server.scan_queue | [INFO] | \n \x1b[3m primary queue / ScanQueueStatus.RUNNING \x1b[0m\n┏━━━━━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━┳━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━┓\n\x1b[1m \x1b[0m\x1b[1m queue_id \x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mscan_id\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mis_scan\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mtype\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mscan_numb…\x1b[0m\x1b[1m \x1b[0m┃\x1b[1m \x1b[0m\x1b[1mIQ status\x1b[0m\x1b[1m \x1b[0m┃\n┡━━━━━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━╇━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━┩\n│ bbe50c82-6… │ None │ False │ mv │ None │ PENDING │\n└─────────────┴─────────┴─────────┴──────┴────────────┴───────────┘\n\n"
TEST_LOG_MESSAGES = [

View File

@@ -5,6 +5,7 @@ from qtpy.QtCore import QEvent, QPoint, QPointF
from qtpy.QtGui import QEnterEvent
from qtpy.QtWidgets import QApplication, QFrame, QLabel
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.containers.main_window.addons.hover_widget import (
HoverWidget,
WidgetTooltip,
@@ -13,7 +14,6 @@ from bec_widgets.widgets.containers.main_window.addons.scroll_label import Scrol
from bec_widgets.widgets.containers.main_window.addons.web_links import BECWebLinksMixin
from bec_widgets.widgets.containers.main_window.main_window import BECMainWindow
from .client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -1,8 +1,8 @@
import numpy as np
import pyqtgraph as pg
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.plots.motor_map.motor_map import MotorMap
from tests.unit_tests.client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -1,7 +1,7 @@
import numpy as np
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.plots.multi_waveform.multi_waveform import MultiWaveform
from tests.unit_tests.client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -1,6 +1,7 @@
import pytest
from qtpy import QtCore, QtGui, QtWidgets
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.utils.error_popups import ErrorPopupUtility
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_banner import (
DARK_PALETTE,
@@ -13,8 +14,6 @@ from bec_widgets.widgets.containers.main_window.addons.notification_center.notif
SeverityKind,
)
from .client_mocks import mocked_client
@pytest.fixture
def toast(qtbot):

View File

@@ -1,8 +1,8 @@
import numpy as np
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.plots.plot_base import PlotBase, UIMode
from .client_mocks import mocked_client
from .conftest import create_widget
# pylint: disable=unused-import

View File

@@ -7,7 +7,8 @@ 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.tests.client_mocks import mocked_client
from bec_widgets.tests.fake_devices import Positioner
from bec_widgets.widgets.control.device_control.positioner_box import (
PositionerBox,
PositionerControlLine,
@@ -16,7 +17,6 @@ from bec_widgets.widgets.control.device_input.device_line_edit.device_line_edit
DeviceLineEdit,
)
from .client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -2,9 +2,9 @@ from unittest import mock
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.control.device_control.positioner_box import PositionerBox2D
from .client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -5,10 +5,9 @@ from unittest.mock import patch
import pytest
from qtpy.QtWidgets import QMessageBox
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.control.buttons.button_reset.button_reset import ResetButton
from .client_mocks import mocked_client
@pytest.fixture
def reset_button(qtbot, mocked_client):

View File

@@ -2,10 +2,9 @@
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.control.buttons.button_resume.button_resume import ResumeButton
from .client_mocks import mocked_client
@pytest.fixture
def resume_button(qtbot, mocked_client):

View File

@@ -4,13 +4,12 @@ import pytest
from bec_lib.endpoints import MessageEndpoints
from pydantic import ValidationError
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.utils import Colors
from bec_widgets.widgets.progress.ring_progress_bar import RingProgressBar
from bec_widgets.widgets.progress.ring_progress_bar.ring import ProgressbarConnections, RingConfig
from bec_widgets.widgets.progress.ring_progress_bar.ring_progress_bar import RingProgressBarConfig
from .client_mocks import mocked_client
@pytest.fixture
def ring_progress_bar(qtbot, mocked_client):

View File

@@ -7,12 +7,11 @@ from bec_lib.endpoints import MessageEndpoints
from bec_lib.messages import AvailableResourceMessage, ScanHistoryMessage
from qtpy.QtCore import QModelIndex, Qt
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.utils.forms_from_types.items import StrFormItem
from bec_widgets.utils.widget_io import WidgetIO
from bec_widgets.widgets.control.scan_control import ScanControl
from .client_mocks import mocked_client
# pylint: disable=no-member
# pylint: disable=missing-function-docstring
# pylint: disable=redefined-outer-name

View File

@@ -2,10 +2,9 @@ from unittest import mock
import pytest
from bec_lib.messages import ScanHistoryMessage, _StoredDataInfo
from pytestqt import qtbot
from qtpy import QtCore
from bec_widgets.utils.colors import get_accent_colors
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.services.scan_history_browser.components import (
ScanHistoryDeviceViewer,
ScanHistoryMetadataViewer,
@@ -15,8 +14,6 @@ from bec_widgets.widgets.services.scan_history_browser.scan_history_browser impo
ScanHistoryBrowser,
)
from .client_mocks import mocked_client
@pytest.fixture
def scan_history_msg():

View File

@@ -4,6 +4,7 @@ import numpy as np
import pytest
from bec_lib import messages
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.progress.bec_progressbar.bec_progressbar import (
BECProgressBar,
ProgressState,
@@ -14,8 +15,6 @@ from bec_widgets.widgets.progress.scan_progressbar.scan_progressbar import (
ScanProgressBar,
)
from .client_mocks import mocked_client
@pytest.fixture
def scan_progressbar(qtbot, mocked_client):

View File

@@ -2,12 +2,12 @@ import json
import numpy as np
from bec_widgets.tests.client_mocks import create_dummy_scan_item, mocked_client
from bec_widgets.widgets.plots.scatter_waveform.scatter_curve import (
ScatterCurveConfig,
ScatterDeviceSignal,
)
from bec_widgets.widgets.plots.scatter_waveform.scatter_waveform import ScatterWaveform
from tests.unit_tests.client_mocks import create_dummy_scan_item, mocked_client
from .conftest import create_widget

View File

@@ -4,15 +4,14 @@ from unittest.mock import MagicMock, patch
import pytest
from qtpy import QtCore
from qtpy.QtWidgets import QDialogButtonBox, QLabel
from qtpy.QtWidgets import QDialogButtonBox
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.control.device_input.base_classes.device_signal_input_base import (
DeviceSignalInputBaseConfig,
)
from bec_widgets.widgets.utility.signal_label.signal_label import ChoiceDialog, SignalLabel
from .client_mocks import mocked_client
SAMX_INFO_DICT = {
"signals": {
"readback": {

View File

@@ -2,10 +2,9 @@
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.control.buttons.stop_button.stop_button import StopButton
from .client_mocks import mocked_client
@pytest.fixture
def stop_button(qtbot, mocked_client):

View File

@@ -1,9 +1,8 @@
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.editors.text_box.text_box import DEFAULT_TEXT, TextBox
from .client_mocks import mocked_client
@pytest.fixture
def text_box_widget(qtbot, mocked_client):

View File

@@ -3,10 +3,10 @@ from unittest import mock
import pyqtgraph as pg
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.utils.bec_signal_proxy import BECSignalProxy
from bec_widgets.widgets.dap.dap_combo_box.dap_combo_box import DapComboBox
from .client_mocks import mocked_client
from .conftest import create_widget

View File

@@ -1,10 +1,9 @@
import pytest
from qtpy.QtCore import QPointF
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from .client_mocks import mocked_client
@pytest.fixture
def plot_widget_with_arrow_item(qtbot, mocked_client):

View File

@@ -5,10 +5,9 @@ from unittest import mock
import pytest
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.editors.vscode.vscode import VSCodeEditor
from .client_mocks import mocked_client
@pytest.fixture
def vscode_widget(qtbot, mocked_client):

View File

@@ -12,13 +12,7 @@ from pyqtgraph.graphicsItems.DateAxisItem import DateAxisItem
from qtpy.QtCore import QTimer
from qtpy.QtWidgets import QApplication, QCheckBox, QDialog, QDialogButtonBox, QDoubleSpinBox
from bec_widgets.widgets.plots.plot_base import UIMode
from bec_widgets.widgets.plots.waveform.curve import DeviceSignal
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from bec_widgets.widgets.services.scan_history_browser.scan_history_browser import (
ScanHistoryBrowser,
)
from tests.unit_tests.client_mocks import (
from bec_widgets.tests.client_mocks import (
DummyData,
create_dummy_scan_item,
dap_plugin_message,
@@ -26,6 +20,12 @@ from tests.unit_tests.client_mocks import (
mocked_client,
mocked_client_with_dap,
)
from bec_widgets.widgets.plots.plot_base import UIMode
from bec_widgets.widgets.plots.waveform.curve import DeviceSignal
from bec_widgets.widgets.plots.waveform.waveform import Waveform
from bec_widgets.widgets.services.scan_history_browser.scan_history_browser import (
ScanHistoryBrowser,
)
from .conftest import create_widget

View File

@@ -3,10 +3,9 @@ from unittest import mock
import pytest
from qtpy.QtNetwork import QAuthenticator
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.editors.web_console.web_console import WebConsole, _web_console_registry
from .client_mocks import mocked_client
@pytest.fixture
def console_widget(qtbot, mocked_client):

View File

@@ -1,10 +1,9 @@
import pytest
from qtpy.QtCore import QUrl
from bec_widgets.tests.client_mocks import mocked_client
from bec_widgets.widgets.editors.website.website import WebsiteWidget
from .client_mocks import mocked_client
@pytest.fixture
def website_widget(qtbot, mocked_client):