mirror of
https://github.com/bec-project/bec_widgets.git
synced 2026-10-22 16:31:28 +02:00
304 lines
9.9 KiB
Python
304 lines
9.9 KiB
Python
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import shiboken6
|
|
from bec_lib.logger import bec_logger
|
|
from qtpy.QtCore import QObject, Qt, Signal
|
|
from qtpy.QtWidgets import QMessageBox
|
|
|
|
from bec_widgets.utils.error_popups import ErrorPopupUtility, ExampleWidget, SafeProperty, SafeSlot
|
|
|
|
|
|
class TestSafePropertyClass(QObject):
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self._my_value = 10 # internal store
|
|
|
|
@SafeProperty(int, default=-1)
|
|
def my_value(self) -> int:
|
|
# artificially raise if it's 999 for testing
|
|
if self._my_value == 999:
|
|
raise ValueError("Invalid internal state in getter!")
|
|
return self._my_value
|
|
|
|
@my_value.setter
|
|
def my_value(self, val: int):
|
|
# artificially raise if user sets -999 for testing
|
|
if val == -999:
|
|
raise ValueError("Invalid user input in setter!")
|
|
self._my_value = val
|
|
|
|
|
|
class TestSafeSlotEmitter(QObject):
|
|
test_signal = Signal()
|
|
|
|
|
|
class TestSafeSlotClass(QObject):
|
|
"""
|
|
Test class to demonstrate the use of SafeSlot decorator.
|
|
"""
|
|
|
|
def __init__(self, parent=None, signal_obj: TestSafeSlotEmitter | None = None):
|
|
super().__init__(parent)
|
|
assert signal_obj is not None, "Signal object must be provided"
|
|
signal_obj.test_signal.connect(self.receive)
|
|
self.calls = 0
|
|
|
|
@SafeSlot()
|
|
def receive(self):
|
|
self.calls += 1
|
|
|
|
|
|
@pytest.fixture
|
|
def widget(qtbot):
|
|
test_widget = ExampleWidget()
|
|
qtbot.addWidget(test_widget)
|
|
qtbot.waitExposed(test_widget)
|
|
yield test_widget
|
|
test_widget.close()
|
|
|
|
|
|
@patch.object(QMessageBox, "exec_", return_value=QMessageBox.Ok)
|
|
def test_show_error_message_global(mock_exec, widget, qtbot):
|
|
"""
|
|
Test that an error popup is shown if global error popups are enabled
|
|
and the error_occurred signal is emitted manually.
|
|
"""
|
|
error_utility = ErrorPopupUtility()
|
|
error_utility.enable_global_error_popups(True)
|
|
|
|
with qtbot.waitSignal(error_utility.error_occurred, timeout=1000) as blocker:
|
|
error_utility.error_occurred.emit("Test Error", "This is a test error message.", widget)
|
|
|
|
assert blocker.signal_triggered
|
|
assert mock_exec.called
|
|
|
|
|
|
@pytest.mark.parametrize("global_pop", [False, True])
|
|
@patch.object(QMessageBox, "exec_", return_value=QMessageBox.Ok)
|
|
def test_slot_with_popup_on_error(mock_exec, widget, qtbot, global_pop):
|
|
"""
|
|
If the slot is decorated with @SafeSlot(popup_error=True),
|
|
we always expect a popup on error (and a signal) even if global popups are off.
|
|
"""
|
|
error_utility = ErrorPopupUtility()
|
|
error_utility.enable_global_error_popups(global_pop)
|
|
|
|
with qtbot.waitSignal(error_utility.error_occurred, timeout=500) as blocker:
|
|
widget.method_with_error_handling()
|
|
|
|
assert blocker.signal_triggered
|
|
assert mock_exec.called # Because popup_error=True forces popup
|
|
|
|
|
|
@pytest.mark.parametrize("global_pop", [False, True])
|
|
@patch.object(bec_logger.logger, "error")
|
|
@patch.object(QMessageBox, "exec_", return_value=QMessageBox.Ok)
|
|
def test_slot_no_popup_by_default_on_error(mock_exec, mock_log_error, widget, qtbot, global_pop):
|
|
"""
|
|
If the slot is decorated with @SafeSlot() (no popup_error=True),
|
|
we never show a popup, even if global popups are on,
|
|
because the code does not check 'enable_error_popup' for normal slots.
|
|
"""
|
|
error_utility = ErrorPopupUtility()
|
|
error_utility.enable_global_error_popups(global_pop)
|
|
|
|
# We do NOT expect a popup or signal in either case, since code only logs
|
|
with qtbot.assertNotEmitted(error_utility.error_occurred):
|
|
widget.method_without_error_handling()
|
|
|
|
assert not mock_exec.called
|
|
|
|
# Confirm logger.error(...) was called
|
|
mock_log_error.assert_called_once()
|
|
logged_msg = mock_log_error.call_args[0][0]
|
|
assert "ValueError" in logged_msg
|
|
assert "SafeSlot error in slot" in logged_msg
|
|
|
|
|
|
@pytest.mark.parametrize("global_pop", [False, True])
|
|
@patch.object(bec_logger.logger, "error")
|
|
@patch.object(QMessageBox, "exec_", return_value=QMessageBox.Ok)
|
|
def test_safe_property_getter_error(mock_exec, mock_log_error, qtbot, global_pop):
|
|
"""
|
|
If a property getter raises an error, we log it by default.
|
|
(No popup is shown unless code specifically calls it.)
|
|
"""
|
|
error_utility = ErrorPopupUtility()
|
|
error_utility.enable_global_error_popups(global_pop)
|
|
|
|
test_obj = TestSafePropertyClass()
|
|
test_obj._my_value = 999 # triggers ValueError in getter => logs => returns default (-1)
|
|
|
|
val = test_obj.my_value
|
|
assert val == -1
|
|
|
|
# No popup => mock_exec not called
|
|
assert not mock_exec.called
|
|
|
|
# logger.error(...) is called once
|
|
mock_log_error.assert_called_once()
|
|
logged_msg = mock_log_error.call_args[0][0]
|
|
assert "SafeProperty error in GETTER" in logged_msg
|
|
assert "ValueError" in logged_msg
|
|
|
|
|
|
@pytest.mark.parametrize("global_pop", [False, True])
|
|
@patch.object(bec_logger.logger, "error")
|
|
@patch.object(QMessageBox, "exec_", return_value=QMessageBox.Ok)
|
|
def test_safe_property_setter_error(mock_exec, mock_log_error, qtbot, global_pop):
|
|
"""
|
|
If a property setter raises an error, we log it by default.
|
|
(No popup is shown unless code specifically calls it.)
|
|
"""
|
|
error_utility = ErrorPopupUtility()
|
|
error_utility.enable_global_error_popups(global_pop)
|
|
|
|
test_obj = TestSafePropertyClass()
|
|
# Setting to -999 triggers setter error => logs => property returns None
|
|
test_obj.my_value = -999
|
|
|
|
# No popup => mock_exec not called
|
|
assert not mock_exec.called
|
|
|
|
# logger.error(...) is called once
|
|
mock_log_error.assert_called_once()
|
|
logged_msg = mock_log_error.call_args[0][0]
|
|
assert "SafeProperty error in SETTER" in logged_msg
|
|
assert "ValueError" in logged_msg
|
|
|
|
|
|
@pytest.mark.timeout(100)
|
|
@pytest.mark.parametrize("delivery", ["signal", "direct"])
|
|
def test_safe_slot_emit(qtbot, delivery):
|
|
"""SafeSlot executes both signal deliveries and ordinary Python calls."""
|
|
signal_obj = TestSafeSlotEmitter()
|
|
test_obj = TestSafeSlotClass(signal_obj=signal_obj)
|
|
if delivery == "signal":
|
|
signal_obj.test_signal.emit()
|
|
else:
|
|
test_obj.receive()
|
|
qtbot.waitUntil(lambda: test_obj.calls == 1, timeout=1000)
|
|
|
|
test_obj.deleteLater()
|
|
signal_obj.deleteLater()
|
|
|
|
|
|
class _ArgSignalEmitter(QObject):
|
|
str_signal = Signal(str)
|
|
plain_signal = Signal()
|
|
|
|
|
|
class _SafeConnectReceiver(QObject):
|
|
def __init__(self, parent=None):
|
|
super().__init__(parent)
|
|
self.calls = []
|
|
|
|
def on_value(self, value=None):
|
|
self.calls.append(value)
|
|
|
|
|
|
def test_safe_connect_slot_called_once_per_emission(qtbot):
|
|
"""Regression test for BW-003: a SafeConnect-ed slot must run exactly once
|
|
per emission, with the signal's payload."""
|
|
from bec_widgets.utils.error_popups import SafeConnect
|
|
|
|
emitter = _ArgSignalEmitter()
|
|
receiver = _SafeConnectReceiver()
|
|
|
|
SafeConnect(receiver, emitter.str_signal, receiver.on_value)
|
|
emitter.str_signal.emit("dark")
|
|
assert receiver.calls == ["dark"]
|
|
|
|
SafeConnect(receiver, emitter.plain_signal, receiver.on_value)
|
|
receiver.calls.clear()
|
|
emitter.plain_signal.emit()
|
|
assert receiver.calls == [None]
|
|
|
|
receiver.deleteLater()
|
|
emitter.deleteLater()
|
|
|
|
|
|
def test_safe_connect_removes_stale_connection(qtbot):
|
|
"""Regression test for BW-005: after the receiver dies, the next emission
|
|
must remove the wrapper from the signal instead of accumulating forever."""
|
|
import gc
|
|
|
|
from bec_widgets.utils.error_popups import SafeConnect
|
|
|
|
emitter = _ArgSignalEmitter()
|
|
receiver = _SafeConnectReceiver()
|
|
wrapper = SafeConnect(receiver, emitter.str_signal, receiver.on_value)
|
|
|
|
receiver.deleteLater()
|
|
qtbot.wait(10)
|
|
del receiver
|
|
gc.collect()
|
|
|
|
# First emission after death: wrapper detects the dead receiver and
|
|
# disconnects itself. A second disconnect attempt must then fail because
|
|
# the connection no longer exists.
|
|
emitter.str_signal.emit("after-death")
|
|
assert emitter.str_signal.disconnect(wrapper) is False
|
|
|
|
emitter.deleteLater()
|
|
|
|
|
|
def test_safe_connect_invalidated_before_queued_delivery(qtbot):
|
|
"""An inactive subscription must not deliver an already queued signal."""
|
|
from bec_widgets.utils.error_popups import SafeConnect
|
|
|
|
emitter = _ArgSignalEmitter()
|
|
receiver = _SafeConnectReceiver()
|
|
wrapper = SafeConnect(receiver, emitter.str_signal, receiver.on_value)
|
|
emitter.str_signal.disconnect(wrapper)
|
|
emitter.str_signal.connect(wrapper, Qt.ConnectionType.QueuedConnection)
|
|
|
|
emitter.str_signal.emit("queued")
|
|
assert receiver.calls == []
|
|
wrapper.invalidate()
|
|
qtbot.wait(10)
|
|
assert shiboken6.isValid(receiver)
|
|
assert receiver.calls == []
|
|
# Also cover direct calls retained by code outside the Qt connection.
|
|
wrapper("direct")
|
|
assert receiver.calls == []
|
|
receiver.deleteLater()
|
|
emitter.deleteLater()
|
|
|
|
|
|
def test_safe_connect_destruction_only_invalidates():
|
|
"""Destruction blocks delivery without disconnecting signals in the destructor."""
|
|
from bec_widgets.utils.error_popups import SafeConnect
|
|
|
|
emitter = _ArgSignalEmitter()
|
|
receiver = _SafeConnectReceiver()
|
|
wrapper = SafeConnect(receiver, emitter.str_signal, receiver.on_value)
|
|
with patch.object(wrapper, "_disconnect") as disconnect:
|
|
shiboken6.delete(receiver)
|
|
assert not wrapper._active
|
|
disconnect.assert_not_called()
|
|
emitter.str_signal.emit("after-death")
|
|
assert not wrapper._connected
|
|
shiboken6.delete(emitter)
|
|
|
|
|
|
def test_safe_slot_override_params_apply_to_single_call():
|
|
"""Regression test for BW-004: _override_slot_params must not permanently
|
|
mutate the decorator's defaults."""
|
|
|
|
class Raising(QObject):
|
|
@SafeSlot()
|
|
def boom(self):
|
|
raise ValueError("boom")
|
|
|
|
obj = Raising()
|
|
|
|
with pytest.raises(ValueError):
|
|
obj.boom(_override_slot_params={"raise_error": True})
|
|
|
|
# Default behavior (swallow + log) must be restored for the next call.
|
|
obj.boom()
|
|
|
|
obj.deleteLater()
|