bec_widgets/tests/unit_tests/test_utils_bec_signal_proxy.py

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, cleanup_signal_proxy
from bec_widgets.widgets.dap.dap_combo_box.dap_combo_box import DapComboBox

from .conftest import create_widget


@pytest.fixture
def dap_combo_box(qtbot, mocked_client):
    """Fixture for TextBox widget to test BECSignalProxy with a simple widget"""
    with mock.patch(
        "bec_widgets.widgets.dap.dap_combo_box.dap_combo_box.DapComboBox._validate_dap_model",
        return_value=True,
    ):
        widget = create_widget(qtbot, DapComboBox, client=mocked_client)
        yield widget


def test_bec_signal_proxy(qtbot, dap_combo_box):
    """Test BECSignalProxy"""
    proxy_container = []

    def proxy_callback(*args):
        """A simple callback function for the proxy"""
        proxy_container.append(args)

    container = []

    def all_callbacks(*args):
        """A simple callback function for all signal calls"""
        container.append(args)

    proxy = BECSignalProxy(dap_combo_box.x_axis_updated, rateLimit=25, slot=proxy_callback)
    pg_proxy = pg.SignalProxy(dap_combo_box.x_axis_updated, rateLimit=25, slot=all_callbacks)
    qtbot.wait(200)
    # Test that the proxy is blocked
    assert container == []
    assert proxy_container == []
    # Set first value
    dap_combo_box.x_axis = "samx"
    qtbot.waitSignal(dap_combo_box.x_axis_updated, timeout=1000)
    qtbot.wait(100)
    assert container == [(("samx",),)]
    assert proxy.blocked is True
    assert proxy_container == [(("samx",),)]
    # Set new value samy
    dap_combo_box.x_axis = "samy"
    qtbot.waitSignal(dap_combo_box.x_axis_updated, timeout=1000)
    qtbot.wait(100)
    assert container == [(("samx",),), (("samy",),)]
    assert proxy.blocked is True
    assert proxy_container == [(("samx",),)]
    # Set new value samz
    dap_combo_box.x_axis = "samz"
    qtbot.waitSignal(dap_combo_box.x_axis_updated, timeout=1000)
    qtbot.wait(100)
    assert container == [(("samx",),), (("samy",),), (("samz",),)]
    assert proxy.blocked is True
    proxy.unblock_proxy()
    qtbot.waitSignal(proxy.is_blocked, timeout=1000)
    qtbot.wait(100)
    assert proxy.blocked is True
    assert proxy_container == [(("samx",),), (("samz",),)]
    # Unblock the proxy again, no new argument received.
    # The callback should not be called again.
    proxy.unblock_proxy()
    qtbot.waitSignal(proxy.is_blocked, timeout=1000)
    qtbot.wait(100)
    assert proxy.blocked is False
    assert proxy_container == [(("samx",),), (("samz",),)]


def test_bec_signal_proxy_timeout(qtbot, dap_combo_box):
    """
    Test that BECSignalProxy auto-unblocks after the specified timeout if no manual unblock
    occurs in the interim.
    """
    proxy_container = []

    def proxy_callback(*args):
        proxy_container.append(args)

    # Create the proxy with a short 1-second timeout
    proxy = BECSignalProxy(
        dap_combo_box.x_axis_updated, rateLimit=25, slot=proxy_callback, timeout=1.0
    )

    # Initially, ensure it's not blocked
    assert proxy.blocked is False

    # Trigger the signal once (samx) -> the proxy should block
    dap_combo_box.x_axis = "samx"
    qtbot.waitSignal(dap_combo_box.x_axis_updated, timeout=1000)
    qtbot.wait(100)
    assert proxy.blocked is True
    # The first signal should be passed immediately to the callback
    assert proxy_container == [(("samx",),)]

    # While still blocked, set another value (samz)
    dap_combo_box.x_axis = "samz"
    qtbot.waitSignal(dap_combo_box.x_axis_updated, timeout=1000)
    qtbot.wait(100)
    # Proxy is still blocked, so the callback shouldn't see "samz" yet
    assert len(proxy_container) == 1

    # Wait just under 1 second -> should still be blocked
    qtbot.wait(700)
    assert proxy.blocked is True

    # Wait a bit more than 1 s
    qtbot.wait(2000)

    # Wait to catch the is_blocked signal that indicates it has unblocked
    qtbot.waitSignal(proxy.is_blocked, timeout=2000)
    # Now it should be unblocked
    assert proxy.blocked is False

    # The second value "samz" should have been forwarded after auto-unblocking
    assert proxy_container == [(("samx",),), (("samz",),)]


def test_bec_signal_proxy_replays_argless_emissions_on_unblock(qtbot):
    """Emissions of an argument-less signal that arrive while the proxy is blocked must be
    replayed on unblock: their args tuple is identical to the forwarded one, so an
    args-comparison alone would silently drop them (the DAP-fit-never-arrives e2e flake:
    one missed dap_response blocked the proxy, and every update emitted during the scan
    coalesced into nothing)."""
    from qtpy.QtCore import QObject, Signal

    class _Src(QObject):
        sig = Signal()

    src = _Src()
    calls = []
    proxy = BECSignalProxy(src.sig, rateLimit=25, slot=lambda *_: calls.append(1), timeout=10.0)

    src.sig.emit()  # forwarded; proxy blocks awaiting unblock_proxy
    qtbot.waitUntil(lambda: len(calls) == 1)
    assert proxy.blocked is True

    src.sig.emit()  # coalesces while blocked - identical (empty) args
    proxy.unblock_proxy()

    # the coalesced emission is replayed instead of being dropped
    qtbot.waitUntil(lambda: len(calls) == 2)
    assert proxy.blocked is True  # blocked again by the replayed emission

    # no pending emission this time: unblocking must not replay anything
    proxy.unblock_proxy()
    qtbot.wait(150)
    assert len(calls) == 2
    assert proxy.blocked is False
    proxy.cleanup()


def test_bec_signal_proxy_still_drops_identical_args_on_unblock(qtbot):
    """Emissions whose arguments equal the already-forwarded ones stay coalesced on
    unblock - only argument-less signals are replayed unconditionally."""
    from qtpy.QtCore import QObject, Signal

    class _Src(QObject):
        sig = Signal(str)

    src = _Src()
    calls = []
    proxy = BECSignalProxy(src.sig, rateLimit=25, slot=lambda *a: calls.append(a), timeout=10.0)

    src.sig.emit("samx")  # forwarded; proxy blocks
    qtbot.waitUntil(lambda: len(calls) == 1)
    src.sig.emit("samx")  # identical args while blocked

    proxy.unblock_proxy()
    qtbot.wait(150)
    assert len(calls) == 1  # unchanged args are not replayed
    assert proxy.blocked is False
    proxy.cleanup()


def test_bec_signal_proxy_cleanup_drops_queued_emission(qtbot):
    """An emission that is still queued in the delivery timer inherited from
    pg.SignalProxy when cleanup() runs must never reach the slot."""
    from qtpy.QtCore import QObject, Signal

    class _Src(QObject):
        sig = Signal()

    src = _Src()
    calls = []
    proxy = BECSignalProxy(src.sig, rateLimit=25, slot=lambda *_: calls.append(1), timeout=10.0)

    src.sig.emit()  # queued: the inherited delivery timer forwards it on the next tick
    assert calls == []
    assert proxy.timer.timer.isActive()  # pg's ThreadsafeTimer wraps a QTimer
    proxy.cleanup()
    assert not proxy.timer.timer.isActive()
    assert proxy.args is None

    qtbot.wait(150)
    assert calls == []
    assert not proxy.timer.timer.isActive()


def test_bec_signal_proxy_ignores_source_after_cleanup(qtbot):
    """After cleanup() the proxy is detached from its source signal: later emissions are
    neither forwarded to the slot nor touch the deleted auto-unblock watchdog."""
    from qtpy.QtCore import QObject, Signal

    class _Src(QObject):
        sig = Signal()

    src = _Src()
    calls = []
    proxy = BECSignalProxy(src.sig, rateLimit=25, slot=lambda *_: calls.append(1), timeout=10.0)
    proxy.cleanup()
    qtbot.wait(20)  # let the watchdog's deleteLater run

    src.sig.emit()
    proxy.unblock_proxy()
    qtbot.wait(150)
    assert calls == []
    assert proxy.blocked is False
    proxy.cleanup()  # a second cleanup is a no-op


def test_cleanup_signal_proxy_stops_queued_delivery(qtbot):
    """
    pg.SignalProxy.disconnect() alone keeps the repeating delivery timer of a queued emission
    running; cleanup_signal_proxy also stops it, drops the queued arguments and detaches the
    source, so nothing reaches the slot afterwards.
    """
    from qtpy.QtCore import QObject, Signal

    class _Src(QObject):
        sig = Signal()

    src = _Src()
    calls, ticks = [], []
    proxy = pg.SignalProxy(src.sig, rateLimit=25, slot=lambda *_: calls.append(1))
    proxy.timer.timeout.connect(lambda: ticks.append(1))

    src.sig.emit()  # queued: the delivery timer forwards it on the next tick
    assert proxy.args is not None
    assert proxy.timer.timer.isActive()  # pg's ThreadsafeTimer wraps a QTimer

    cleanup_signal_proxy(proxy)
    assert not proxy.timer.timer.isActive()
    assert proxy.args is None

    src.sig.emit()  # detached from the source
    qtbot.wait(100)
    assert calls == []
    assert ticks == []
    assert not proxy.timer.timer.isActive()
    cleanup_signal_proxy(proxy)  # a second call is a no-op