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}) 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( "Preserved page" ) 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"]