bec_widgets/tests/unit_tests/test_rpc_server.py

import argparse
from threading import Thread
from unittest.mock import MagicMock, patch

import pytest
from bec_lib.service_config import ServiceConfig
from qtpy.QtCore import QEvent, QObject, Qt, QThread
from qtpy.QtTest import QSignalSpy
from qtpy.QtWidgets import QVBoxLayout, QWidget

from bec_widgets.applications import companion_app as companion_app_module
from bec_widgets.applications.companion_app import GUIServer
from bec_widgets.utils import rpc_server as rpc_server_module
from bec_widgets.utils.bec_connector import BECConnector
from bec_widgets.utils.bec_dispatcher import QtThreadSafeCallback
from bec_widgets.utils.rpc_server import RegistryNotReadyError, RPCServer, SingleshotRPCRepeat

from .client_mocks import mocked_client


class DummyWidget(BECConnector, QWidget):
    def __init__(self, parent=None, client=None, **kwargs):
        super().__init__(parent=parent, client=client, **kwargs)
        self.setObjectName("DummyWidget")


@pytest.fixture
def dummy_widget(qtbot, mocked_client):
    widget = DummyWidget(client=mocked_client)
    qtbot.addWidget(widget)
    return widget


@pytest.fixture
def gui_server():
    args = argparse.Namespace(
        config=None, id="gui_id", gui_class="LaunchWindow", gui_class_id="bec", hide=False
    )
    return GUIServer(args=args)


@pytest.fixture
def rpc_server(mocked_client):
    rpc_server = RPCServer(gui_id="test_gui", client=mocked_client)
    yield rpc_server
    rpc_server.shutdown()


def test_gui_server_start_server_without_service_config(gui_server):
    """
    Test that the server is started with the correct arguments.
    """
    assert gui_server.config is None
    assert gui_server.gui_id == "gui_id"
    assert gui_server.gui_class == "LaunchWindow"
    assert gui_server.gui_class_id == "bec"
    assert gui_server.hide is False


def test_gui_server_get_service_config(gui_server):
    """
    Test that the server is started with the correct arguments.
    """
    assert gui_server._get_service_config().config == ServiceConfig().config


def test_gui_server_signal_shutdown_closes_widgets_and_quits_app(gui_server):
    widget = MagicMock()
    gui_server.app = MagicMock()
    gui_server.app.topLevelWidgets.return_value = [widget]

    gui_server.request_shutdown()

    widget.close.assert_called_once()
    gui_server.app.quit.assert_called_once()


def test_gui_server_shutdown_is_idempotent(gui_server):
    gui_server.launcher_window = MagicMock()
    gui_server.dispatcher = MagicMock()

    with (
        patch.object(companion_app_module.shiboken6, "isValid", return_value=True),
        patch.object(companion_app_module.pylsp_server, "is_running", return_value=False),
    ):
        gui_server.shutdown()
        gui_server.shutdown()

    gui_server.launcher_window.close.assert_called_once()
    gui_server.launcher_window.deleteLater.assert_called_once()
    gui_server.dispatcher.stop_cli_server.assert_called_once()
    gui_server.dispatcher.disconnect_all.assert_called_once()


def test_rpc_server_system_capabilities_include_shutdown(rpc_server):
    assert rpc_server.run_system_rpc("system.list_capabilities", [], {}) == {
        "system.launch_dock_area": True,
        "system.shutdown": True,
        "system.get_display_info": True,
    }


class WindowEvents(QObject):
    def __init__(self, widget):
        super().__init__(widget)
        self.hide_events = 0
        self.show_events = 0
        widget.installEventFilter(self)

    def eventFilter(self, watched, event):
        if event.type() == QEvent.Hide:
            self.hide_events += 1
        elif event.type() == QEvent.Show:
            self.show_events += 1
        return False


@pytest.mark.parametrize("method", ["show", "raise"])
@pytest.mark.parametrize("stay_on_top", [False, True])
@pytest.mark.parametrize("platform", ["wayland", "wayland-egl", "cocoa"])
def test_repeated_show_and_raise_preserve_visibility_and_flags(
    rpc_server, dummy_widget, qapp, monkeypatch, method, stay_on_top, platform
):
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: platform)
    # A background window remains visible. Exercise raising, even on an offscreen test backend.
    monkeypatch.setattr(dummy_widget, "isActiveWindow", lambda: False)
    dummy_widget.setWindowFlag(Qt.WindowStaysOnTopHint, stay_on_top)
    dummy_widget.show()
    qapp.processEvents()
    flags = dummy_widget.windowFlags()
    events = WindowEvents(dummy_widget)

    for _ in range(3):
        handle = dummy_widget.windowHandle()
        rpc_server.run_rpc(dummy_widget, method, [], {})
        qapp.processEvents()
        assert dummy_widget.isVisible()
        assert dummy_widget.windowFlags() == flags
        if method == "raise" and platform.startswith("wayland"):
            assert dummy_widget.windowHandle() is not handle
        else:
            assert dummy_widget.windowHandle() is handle
    expected_remaps = 3 if method == "raise" and platform.startswith("wayland") else 0
    assert events.hide_events == events.show_events == expected_remaps


@pytest.mark.parametrize("method", ["show", "raise"])
def test_show_and_raise_restore_hidden_window(rpc_server, dummy_widget, method):
    dummy_widget.hide()
    rpc_server.run_rpc(dummy_widget, method, [], {})
    assert dummy_widget.isVisible()


@pytest.mark.parametrize("state", [Qt.WindowNoState, Qt.WindowMaximized, Qt.WindowFullScreen])
@pytest.mark.parametrize("platform", ["wayland", "xcb"])
def test_raise_restores_minimized_window_state(
    rpc_server, dummy_widget, monkeypatch, qtbot, state, platform
):
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: platform)
    dummy_widget.setWindowState(state | Qt.WindowMinimized)
    dummy_widget.show()
    assert dummy_widget.isMinimized()

    rpc_server.run_rpc(dummy_widget, "raise", [], {})

    qtbot.waitUntil(dummy_widget.isVisible)
    assert not dummy_widget.isMinimized()
    assert dummy_widget.windowState() & (Qt.WindowMaximized | Qt.WindowFullScreen) == state


def test_raise_requests_activation_after_showing_window(rpc_server, dummy_widget):
    dummy_widget.hide()
    visible_during_activation = []
    with patch.object(
        dummy_widget,
        "activateWindow",
        side_effect=lambda: visible_during_activation.append(dummy_widget.isVisible()),
    ):
        rpc_server.run_rpc(dummy_widget, "raise", [], {})
    assert visible_during_activation == [True]


@pytest.mark.parametrize("platform", ["wayland", "xcb", "cocoa"])
def test_raise_active_window_is_noop(rpc_server, dummy_widget, monkeypatch, platform):
    dummy_widget.show()
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: platform)
    monkeypatch.setattr(dummy_widget, "isActiveWindow", lambda: True)
    with (
        patch.object(dummy_widget, "setWindowFlags") as flags,
        patch.object(dummy_widget, "setWindowState") as state,
        patch.object(dummy_widget, "raise_") as raise_window,
        patch.object(dummy_widget, "activateWindow") as activate,
        patch.object(dummy_widget, "show") as show,
        patch.object(dummy_widget, "hide") as hide,
        patch.object(dummy_widget, "destroy") as destroy,
    ):
        rpc_server.run_rpc(dummy_widget, "raise", [], {})
    for operation in (flags, state, raise_window, activate, show, hide, destroy):
        operation.assert_not_called()


@pytest.mark.parametrize("platform", ["wayland", "wayland-egl", "cocoa"])
def test_raise_visible_inactive_window_requests_activation(
    rpc_server, dummy_widget, monkeypatch, qtbot, platform
):
    dummy_widget.show()
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: platform)
    monkeypatch.setattr(dummy_widget, "isActiveWindow", lambda: False)
    with (
        patch.object(dummy_widget, "activateWindow") as activate,
        patch.object(dummy_widget, "setWindowFlags") as flags,
    ):
        rpc_server.run_rpc(dummy_widget, "raise", [], {})
        qtbot.waitUntil(lambda: activate.called)
    activate.assert_called_once_with()
    flags.assert_not_called()


@pytest.mark.parametrize("platform", ["wayland", "wayland-egl"])
@pytest.mark.parametrize("initial_state", ["never_shown", "hidden", "visible"])
def test_wayland_raise_preserves_widget_and_replaces_existing_native_window(
    rpc_server, dummy_widget, monkeypatch, qapp, platform, initial_state
):
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: platform)
    monkeypatch.setattr(dummy_widget, "isActiveWindow", lambda: False)
    if initial_state != "never_shown":
        dummy_widget.show()
        qapp.processEvents()
    if initial_state == "hidden":
        dummy_widget.hide()
    handle = dummy_widget.windowHandle()
    native_destroyed = QSignalSpy(handle.destroyed) if handle is not None else None
    widget_destroyed = QSignalSpy(dummy_widget.destroyed)
    last_window_closed = QSignalSpy(qapp.lastWindowClosed)
    screen = dummy_widget.screen()
    events = WindowEvents(dummy_widget)
    visible_during_activation = []

    with patch.object(
        dummy_widget,
        "activateWindow",
        side_effect=lambda: visible_during_activation.append(dummy_widget.isVisible()),
    ):
        rpc_server.run_rpc(dummy_widget, "raise", [], {})
        assert dummy_widget.isVisible()
        qapp.processEvents()

    assert events.hide_events == int(initial_state == "visible")
    assert events.show_events == 1
    assert visible_during_activation == [True]
    assert dummy_widget.windowHandle() is not handle
    assert dummy_widget.screen() is screen
    if native_destroyed is not None:
        assert native_destroyed.count() == 1
    assert widget_destroyed.count() == last_window_closed.count() == 0


def test_wayland_raise_from_dispatcher_worker_runs_on_gui_thread(
    rpc_server, dummy_widget, monkeypatch, qtbot, qapp
):
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: "wayland")
    monkeypatch.setattr(dummy_widget, "isActiveWindow", lambda: False)
    dummy_widget.show()
    events = WindowEvents(dummy_widget)
    activation_threads = []
    callback = QtThreadSafeCallback(rpc_server.on_rpc_update)
    instruction = {
        "action": "raise",
        "parameter": {"gui_id": dummy_widget.gui_id, "args": [], "kwargs": {}},
    }
    with (
        patch.object(rpc_server, "send_response") as response,
        patch.object(
            dummy_widget,
            "activateWindow",
            side_effect=lambda: activation_threads.append(QThread.currentThread()),
        ),
    ):
        worker = Thread(target=callback, args=(instruction, {"request_id": "raise-from-worker"}))
        worker.start()
        worker.join(timeout=2)
        assert not worker.is_alive()
        qtbot.waitUntil(lambda: bool(activation_threads) and response.called)

    assert activation_threads == [qapp.thread()]
    assert events.hide_events == events.show_events == 1
    assert dummy_widget.isVisible()
    response.assert_called_once_with("raise-from-worker", True, {"result": None})


@pytest.mark.parametrize("method", ["hide", "close", "show"])
def test_visibility_request_after_wayland_raise_is_not_undone(
    rpc_server, dummy_widget, monkeypatch, qapp, method
):
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: "wayland")
    monkeypatch.setattr(dummy_widget, "isActiveWindow", lambda: False)
    dummy_widget.show()

    with patch.object(dummy_widget, "activateWindow") as activate:
        rpc_server.run_rpc(dummy_widget, "raise", [], {})
        if method == "close":
            dummy_widget.close()  # Also cover a close that does not arrive through RPC.
        else:
            rpc_server.run_rpc(dummy_widget, method, [], {})
        qapp.processEvents()

    assert dummy_widget.isVisible() == (method == "show")
    activate.assert_called_once_with()


@pytest.mark.parametrize("method", ["hide", "show"])
def test_child_visibility_after_wayland_raise_does_not_hide_window(
    rpc_server, dummy_widget, qtbot, qapp, monkeypatch, method
):
    window = QWidget()
    qtbot.addWidget(window)
    dummy_widget.setParent(window)
    dummy_widget.setVisible(method == "hide")
    window.show()
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: "wayland")
    monkeypatch.setattr(window, "isActiveWindow", lambda: False)

    rpc_server.run_rpc(window, "raise", [], {})
    rpc_server.run_rpc(dummy_widget, method, [], {})
    qapp.processEvents()

    assert window.isVisible()
    assert dummy_widget.isVisible() == (method == "show")


def test_wayland_recreation_preserves_dock_data_and_bec_connections(
    rpc_server, qtbot, qapp, monkeypatch, tmp_path
):
    monkeypatch.setenv("BECWIDGETS_PROFILE_DIR", str(tmp_path))
    dock_area = rpc_server._launch_dock_area(name="flomni", startup_profile="skip")
    window = dock_area.window()
    qtbot.addWidget(window)
    content = dock_area.new("TextBox", object_name="status")
    content.set_plain_text("Preserved status")
    waveform = dock_area.new("Waveform", object_name="waveform")
    curve = waveform.plot(x=[1, 2, 3], y=[4, 5, 6])
    qapp.processEvents()
    widgets = dock_area.widget_map()
    slots = dict(dock_area.bec_dispatcher._registered_slots)
    objects = [window, dock_area, content, waveform, curve]
    destroyed = [QSignalSpy(obj.destroyed) for obj in objects]
    size = window.size()
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: "wayland")
    monkeypatch.setattr(window, "isActiveWindow", lambda: False)

    for _ in range(3):
        rpc_server.run_rpc(window, "raise", [], {})
        qapp.processEvents()
        assert window.isVisible()
        assert window.size() == size
        assert content.isVisible()
        assert content.plain_text == "Preserved status"
        x, y = curve.get_data()
        assert list(x) == [1, 2, 3]
        assert list(y) == [4, 5, 6]
        assert dock_area.widget_map() == widgets
        assert dock_area.bec_dispatcher._registered_slots == slots
        for obj in objects[1:]:
            assert rpc_server.rpc_register.get_rpc_by_id(obj.gui_id) is obj
        assert all(spy.count() == 0 for spy in destroyed)

    # The same widgets can still receive RPC updates after recreating their native resources.
    rpc_server.run_rpc(content, "set_plain_text", ["Updated status"], {})
    assert content.plain_text == "Updated status"
    curve.set_data([7, 8], [9, 10])
    x, y = curve.get_data()
    assert list(x) == [7, 8]
    assert list(y) == [9, 10]


def test_wayland_recreation_preserves_native_child_widget(rpc_server, qtbot, monkeypatch, qapp):
    window = QWidget()
    qtbot.addWidget(window)
    child = QWidget(window)
    child.setAttribute(Qt.WA_NativeWindow)
    QVBoxLayout(window).addWidget(child)
    window.show()
    qapp.processEvents()
    native_visible = child.windowHandle().isVisible()
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: "wayland")
    monkeypatch.setattr(window, "isActiveWindow", lambda: False)
    destroyed = QSignalSpy(child.destroyed)

    for _ in range(3):
        rpc_server.run_rpc(window, "raise", [], {})
        qapp.processEvents()
        assert child.parentWidget() is window
        assert child.isVisible()
        # Some QPA plugins (including offscreen) do not map native child windows.
        assert child.windowHandle().isVisible() == native_visible
        assert child.windowHandle().parent() is window.windowHandle()
        assert destroyed.count() == 0


def test_wayland_recreation_preserves_web_page_state(
    rpc_server, qtbot, monkeypatch, qapp, tmp_path
):
    monkeypatch.setenv("BECWIDGETS_PROFILE_DIR", str(tmp_path))
    dock_area = rpc_server._launch_dock_area(name="web_probe", startup_profile="skip")
    window = dock_area.window()
    qtbot.addWidget(window)
    content = dock_area.new("WebsiteWidget", object_name="web_page")
    view = content.website
    page = view.page()
    with qtbot.waitSignal(page.loadFinished, timeout=10000):
        view.setHtml(
            "<html><body>Preserved page<script>window.probeCounter = 40;</script></body></html>"
        )
    reloads = QSignalSpy(page.loadStarted)
    destroyed = QSignalSpy(page.destroyed)
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: "wayland")
    monkeypatch.setattr(window, "isActiveWindow", lambda: False)

    for expected in (41, 42, 43):
        rpc_server.run_rpc(window, "raise", [], {})
        qapp.processEvents()
        assert content.website is view
        assert view.page() is page
        result = []
        page.runJavaScript("window.probeCounter += 1", result.append)
        qtbot.waitUntil(lambda: bool(result))
        assert result == [expected]
        assert reloads.count() == destroyed.count() == 0


@pytest.mark.parametrize("stay_on_top", [False, True])
def test_xcb_raise_preserves_workaround_and_original_flags(
    rpc_server, dummy_widget, monkeypatch, stay_on_top
):
    # The actual QPA backend, not the desktop session or an environment override, selects the path.
    monkeypatch.setenv("XDG_SESSION_TYPE", "wayland")
    monkeypatch.setenv("QT_QPA_PLATFORM", "wayland")
    monkeypatch.setattr(rpc_server_module.QApplication, "platformName", lambda: "xcb")
    monkeypatch.setattr(dummy_widget, "isActiveWindow", lambda: False)
    dummy_widget.setWindowFlag(Qt.WindowStaysOnTopHint, stay_on_top)
    dummy_widget.show()
    original_flags = dummy_widget.windowFlags()

    with (
        patch.object(dummy_widget, "setWindowFlags", wraps=dummy_widget.setWindowFlags) as flags,
        patch.object(dummy_widget, "activateWindow") as activate,
    ):
        for _ in range(3):
            rpc_server.run_rpc(dummy_widget, "raise", [], {})
            assert dummy_widget.isVisible()
            assert dummy_widget.windowFlags() == original_flags
    assert flags.call_count == 6
    assert flags.call_args_list[0].args == (original_flags | Qt.WindowStaysOnTopHint,)
    assert flags.call_args_list[1].args == (original_flags,)
    assert activate.call_count == 3


def test_display_info_is_available_through_system_rpc(rpc_server, qapp):
    info = rpc_server.run_system_rpc("system.get_display_info", [], {})
    assert info["qt_platform"] == qapp.platformName()


def test_rpc_server_system_shutdown_requests_gui_server_shutdown(rpc_server, qapp):
    gui_server = MagicMock()
    qapp.gui_server = gui_server

    rpc_server.run_system_rpc("system.shutdown", [], {})
    qapp.processEvents()

    gui_server.request_shutdown.assert_called_once()
    del qapp.gui_server


def test_on_rpc_update_system_shutdown_sends_response_before_return(rpc_server):
    order = []
    rpc_server.run_system_rpc = MagicMock(side_effect=lambda *_args: order.append("shutdown"))
    rpc_server.send_response = MagicMock(side_effect=lambda *_args: order.append("response"))
    rpc_server.serialize_result_and_send = MagicMock()

    rpc_server.on_rpc_update(
        {"action": "system.shutdown", "parameter": {"args": [], "kwargs": {}}},
        {"request_id": "shutdown-request", "sent_at": 1.0, "deadline": 10.0, "timeout": 2},
    )

    assert order == ["shutdown", "response"]
    rpc_server.send_response.assert_called_once_with("shutdown-request", True, {"result": None})
    rpc_server.serialize_result_and_send.assert_not_called()


def test_singleshot_rpc_repeat_raises_on_repeated_singleshot(rpc_server):
    """
    Test that a singleshot RPC method raises an error when called multiple times.
    """
    repeat = SingleshotRPCRepeat()
    rpc_server._rpc_singleshot_repeats["test_method"] = repeat

    repeat += 100  # First call should work fine
    with pytest.raises(RegistryNotReadyError):
        repeat += 2000  # Should raise here


def test_serialize_result_and_send_with_singleshot_retry(rpc_server, qtbot, dummy_widget):
    """
    Test that serialize_result_and_send retries when RegistryNotReadyError is raised,
    and eventually succeeds when the object becomes registered.
    """
    request_id = "test_request_123"

    dummy = dummy_widget

    # Track how many times serialize_object is called
    call_count = 0

    def serialize_side_effect(obj):
        nonlocal call_count
        call_count += 1
        # First 2 calls raise RegistryNotReadyError
        if call_count <= 2:
            raise RegistryNotReadyError(f"Not ready yet (call {call_count})")
        # Third call succeeds
        return {"gui_id": dummy.gui_id, "success": True}

    warning_mock = MagicMock()

    # Patch serialize_object to control when it raises RegistryNotReadyError
    with patch.object(rpc_server, "serialize_object", side_effect=serialize_side_effect):
        with patch.object(rpc_server, "send_response") as mock_send_response:
            with patch.object(rpc_server_module.logger, "warning", warning_mock):
                # Start the serialization process
                rpc_server._rpc_singleshot_repeats[request_id] = SingleshotRPCRepeat()
                rpc_server.serialize_result_and_send(request_id, dummy)

                # Verify that serialize_object was called 3 times
                qtbot.waitUntil(lambda: call_count >= 3, timeout=5000)

                # Verify that send_response was called with success
                mock_send_response.assert_called_once()
                args = mock_send_response.call_args[0]
                assert args[0] == request_id
                assert args[1] is True  # accepted=True
                assert "result" in args[2]

    assert warning_mock.call_count == 2
    warning_logs = "\n".join(call.args[0] for call in warning_mock.call_args_list)
    assert "result serialization delayed; retrying" in warning_logs
    assert "request_id=test_request_123" in warning_logs
    assert "retry_delay_ms=100" in warning_logs
    assert "accumulated_delay_ms=100" in warning_logs
    assert "accumulated_delay_ms=200" in warning_logs
    assert "max_delay_ms=2000" in warning_logs


def test_serialize_result_and_send_max_delay_exceeded(rpc_server, qtbot, dummy_widget):
    """
    Test that serialize_result_and_send sends an error response when max delay is exceeded.
    """
    request_id = "test_request_456"

    dummy = dummy_widget

    # Always raise RegistryNotReadyError
    with patch.object(
        rpc_server, "serialize_object", side_effect=RegistryNotReadyError("Always not ready")
    ):
        with patch.object(rpc_server, "send_response") as mock_send_response:
            # Start the serialization process
            rpc_server._rpc_singleshot_repeats[request_id] = SingleshotRPCRepeat()
            rpc_server.serialize_result_and_send(request_id, dummy)

            # Process event loop to allow all singleshot timers to fire
            # Max delay is 2000ms, with 100ms retry intervals = ~20 retries
            # Wait for the max delay plus some buffer
            qtbot.wait(2500)

            # Verify that send_response was called with an error
            mock_send_response.assert_called()
            args = mock_send_response.call_args[0]
            assert args[0] == request_id
            assert args[1] is False  # accepted=False
            assert "error" in args[2]
            assert "Max delay exceeded" in args[2]["error"]


def test_send_response_logs_publish_status(rpc_server, monkeypatch):
    info_mock = MagicMock()
    error_mock = MagicMock()
    monkeypatch.setattr(rpc_server_module.logger, "info", info_mock)
    monkeypatch.setattr(rpc_server_module.logger, "error", error_mock)

    with patch.object(rpc_server.client.connector, "set_and_publish") as publish_mock:
        rpc_server.send_response("request-ok", True, {"result": None})
        rpc_server.send_response("request-failed", False, {"error": "bad"})

    assert publish_mock.call_count == 2
    assert "request_id=request-ok" in info_mock.call_args.args[0]
    assert "accepted=True" in info_mock.call_args.args[0]
    assert "request_id=request-failed" in error_mock.call_args.args[0]
    assert "accepted=False" in error_mock.call_args.args[0]


def test_on_rpc_update_logs_late_client_deadline(rpc_server, monkeypatch):
    info_mock = MagicMock()
    warning_mock = MagicMock()
    monkeypatch.setattr(rpc_server_module.logger, "info", info_mock)
    monkeypatch.setattr(rpc_server_module.logger, "warning", warning_mock)

    rpc_server.rpc_register.get_rpc_by_id = MagicMock()
    rpc_server.run_rpc = MagicMock(return_value=None)
    rpc_server.serialize_result_and_send = MagicMock()

    rpc_server.on_rpc_update(
        {
            "action": "set_value",
            "parameter": {"args": [1], "kwargs": {"source": "test"}, "gui_id": "ring"},
        },
        {"request_id": "late-request", "timeout": 0.1, "sent_at": 1.0, "deadline": 1.1},
    )

    received_log = info_mock.call_args_list[0].args[0]
    executed_log = info_mock.call_args_list[1].args[0]
    warning_logs = "\n".join(call.args[0] for call in warning_mock.call_args_list)

    assert "GUI RPC server received request" in received_log
    assert "request_id=late-request" in received_log
    assert "method=set_value" in received_log
    assert "target_gui_id=ring" in received_log
    assert "timeout=0.1" in received_log
    assert "stale_on_receive=True" in received_log
    assert "response_after_client_deadline=True" in executed_log
    assert "received request after client timeout deadline" in warning_logs
    assert "response is late for client timeout" in warning_logs


def test_run_rpc_delegates_to_rpc_content_class(rpc_server):
    class Content:
        USER_ACCESS = ["foo", "mode", "mode.setter"]

        def __init__(self):
            self._mode = "initial"

        def foo(self):
            return "ok"

        @property
        def mode(self):
            return self._mode

        @mode.setter
        def mode(self, value):
            self._mode = value

    class View:
        RPC_CONTENT_CLASS = Content
        RPC_CONTENT_ATTR = "content"

        def __init__(self):
            self.content = Content()

    view = View()

    assert rpc_server.run_rpc(view, "foo", [], {}) == "ok"
    assert rpc_server.run_rpc(view, "mode", [], {}) == "initial"
    assert rpc_server.run_rpc(view, "mode", ["creator"], {}) is None
    assert view.content.mode == "creator"


def test_rpc_server_shutdown_releases_registrations(mocked_client):
    """Regression test for BW-008/BW-009: shutdown must disconnect the
    gui_instructions dispatcher slot and remove the registry callback so a
    restarted server does not duplicate registrations and the old server can
    be collected."""
    from bec_widgets.utils.rpc_register import RPCRegister

    register = RPCRegister()
    callbacks_before = len(register.callbacks)

    server = RPCServer(gui_id="lifecycle_gui", client=mocked_client)
    assert len(register.callbacks) == callbacks_before + 1

    dispatcher_slots_with_topic = [
        slot
        for slot in server.dispatcher._registered_slots.values()
        if any("lifecycle_gui" in topic for topic in slot.topics)
    ]
    assert len(dispatcher_slots_with_topic) == 1

    server.shutdown()

    assert len(register.callbacks) == callbacks_before
    dispatcher_slots_with_topic = [
        slot
        for slot in server.dispatcher._registered_slots.values()
        if any("lifecycle_gui" in topic for topic in slot.topics)
    ]
    assert dispatcher_slots_with_topic == []

    # Shutdown must be idempotent.
    server.shutdown()
    assert len(register.callbacks) == callbacks_before


def test_rpc_register_remove_callback_is_noop_for_unknown(rpc_register=None):
    from bec_widgets.utils.rpc_register import RPCRegister

    register = RPCRegister()
    register.remove_callback(lambda connections: None)  # must not raise


def test_rpc_server_shutdown_does_not_shut_down_client(mocked_client):
    """RPCServer.shutdown must not tear down the shared client: doing so nulls the
    connector's listener thread and turns a later disconnect_all() into a silent no-op."""
    from unittest.mock import patch

    server = RPCServer(gui_id="no_client_teardown", client=mocked_client)
    with patch.object(mocked_client, "shutdown") as client_shutdown:
        server.shutdown()
    client_shutdown.assert_not_called()
    # idempotent + still no client shutdown
    with patch.object(mocked_client, "shutdown") as client_shutdown2:
        server.shutdown()
    client_shutdown2.assert_not_called()


def test_gui_server_stop_dispatcher_disconnects_before_stopping_cli_server(gui_server):
    """disconnect_all() must run before stop_cli_server() so subscriptions are released
    while the connector's listener thread is still alive."""
    from unittest.mock import MagicMock, patch

    calls = []
    disp = MagicMock()
    disp.disconnect_all.side_effect = lambda *a, **k: calls.append("disconnect_all")
    disp.stop_cli_server.side_effect = lambda *a, **k: calls.append("stop_cli_server")
    gui_server.dispatcher = disp
    gui_server.launcher_window = MagicMock()
    with (
        patch.object(companion_app_module.shiboken6, "isValid", return_value=True),
        patch.object(companion_app_module.pylsp_server, "is_running", return_value=False),
    ):
        gui_server.shutdown()
    assert calls == ["disconnect_all", "stop_cli_server"]