1
0
mirror of https://github.com/bec-project/bec_widgets.git synced 2026-04-18 14:25:37 +02:00

Compare commits

...

12 Commits

10 changed files with 1880 additions and 50 deletions

View File

@@ -1,6 +1,53 @@
# CHANGELOG
## v2.30.0 (2025-07-22)
### Bug Fixes
- **device_browser**: Display signal for signals
([`3384ca0`](https://github.com/bec-project/bec_widgets/commit/3384ca02bdb5a2798ad3339ecf3e2ba7c121e28f))
- **device_signal_display**: Don't read omitted
([`b9af36a`](https://github.com/bec-project/bec_widgets/commit/b9af36a4f1c91e910d4fc738b17b90e92287a7e3))
- **signal_label**: Rewrite reading selection logic
([`cd17a4a`](https://github.com/bec-project/bec_widgets/commit/cd17a4aad905296eb0460ecc27e5920f5c2e8fe5))
- **signal_label**: Show all signals by default
([`22beadc`](https://github.com/bec-project/bec_widgets/commit/22beadcad061b328c986414f30fef57b64bad693))
- **signal_label**: Update signal from dialog correctly
([`959cedb`](https://github.com/bec-project/bec_widgets/commit/959cedbbd5a123eef5f3370287bf6476c48caab9))
- **signal_label**: Use read() instead of get() for init
([`f0dc992`](https://github.com/bec-project/bec_widgets/commit/f0dc99258607a5cc8af51686d01f7fd54ae2779f))
### Chores
- Update client.py
([`fd1f994`](https://github.com/bec-project/bec_widgets/commit/fd1f9941e046b7ae1e247dde39c20bcbc37ac189))
### Features
- **signal_label**: Property to display array data or not
([`ca4f975`](https://github.com/bec-project/bec_widgets/commit/ca4f97503bf06363e8e8a5d494a9857223da4104))
## v2.29.0 (2025-07-22)
### Features
- **notification_banner**: Notification centre for alarms implemented into BECMainWindow
([`cd9d22d`](https://github.com/bec-project/bec_widgets/commit/cd9d22d0b40d633af76cb1188b57feb7b6a5dbf2))
### Refactoring
- **notification_banner**: Becnotificationbroker done as singleton to sync all windows in the
session
([`7cda2ed`](https://github.com/bec-project/bec_widgets/commit/7cda2ed846d3c27799f4f15f6c5c667631b1ca55))
## v2.28.0 (2025-07-21)
### Features

View File

@@ -4378,6 +4378,62 @@ class SignalLabel(RPCBase):
Show the button to select the signal to display
"""
@property
@rpc_call
def show_hinted_signals(self) -> "bool":
"""
In the signal selection menu, show hinted signals
"""
@show_hinted_signals.setter
@rpc_call
def show_hinted_signals(self) -> "bool":
"""
In the signal selection menu, show hinted signals
"""
@property
@rpc_call
def show_normal_signals(self) -> "bool":
"""
In the signal selection menu, show normal signals
"""
@show_normal_signals.setter
@rpc_call
def show_normal_signals(self) -> "bool":
"""
In the signal selection menu, show normal signals
"""
@property
@rpc_call
def show_config_signals(self) -> "bool":
"""
In the signal selection menu, show config signals
"""
@show_config_signals.setter
@rpc_call
def show_config_signals(self) -> "bool":
"""
In the signal selection menu, show config signals
"""
@property
@rpc_call
def display_array_data(self) -> "bool":
"""
Displays the full data from array signals if set to True.
"""
@display_array_data.setter
@rpc_call
def display_array_data(self) -> "bool":
"""
Displays the full data from array signals if set to True.
"""
class SignalLineEdit(RPCBase):
"""Line edit widget for device input with autocomplete for device names."""

View File

@@ -19,10 +19,15 @@ from qtpy.QtWidgets import (
import bec_widgets
from bec_widgets.utils import UILoader
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.colors import apply_theme
from bec_widgets.utils.colors import apply_theme, set_theme
from bec_widgets.utils.error_popups import SafeSlot
from bec_widgets.utils.widget_io import WidgetHierarchy
from bec_widgets.widgets.containers.main_window.addons.hover_widget import HoverWidget
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_banner import (
BECNotificationBroker,
NotificationCentre,
NotificationIndicator,
)
from bec_widgets.widgets.containers.main_window.addons.scroll_label import ScrollLabel
from bec_widgets.widgets.containers.main_window.addons.web_links import BECWebLinksMixin
from bec_widgets.widgets.progress.scan_progressbar.scan_progressbar import ScanProgressBar
@@ -50,6 +55,14 @@ class BECMainWindow(BECWidget, QMainWindow):
self.app = QApplication.instance()
self.status_bar = self.statusBar()
self.setWindowTitle(window_title)
# Notification Centre overlay
self.notification_centre = NotificationCentre(parent=self) # Notification layer
self.notification_broker = BECNotificationBroker()
self._nc_margin = 16
self._position_notification_centre()
# Init ui
self._init_ui()
self._connect_to_theme_change()
@@ -58,6 +71,34 @@ class BECMainWindow(BECWidget, QMainWindow):
self.display_client_message, MessageEndpoints.client_info()
)
def setCentralWidget(self, widget: QWidget, qt_default: bool = False): # type: ignore[override]
"""
Reimplement QMainWindow.setCentralWidget so that the *main content*
widget always lives on the lower layer of the stacked layout that
hosts our notification overlays.
Args:
widget: The widget that should become the new central content.
qt_default: When *True* the call is forwarded to the base class so
that Qt behaves exactly as the original implementation (used
during __init__ when we first install ``self._full_content``).
"""
super().setCentralWidget(widget)
self.notification_centre.raise_()
self.statusBar().raise_()
def resizeEvent(self, event):
super().resizeEvent(event)
self._position_notification_centre()
def _position_notification_centre(self):
"""Keep the notification panel at a fixed margin top-right."""
if not hasattr(self, "notification_centre"):
return
margin = getattr(self, "_nc_margin", 16) # px
nc = self.notification_centre
nc.move(self.width() - nc.width() - margin, margin)
################################################################################
# MainWindow Elements Initialization
################################################################################
@@ -94,6 +135,26 @@ class BECMainWindow(BECWidget, QMainWindow):
# Add scan_progress bar with display logic
self._add_scan_progress_bar()
# Setup NotificationIndicator to bottom right of the status bar
self._add_notification_indicator()
################################################################################
# Notification indicator and Notification Centre helpers
def _add_notification_indicator(self):
"""
Add the notification indicator to the status bar and hook the signals.
"""
# Add the notification indicator to the status bar
self.notification_indicator = NotificationIndicator(self)
self.status_bar.addPermanentWidget(self.notification_indicator)
# Connect the notification broker to the indicator
self.notification_centre.counts_updated.connect(self.notification_indicator.update_counts)
self.notification_indicator.filter_changed.connect(self.notification_centre.apply_filter)
self.notification_indicator.show_all_requested.connect(self.notification_centre.show_all)
self.notification_indicator.hide_all_requested.connect(self.notification_centre.hide_all)
################################################################################
# Client message status bar widget helpers
@@ -379,12 +440,12 @@ class BECMainWindow(BECWidget, QMainWindow):
@SafeSlot(str)
def change_theme(self, theme: str):
"""
Change the theme of the application.
Change the theme of the application and propagate it to widgets.
Args:
theme(str): The theme to apply, either "light" or "dark".
theme(str): Either "light" or "dark".
"""
apply_theme(theme)
set_theme(theme) # emits theme_updated and applies palette globally
def event(self, event):
if event.type() == QEvent.Type.StatusTip:

View File

@@ -1,3 +1,4 @@
from bec_lib.device import Device
from bec_qthemes import material_icon
from qtpy.QtCore import Qt
from qtpy.QtWidgets import QHBoxLayout, QLabel, QToolButton, QVBoxLayout, QWidget
@@ -5,6 +6,7 @@ from qtpy.QtWidgets import QHBoxLayout, QLabel, QToolButton, QVBoxLayout, QWidge
from bec_widgets.utils.bec_connector import ConnectionConfig
from bec_widgets.utils.bec_widget import BECWidget
from bec_widgets.utils.error_popups import SafeProperty, SafeSlot
from bec_widgets.utils.ophyd_kind_util import Kind
from bec_widgets.widgets.containers.dock.dock import BECDock
from bec_widgets.widgets.utility.signal_label.signal_label import SignalLabel
@@ -35,9 +37,9 @@ class SignalDisplay(BECWidget, QWidget):
@SafeSlot()
def _refresh(self):
if self.device in self.dev:
self.dev.get(self.device).read(cached=False)
self.dev.get(self.device).read_configuration(cached=False)
if (dev := self.dev.get(self.device)) is not None:
dev.read()
dev.read_configuration()
def _add_refresh_button(self):
button_holder = QWidget()
@@ -63,11 +65,26 @@ class SignalDisplay(BECWidget, QWidget):
self._add_refresh_button()
if self._device in self.dev:
for sig in self.dev[self.device]._info.get("signals", {}).keys():
if isinstance(self.dev[self.device], Device):
for sig, info in self.dev[self.device]._info.get("signals", {}).items():
if info.get("kind_str") in [
Kind.hinted.name,
Kind.normal.name,
Kind.config.name,
]:
self._content_layout.addWidget(
SignalLabel(
device=self._device,
signal=sig,
show_select_button=False,
show_default_units=True,
)
)
else:
self._content_layout.addWidget(
SignalLabel(
device=self._device,
signal=sig,
signal=self._device,
show_select_button=False,
show_default_units=True,
)

View File

@@ -2,7 +2,7 @@ from __future__ import annotations
import sys
import traceback
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
from bec_lib.device import Device, Signal
from bec_lib.endpoints import MessageEndpoints
@@ -143,6 +143,14 @@ class SignalLabel(BECWidget, QWidget):
"show_default_units.setter",
"show_select_button",
"show_select_button.setter",
"show_hinted_signals",
"show_hinted_signals.setter",
"show_normal_signals",
"show_normal_signals.setter",
"show_config_signals",
"show_config_signals.setter",
"display_array_data",
"display_array_data.setter",
]
def __init__(
@@ -183,8 +191,9 @@ class SignalLabel(BECWidget, QWidget):
self._dtype = None
self._show_hinted_signals: bool = True
self._show_normal_signals: bool = False
self._show_config_signals: bool = False
self._show_normal_signals: bool = True
self._show_config_signals: bool = True
self._display_array_data: bool = False
self._outer_layout = QHBoxLayout()
self._layout = QHBoxLayout()
@@ -197,7 +206,7 @@ class SignalLabel(BECWidget, QWidget):
self._update_label()
self._label.setLayout(self._layout)
self._value: str = ""
self._value: Any = ""
self._display = QLabel()
self._layout.addWidget(self._display)
@@ -210,6 +219,8 @@ class SignalLabel(BECWidget, QWidget):
self._select_button.clicked.connect(self.show_choice_dialog)
self.get_bec_shortcuts()
self._device_obj = self.dev.get(self._device)
self._signal_key, self._signal_info = "", {}
self._connected: bool = False
self.connect_device()
@@ -226,6 +237,7 @@ class SignalLabel(BECWidget, QWidget):
@SafeSlot()
def _process_dialog(self, device: str, signal: str):
signal = signal or device
self.disconnect_device()
self.device = device
self.signal = signal
@@ -241,45 +253,34 @@ class SignalLabel(BECWidget, QWidget):
def connect_device(self):
"""Subscribe to the Redis topic for the device to display"""
if not self._connected and self._device and self._device in self.dev:
self._connected = True
self._read_endpoint = MessageEndpoints.device_read(self._device)
self._signal_key, self._signal_info = self._signal_key_and_info()
self._manual_read()
self._read_endpoint = MessageEndpoints.device_readback(self._device)
self._read_config_endpoint = MessageEndpoints.device_read_configuration(self._device)
self.bec_dispatcher.connect_slot(self.on_device_readback, self._read_endpoint)
self.bec_dispatcher.connect_slot(self.on_device_readback, self._read_config_endpoint)
self._manual_read()
self._connected = True
self.set_display_value(self._value)
def disconnect_device(self):
"""Unsubscribe from the Redis topic for the device to display"""
if self._connected:
self._connected = False
self.bec_dispatcher.disconnect_slot(self.on_device_readback, self._read_endpoint)
self.bec_dispatcher.disconnect_slot(self.on_device_readback, self._read_config_endpoint)
self._connected = False
def _manual_read(self):
if self._device is None or not isinstance(
(device := self.dev.get(self._device)), Device | Signal
):
self._units = ""
self._value = "__"
if not isinstance(self._device_obj, Device | Signal):
self._value, self._units = "__", ""
return
signal, info = (
(
getattr(device, self.signal, None),
device._info.get("signals", {}).get(self._signal, {}).get("describe", {}),
)
if isinstance(device, Device)
else (device, device.describe().get(self._device))
)
if not isinstance(signal, Signal): # Avoid getting other attributes of device, e.g. methods
signal = None
if signal is None:
self._units = ""
self._value = "__"
reading = (self._device_obj.read() or {}) | (self._device_obj.read_configuration() or {})
value = reading.get(self._signal_key, {}).get("value")
if value is None:
self._value, self._units = "__", ""
return
self._value = signal.get()
self._units = info.get("egu", "")
self._dtype = info.get("dtype", "float")
self._value = value
self._units = self._signal_info.get("egu", "")
self._dtype = self._signal_info.get("dtype", "float")
@SafeSlot(dict, dict)
def on_device_readback(self, msg: dict, metadata: dict) -> None:
@@ -287,8 +288,7 @@ class SignalLabel(BECWidget, QWidget):
Update the display with the new value.
"""
try:
signal_to_read = self._patch_hinted_signal()
_value = msg["signals"].get(signal_to_read, {}).get("value")
_value = msg["signals"].get(self._signal_key, {}).get("value")
if _value is not None:
self._value = _value
self.set_display_value(self._value)
@@ -298,13 +298,25 @@ class SignalLabel(BECWidget, QWidget):
f"Error processing incoming reading: {msg}, handled with exception: {''.join(traceback.format_exception(e))}"
)
def _patch_hinted_signal(self):
if self.dev[self._device]._info["signals"] == {}:
return self._signal
signal_info = self.dev[self._device]._info["signals"][self._signal]
return (
signal_info["obj_name"] if signal_info["kind_str"] == Kind.hinted.name else self._signal
)
def _signal_key_and_info(self) -> tuple[str, dict]:
if isinstance(self._device_obj, Device):
signal_info = self._device_obj._info["signals"][self._signal]
if signal_info["kind_str"] == Kind.hinted.name:
return signal_info["obj_name"], signal_info
else:
return f"{self._device}_{self._signal}", signal_info
elif isinstance(self._device_obj, Signal):
return self._device, self._device_obj._info["describe_configuration"]
return "", {}
# if self.dev[self._device]._info["signals"] == {}:
# return self._signal or self._device
# signal_info = self.dev[self._device]._info["signals"][self._signal]
# return (
# signal_info["obj_name"]
# if signal_info["kind_str"] == Kind.hinted.name
# else (self._signal or self._device)
# )
@SafeProperty(str)
def device(self) -> str:
@@ -315,6 +327,7 @@ class SignalLabel(BECWidget, QWidget):
def device(self, value: str) -> None:
self.disconnect_device()
self._device = value
self._device_obj = self.dev.get(self._device)
self._config.device = value
self.connect_device()
self._update_label()
@@ -382,6 +395,16 @@ class SignalLabel(BECWidget, QWidget):
self._decimal_places = value
self._update_label()
@SafeProperty(bool)
def display_array_data(self) -> bool:
"""Displays the full data from array signals if set to True."""
return self._display_array_data
@display_array_data.setter
def display_array_data(self, value: bool) -> None:
self._display_array_data = value
self.set_display_value(self._value)
@SafeProperty(bool)
def show_hinted_signals(self) -> bool:
"""In the signal selection menu, show hinted signals"""
@@ -409,7 +432,9 @@ class SignalLabel(BECWidget, QWidget):
def show_normal_signals(self, value: bool) -> None:
self._show_normal_signals = value
def _format_value(self, value: str):
def _format_value(self, value: Any):
if self._dtype == "array" and not self.display_array_data:
return "ARRAY DATA"
if self._decimal_places == 0:
return value
try:

View File

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "bec_widgets"
version = "2.28.0"
version = "2.30.0"
description = "BEC Widgets"
requires-python = ">=3.10"
classifiers = [

View File

@@ -2,6 +2,7 @@ from typing import TYPE_CHECKING
from unittest import mock
import pytest
from bec_lib.device import Device
from qtpy.QtCore import QPoint, Qt
from qtpy.QtWidgets import QTabWidget
@@ -9,6 +10,9 @@ from bec_widgets.widgets.services.device_browser.device_browser import DeviceBro
from bec_widgets.widgets.services.device_browser.device_item.device_config_form import (
DeviceConfigForm,
)
from bec_widgets.widgets.services.device_browser.device_item.device_signal_display import (
SignalDisplay,
)
from .client_mocks import mocked_client
@@ -142,3 +146,39 @@ def test_device_deletion(device_browser, qtbot):
assert widget.device in device_browser._device_items
qtbot.mouseClick(widget.delete_button, Qt.LeftButton)
qtbot.waitUntil(lambda: widget.device not in device_browser._device_items, timeout=10000)
def test_signal_display(mocked_client, qtbot):
signal_display = SignalDisplay(client=mocked_client, device="test_device")
qtbot.addWidget(signal_display)
device_mock = mock.MagicMock()
signal_display.dev = {"test_device": device_mock}
signal_display._refresh()
device_mock.read.assert_called()
device_mock.read_configuration.assert_called()
def test_signal_display_no_device(mocked_client, qtbot):
device_mock = mock.MagicMock()
mocked_client.client.device_manager.devices = {"test_device_1": device_mock}
signal_display = SignalDisplay(client=mocked_client, device="test_device_2")
qtbot.addWidget(signal_display)
assert (
signal_display._content_layout.itemAt(1).widget().text()
== "Device test_device_2 not found in device manager!"
)
signal_display._refresh()
device_mock.read.assert_not_called()
device_mock.read_configuration.assert_not_called()
def test_signal_display_omitted_not_added(mocked_client, qtbot):
device_mock = mock.MagicMock(spec=Device)
device_mock._info = {"signals": {"signal_1": {"kind_str": "omitted"}}}
signal_display = SignalDisplay(client=mocked_client, device="test_device_1")
signal_display.dev = {"test_device_1": device_mock}
signal_display._populate()
qtbot.addWidget(signal_display)
assert signal_display._content_layout.itemAt(1).widget() is None

View File

@@ -0,0 +1,340 @@
import pytest
from qtpy import QtCore, QtGui, QtWidgets
from bec_widgets.utils.error_popups import ErrorPopupUtility
from bec_widgets.widgets.containers.main_window.addons.notification_center.notification_banner import (
DARK_PALETTE,
LIGHT_PALETTE,
SEVERITY,
BECNotificationBroker,
NotificationCentre,
NotificationIndicator,
NotificationToast,
SeverityKind,
)
from .client_mocks import mocked_client
@pytest.fixture
def toast(qtbot):
"""Return a NotificationToast with a very short lifetime (50 ms) for fast tests."""
t = NotificationToast(
title="Test Title", body="Test Body", kind=SeverityKind.WARNING, lifetime_ms=50 # 0.05 s
)
qtbot.addWidget(t)
qtbot.waitExposed(t)
return t
def test_initial_state(toast):
"""Constructor should correctly propagate title / body / kind."""
assert toast.title == "Test Title"
assert toast.body == "Test Body"
assert toast.kind == SeverityKind.WARNING
# progress bar height fixed at 4 px
assert toast.progress.maximumHeight() == 4
def test_apply_theme_updates_colours(qtbot, toast):
"""apply_theme("light") should inject LIGHT palette colours into stylesheets."""
toast.apply_theme("light")
assert LIGHT_PALETTE["title"] in toast._title_lbl.styleSheet()
toast.apply_theme("dark")
assert DARK_PALETTE["title"] in toast._title_lbl.styleSheet()
def test_expired_signal(qtbot, toast):
"""Toast must emit expired once its lifetime finishes."""
with qtbot.waitSignal(toast.expired, timeout=1000):
pass
assert toast._expired
def test_closed_signal(qtbot, toast):
"""Calling close() must emit closed."""
with qtbot.waitSignal(toast.closed, timeout=1000):
toast.close()
def test_property_setters_update_ui(qtbot, toast):
"""Changing properties through setters should update both state and label text."""
# title
toast.title = "New Title"
assert toast.title == "New Title"
assert toast._title_lbl.text() == "New Title"
# body
toast.body = "New Body"
assert toast.body == "New Body"
assert toast._body_lbl.text() == "New Body"
# kind
toast.kind = SeverityKind.MINOR
assert toast.kind == SeverityKind.MINOR
expected_color = SEVERITY["minor"]["color"]
assert toast._accent_color.name() == QtGui.QColor(expected_color).name()
# traceback
new_tb = "Traceback: divide by zero"
toast.traceback = new_tb
assert toast.traceback == new_tb
assert toast.trace_view.toPlainText() == new_tb
def _make_enter_event(widget):
"""Utility: synthetic QEnterEvent centred on *widget*."""
centre = widget.rect().center()
local = QtCore.QPointF(centre)
scene = QtCore.QPointF(widget.mapTo(widget.window(), centre))
global_ = QtCore.QPointF(widget.mapToGlobal(centre))
return QtGui.QEnterEvent(local, scene, global_)
def test_time_label_toggle_absolute(qtbot, toast):
"""Hovering time-label switches between relative and absolute timestamp."""
rel_text = toast.time_lbl.text()
# Enter
QtWidgets.QApplication.sendEvent(toast.time_lbl, _make_enter_event(toast.time_lbl))
qtbot.wait(100)
abs_text = toast.time_lbl.text()
assert abs_text != rel_text and "-" in abs_text and ":" in abs_text
# Leave
QtWidgets.QApplication.sendEvent(toast.time_lbl, QtCore.QEvent(QtCore.QEvent.Leave))
qtbot.wait(100)
assert toast.time_lbl.text() != abs_text
def test_hover_pauses_and_resumes_expiry(qtbot):
"""Countdown must pause on hover and resume on leave."""
t = NotificationToast(title="Hover", body="x", kind=SeverityKind.INFO, lifetime_ms=200)
qtbot.addWidget(t)
qtbot.waitExposed(t)
qtbot.wait(50) # allow animation to begin
# Pause
QtWidgets.QApplication.sendEvent(t, _make_enter_event(t))
qtbot.wait(250) # longer than lifetime, but hover keeps it alive
assert not t._expired
# Resume
QtWidgets.QApplication.sendEvent(t, QtCore.QEvent(QtCore.QEvent.Leave))
with qtbot.waitSignal(t.expired, timeout=500):
pass
assert t._expired
def test_toast_paint_event(qtbot):
"""
Grabbing the widget as a pixmap forces paintEvent to execute.
The test passes if no exceptions occur and the resulting pixmap is valid.
"""
t = NotificationToast(title="Paint", body="Check", kind=SeverityKind.INFO, lifetime_ms=0)
qtbot.addWidget(t)
t.resize(420, 160)
t.show()
qtbot.waitExposed(t)
pix = t.grab()
assert not pix.isNull()
# ------------------------------------------------------------------------
# NotificationCentre tests
# ------------------------------------------------------------------------
@pytest.fixture
def centre(qtbot, mocked_client):
"""NotificationCentre embedded in a live parent widget kept alive for the test."""
parent = QtWidgets.QWidget()
parent.resize(600, 400)
ctr = NotificationCentre(parent=parent, fixed_width=300, margin=8)
broker = BECNotificationBroker(client=mocked_client)
layout = QtWidgets.QVBoxLayout(parent)
layout.setContentsMargins(0, 0, 0, 0)
layout.addWidget(ctr)
# Keep a Python reference so GC doesn't drop the parent (and cascadedelete centre)
ctr._test_parent_ref = parent # type: ignore[attr-defined]
qtbot.addWidget(parent)
qtbot.addWidget(ctr)
parent.show()
qtbot.waitExposed(parent)
yield ctr
broker.reset_singleton()
def _post(ctr: NotificationCentre, kind=SeverityKind.INFO, title="T", body="B"):
"""Convenience wrapper that posts a toast and returns it."""
return ctr.add_notification(title=title, body=body, kind=kind, lifetime_ms=0)
# ------------------------------------------------------------------------
# Tests
# ------------------------------------------------------------------------
def test_add_notification_emits_signal(qtbot, centre):
"""Adding a toast emits toast_added and makes centre visible."""
with qtbot.waitSignal(centre.toast_added, timeout=500) as sig:
toast = _post(centre, SeverityKind.INFO)
assert toast in centre.toasts
assert sig.args == [SeverityKind.INFO.value]
def test_counts_updated(qtbot, centre):
"""counts_updated reflects current per-kind counts."""
seen = []
centre.counts_updated.connect(lambda d: seen.append(d.copy()))
_post(centre, SeverityKind.INFO)
_post(centre, SeverityKind.WARNING)
qtbot.wait(100)
assert seen[-1][SeverityKind.INFO] == 1
assert seen[-1][SeverityKind.WARNING] == 1
centre.clear_all()
qtbot.wait(100)
assert seen[-1][SeverityKind.INFO] == 0
assert seen[-1][SeverityKind.WARNING] == 0
def test_filtering_hides_unrelated_toasts(centre):
info = _post(centre, SeverityKind.INFO)
warn = _post(centre, SeverityKind.WARNING)
centre.apply_filter({SeverityKind.INFO})
assert info.isVisible()
assert not warn.isVisible()
centre.apply_filter(None)
assert warn.isVisible()
def test_hide_show_all(qtbot, centre):
_post(centre, SeverityKind.MINOR)
centre.hide_all()
assert not centre.isVisible()
centre.show_all()
assert centre.isVisible()
assert all(t.isVisible() for t in centre.toasts)
def test_clear_all(qtbot, centre):
_post(centre, SeverityKind.INFO)
_post(centre, SeverityKind.WARNING)
# expect two toast_removed emissions
for _ in range(2):
qtbot.waitSignal(centre.toast_removed, timeout=500, raising=False)
centre.clear_all()
assert not centre.toasts
assert not centre.isVisible()
def test_theme_propagation(qtbot, centre):
toast = _post(centre, SeverityKind.INFO)
centre.apply_theme("light")
assert LIGHT_PALETTE["title"] in toast._title_lbl.styleSheet()
# ------------------------------------------------------------------------
# NotificationIndicator tests
# ------------------------------------------------------------------------
@pytest.fixture
def indicator(qtbot, centre):
"""Indicator wired to the same centre used in centre fixture."""
ind = NotificationIndicator()
qtbot.addWidget(ind)
# wire signals
centre.counts_updated.connect(ind.update_counts)
ind.filter_changed.connect(centre.apply_filter)
ind.show_all_requested.connect(centre.show_all)
ind.hide_all_requested.connect(centre.hide_all)
return ind
def _emit_counts(centre: NotificationCentre, info=0, warn=0, minor=0, major=0):
"""Helper to create dummy toasts and update counts."""
for _ in range(info):
_post(centre, SeverityKind.INFO)
for _ in range(warn):
_post(centre, SeverityKind.WARNING)
for _ in range(minor):
_post(centre, SeverityKind.MINOR)
for _ in range(major):
_post(centre, SeverityKind.MAJOR)
def test_indicator_updates_visibility(qtbot, centre, indicator):
"""Indicator shows/hides buttons based on counts."""
_emit_counts(centre, info=1)
qtbot.wait(50)
# "info" button visible, others hidden
assert indicator._btn[SeverityKind.INFO].isVisible()
assert not indicator._btn[SeverityKind.WARNING].isVisible()
# add warning toast → warning button appears
_emit_counts(centre, warn=1)
qtbot.wait(50)
assert indicator._btn[SeverityKind.WARNING].isVisible()
# clear all → indicator hides itself
centre.clear_all()
qtbot.wait(50)
assert not indicator.isVisible()
def test_indicator_filter_buttons(qtbot, centre, indicator):
"""Toggling buttons emits appropriate filter signals."""
# add two kinds so indicator is visible
_emit_counts(centre, info=1, warn=1)
qtbot.wait(200)
# click INFO button
with qtbot.waitSignal(indicator.filter_changed, timeout=500) as sig:
qtbot.mouseClick(indicator._btn[SeverityKind.INFO], QtCore.Qt.LeftButton)
assert sig.args[0] == {SeverityKind.INFO}
def test_broker_posts_notification(qtbot, centre, mocked_client):
"""post_notification should create a toast in the centre with correct data."""
broker = BECNotificationBroker(parent=None, client=mocked_client, centre=centre)
broker._err_util = ErrorPopupUtility()
msg = {
"alarm_type": "ValueError",
"msg": "test alarm",
"severity": 2, # MAJOR
"source": {"device": "samx", "source": "async_file_writer"},
}
broker.post_notification(msg, meta={})
qtbot.wait(200) # allow toast to be posted
# One toast should now exist
assert len(centre.toasts) == 1
toast = centre.toasts[0]
assert toast.title == "ValueError"
assert "Error occurred. See details." in toast.body
assert toast.kind == SeverityKind.MAJOR
assert toast._lifetime == 0

View File

@@ -84,7 +84,15 @@ def test_initialization(signal_label: SignalLabel):
def test_initialization_with_device(qtbot, mocked_client: MagicMock):
with patch.object(mocked_client.device_manager.devices.samx, "_info", SAMX_INFO_DICT):
with (
patch.object(mocked_client.device_manager.devices.samx, "_info", SAMX_INFO_DICT),
patch.object(
mocked_client.device_manager.devices.samx,
"_get_root_recursively",
lambda *_: (MagicMock(),),
),
):
widget = SignalLabel(device="samx", signal="readback", client=mocked_client)
qtbot.addWidget(widget)
qtbot.waitExposed(widget)
@@ -188,6 +196,8 @@ def test_choice_dialog_with_no_client(qtbot):
def test_dialog_has_signals(signal_label: SignalLabel, qtbot):
signal_label.show_config_signals = False
signal_label.show_normal_signals = False
signal_label._process_dialog = MagicMock()
dialog = signal_label.show_choice_dialog()
qtbot.waitUntil(dialog.button_box.button(QDialogButtonBox.Ok).isVisible, timeout=500)