From 0708b6cf63d8e6a381dd5c83f8bf1531eaabfff4 Mon Sep 17 00:00:00 2001 From: wyzula-jan Date: Sat, 3 Oct 2026 16:31:59 +0200 Subject: [PATCH] fix(pylsp): keep Ctrl-C and SIGTERM working after qtmonaco is imported --- bec_widgets/applications/companion_app.py | 6 +- bec_widgets/applications/main_app.py | 39 +++ bec_widgets/utils/pylsp_server.py | 96 ++++++ .../widgets/editors/monaco/monaco_widget.py | 4 + tests/unit_tests/test_pylsp_server.py | 200 +++++++++++++ tests/unit_tests/test_rpc_server.py | 4 +- tests/unit_tests/test_signal_termination.py | 273 ++++++++++++++++++ 7 files changed, 615 insertions(+), 7 deletions(-) create mode 100644 bec_widgets/utils/pylsp_server.py create mode 100644 tests/unit_tests/test_pylsp_server.py create mode 100644 tests/unit_tests/test_signal_termination.py diff --git a/bec_widgets/applications/companion_app.py b/bec_widgets/applications/companion_app.py index 977f4c09..46ef7f04 100644 --- a/bec_widgets/applications/companion_app.py +++ b/bec_widgets/applications/companion_app.py @@ -19,7 +19,6 @@ import shiboken6 from bec_lib.logger import bec_logger from bec_lib.service_config import ServiceConfig from bec_qthemes import apply_theme -from qtmonaco.pylsp_provider import pylsp_server from qtpy.QtCore import QSize, Qt, QTimer from qtpy.QtGui import QIcon from qtpy.QtWidgets import QApplication @@ -27,6 +26,7 @@ from qtpy.QtWidgets import QApplication import bec_widgets from bec_widgets.applications.launch_window import LaunchWindow from bec_widgets.utils.bec_dispatcher import BECDispatcher +from bec_widgets.utils.pylsp_server import stop_pylsp_server from bec_widgets.utils.rpc_register import RPCRegister # isort: on @@ -223,10 +223,6 @@ class GUIServer: self.launcher_window.close() self.launcher_window.deleteLater() - def stop_pylsp_server(): - if pylsp_server.is_running(): - pylsp_server.stop() - def stop_dispatcher(): if self.dispatcher: # disconnect_all() must run BEFORE stop_cli_server(). stop_cli_server() diff --git a/bec_widgets/applications/main_app.py b/bec_widgets/applications/main_app.py index e238599d..980e120b 100644 --- a/bec_widgets/applications/main_app.py +++ b/bec_widgets/applications/main_app.py @@ -1,5 +1,7 @@ from bec_widgets.applications.startup_profiler import startup_profiler # isort: skip +import signal + from bec_qthemes import material_icon from qtpy.QtCore import QTimer from qtpy.QtGui import QAction # type: ignore @@ -17,6 +19,7 @@ from bec_widgets.utils.colors import apply_theme from bec_widgets.utils.guided_tour import GuidedTour from bec_widgets.utils.launcher_ready import notify_launcher_ready from bec_widgets.utils.name_utils import sanitize_namespace +from bec_widgets.utils.pylsp_server import stop_pylsp_server from bec_widgets.utils.screen_utils import ( apply_centered_size, available_screen_geometry, @@ -399,6 +402,40 @@ class BECMainApp(BECMainWindow): super().cleanup() +def install_shutdown_signal_handlers(app: QApplication) -> None: + """ + Close the application gracefully on SIGINT (Ctrl-C) and SIGTERM. + + Without Python handlers SIGTERM kills the process before any cleanup runs, and the + KeyboardInterrupt raised for SIGINT is swallowed by whichever Qt slot it lands in, so Ctrl-C + never stops a Qt application. The handler only schedules the shutdown; the event loop then + closes all top-level windows (running the widgets' cleanup) and quits. A second signal while + that shutdown is still pending terminates the process right away. + + Args: + app(QApplication): The application to quit; install after its main window exists. + """ + shutdown_requested = False + + def _shutdown(): + for widget in app.topLevelWidgets(): + widget.close() + app.quit() + + def _request_shutdown(signum, _frame): + nonlocal shutdown_requested + if shutdown_requested: + stop_pylsp_server() + signal.signal(signum, signal.SIG_DFL) + signal.raise_signal(signum) + return + shutdown_requested = True + QTimer.singleShot(0, _shutdown) + + signal.signal(signal.SIGINT, _request_shutdown) + signal.signal(signal.SIGTERM, _request_shutdown) + + def main(): # pragma: no cover """ Main function to run the BEC main application, exposed as a script entry point through @@ -436,6 +473,8 @@ def main(): # pragma: no cover # First event-loop iteration -> the window is actually painted/interactive. QTimer.singleShot(0, lambda: startup_profiler.mark("interactive", final=True)) + app.aboutToQuit.connect(stop_pylsp_server) + install_shutdown_signal_handlers(app) sys.exit(app.exec()) diff --git a/bec_widgets/utils/pylsp_server.py b/bec_widgets/utils/pylsp_server.py new file mode 100644 index 00000000..83702b8c --- /dev/null +++ b/bec_widgets/utils/pylsp_server.py @@ -0,0 +1,96 @@ +""" +Signal-safe access to qtmonaco's shared PyLSP language server. + +``qtmonaco.pylsp_provider`` creates its ``pylsp_server`` singleton at import time, and the +singleton's constructor replaces the process's SIGINT and SIGTERM handlers with one that only +stops the language server and returns. A process that imports the module (directly, or by +creating its first ``qtmonaco.Monaco`` editor) therefore ignores Ctrl-C and ``kill`` for the rest +of its life. Import the provider only through :func:`get_pylsp_server`, and stop the server with +:func:`stop_pylsp_server`, which never imports qtmonaco just to find out that nothing runs. +""" + +from __future__ import annotations + +import importlib +import signal +import sys +import threading +from typing import TYPE_CHECKING + +if TYPE_CHECKING: # pragma: no cover + from qtmonaco.pylsp_provider import PyLSPProvider + +PYLSP_PROVIDER_MODULE = "qtmonaco.pylsp_provider" + +# The signals qtmonaco takes over, with the disposition a Python process starts with. +_DEFAULT_HANDLERS = {signal.SIGINT: signal.default_int_handler, signal.SIGTERM: signal.SIG_DFL} + + +def get_pylsp_server() -> PyLSPProvider: + """ + Return qtmonaco's shared PyLSP provider without letting it take over SIGINT and SIGTERM. + + The provider module is imported on the first call; right after the import, the handlers it + installed are replaced by the ones the process had before. A signal whose previous + disposition was ``SIG_DFL`` (SIGTERM, usually) gets a handler that stops the language server + and then terminates the process exactly like ``SIG_DFL`` would have, so a ``kill`` does not + leave an orphaned pylsp process behind. Python handlers (``default_int_handler``, the GUI + server's ``request_shutdown``, ...) and ``SIG_IGN`` are restored as they were; their exit + paths stop the server through ``atexit`` or :func:`stop_pylsp_server`. + + Call it from the main thread before creating a ``qtmonaco.Monaco`` editor. If another + component imported the provider first, the handlers it left behind are reset to the + interpreter defaults (with the same stop-then-terminate treatment for SIGTERM). + + Returns: + PyLSPProvider: The process-wide provider (``qtmonaco.pylsp_provider.pylsp_server``). + """ + module = sys.modules.get(PYLSP_PROVIDER_MODULE) + previous = {} + if module is None: + previous = {signum: signal.getsignal(signum) for signum in _DEFAULT_HANDLERS} + module = importlib.import_module(PYLSP_PROVIDER_MODULE) + server = module.pylsp_server + _replace_provider_signal_handlers(server, previous) + return server + + +def stop_pylsp_server() -> None: + """ + Stop qtmonaco's PyLSP server if this process started it. + + Does nothing (and imports nothing) when no Monaco editor was ever created in this process. + """ + module = sys.modules.get(PYLSP_PROVIDER_MODULE) + if module is None: + return + server = module.pylsp_server + if server.is_running(): + server.stop() + + +def _replace_provider_signal_handlers(server: PyLSPProvider, previous: dict) -> None: + """Swap the provider's own SIGINT/SIGTERM handlers for the process's previous ones.""" + if threading.current_thread() is not threading.main_thread(): + # signal.signal() only works in the main thread; qtmonaco's own import fails there too. + return + # pylint: disable=protected-access + provider_handler = server._handle_signal + for signum, default in _DEFAULT_HANDLERS.items(): + if signal.getsignal(signum) != provider_handler: + continue # already replaced, e.g. by the GUI server's own shutdown handler + handler = previous.get(signum) + if handler is None: # not installed from Python, or the provider was imported elsewhere + handler = default + if handler is signal.SIG_DFL: + handler = _stop_server_then_terminate + signal.signal(signum, handler) + + +def _stop_server_then_terminate(signum: int, _frame) -> None: + """Stop the language server, then let the signal terminate the process like SIG_DFL.""" + try: + stop_pylsp_server() + finally: + signal.signal(signum, signal.SIG_DFL) + signal.raise_signal(signum) diff --git a/bec_widgets/widgets/editors/monaco/monaco_widget.py b/bec_widgets/widgets/editors/monaco/monaco_widget.py index f2cbe2c5..0487a926 100644 --- a/bec_widgets/widgets/editors/monaco/monaco_widget.py +++ b/bec_widgets/widgets/editors/monaco/monaco_widget.py @@ -14,6 +14,7 @@ from qtpy.QtWidgets import QApplication, QDialog, QVBoxLayout, QWidget from bec_widgets.utils.bec_widget import BECWidget from bec_widgets.utils.colors import get_theme_name from bec_widgets.utils.error_popups import SafeSlot +from bec_widgets.utils.pylsp_server import get_pylsp_server if TYPE_CHECKING: from bec_widgets.widgets.editors.monaco.scan_control_dialog import ScanControlDialog @@ -58,6 +59,9 @@ class MonacoWidget(BECWidget, QWidget): super().__init__(parent=parent, client=client, gui_id=gui_id, config=config, **kwargs) layout = QVBoxLayout() layout.setContentsMargins(0, 0, 0, 0) + # qtmonaco.Monaco imports qtmonaco.pylsp_provider, whose import takes over SIGINT/SIGTERM; + # importing it here first keeps the process's own handlers (Ctrl-C and kill keep working). + get_pylsp_server() self.editor = qtmonaco.Monaco(self) layout.addWidget(self.editor) self.setLayout(layout) diff --git a/tests/unit_tests/test_pylsp_server.py b/tests/unit_tests/test_pylsp_server.py new file mode 100644 index 00000000..a70fb702 --- /dev/null +++ b/tests/unit_tests/test_pylsp_server.py @@ -0,0 +1,200 @@ +"""Unit tests for bec_widgets.utils.pylsp_server and bec-app's shutdown signal handlers.""" + +from __future__ import annotations + +import signal +import sys +import threading +import types +from unittest import mock + +import pytest + +from bec_widgets.applications import main_app +from bec_widgets.utils import pylsp_server as pylsp_module + +SIGNALS = (signal.SIGINT, signal.SIGTERM) + + +class FakeProvider: + """Stand-in for qtmonaco's PyLSPProvider: takes over SIGINT/SIGTERM when constructed.""" + + def __init__(self, install_handlers: bool = True): + self.running = False + self.stop_calls = 0 + if install_handlers: + for signum in SIGNALS: + signal.signal(signum, self._handle_signal) + + def _handle_signal(self, signum, frame): + self.stop() + + def is_running(self): + return self.running + + def stop(self): + self.stop_calls += 1 + self.running = False + + +@pytest.fixture +def keep_signal_handlers(): + """Restore the test process's SIGINT/SIGTERM handlers whatever a test installs.""" + saved = {signum: signal.getsignal(signum) for signum in SIGNALS} + yield + for signum, handler in saved.items(): + signal.signal(signum, handler) + + +@pytest.fixture +def fake_provider_import(monkeypatch, keep_signal_handlers): + """Make get_pylsp_server import a fake provider module that hijacks the handlers.""" + monkeypatch.delitem(sys.modules, pylsp_module.PYLSP_PROVIDER_MODULE, raising=False) + created = {} + + def import_module(name): + assert name == pylsp_module.PYLSP_PROVIDER_MODULE + module = types.ModuleType(name) + module.pylsp_server = created["server"] = FakeProvider() + sys.modules[name] = module + return module + + monkeypatch.setattr(pylsp_module.importlib, "import_module", import_module) + yield created + sys.modules.pop(pylsp_module.PYLSP_PROVIDER_MODULE, None) + + +def test_get_pylsp_server_restores_python_handlers(fake_provider_import): + def custom_handler(signum, frame): # pragma: no cover - never delivered + pass + + signal.signal(signal.SIGINT, signal.default_int_handler) + signal.signal(signal.SIGTERM, custom_handler) + + server = pylsp_module.get_pylsp_server() + + assert server is fake_provider_import["server"] + assert signal.getsignal(signal.SIGINT) is signal.default_int_handler + assert signal.getsignal(signal.SIGTERM) is custom_handler + + +def test_get_pylsp_server_keeps_sig_dfl_terminating(fake_provider_import): + signal.signal(signal.SIGINT, signal.SIG_IGN) + signal.signal(signal.SIGTERM, signal.SIG_DFL) + + pylsp_module.get_pylsp_server() + + assert signal.getsignal(signal.SIGINT) is signal.SIG_IGN + # SIG_DFL would kill the process without stopping the server: stop it first, then terminate + assert signal.getsignal(signal.SIGTERM) is pylsp_module._stop_server_then_terminate + + +def test_get_pylsp_server_is_idempotent_and_leaves_later_handlers_alone(fake_provider_import): + def request_shutdown(signum, frame): # pragma: no cover - never delivered + pass + + first = pylsp_module.get_pylsp_server() + signal.signal(signal.SIGTERM, request_shutdown) # e.g. GUIServer.start() + + assert pylsp_module.get_pylsp_server() is first + assert signal.getsignal(signal.SIGTERM) is request_shutdown + + +def test_get_pylsp_server_resets_handlers_left_by_an_earlier_import( + monkeypatch, keep_signal_handlers +): + module = types.ModuleType(pylsp_module.PYLSP_PROVIDER_MODULE) + module.pylsp_server = FakeProvider() # imported elsewhere, handlers already hijacked + monkeypatch.setitem(sys.modules, pylsp_module.PYLSP_PROVIDER_MODULE, module) + + assert pylsp_module.get_pylsp_server() is module.pylsp_server + assert signal.getsignal(signal.SIGINT) is signal.default_int_handler + assert signal.getsignal(signal.SIGTERM) is pylsp_module._stop_server_then_terminate + + +def test_get_pylsp_server_outside_main_thread_does_not_touch_signals( + monkeypatch, keep_signal_handlers +): + module = types.ModuleType(pylsp_module.PYLSP_PROVIDER_MODULE) + module.pylsp_server = FakeProvider() + monkeypatch.setitem(sys.modules, pylsp_module.PYLSP_PROVIDER_MODULE, module) + result = {} + thread = threading.Thread(target=lambda: result.update(server=pylsp_module.get_pylsp_server())) + thread.start() + thread.join() + + assert result["server"] is module.pylsp_server + assert signal.getsignal(signal.SIGTERM) == module.pylsp_server._handle_signal + + +def test_stop_pylsp_server_never_imports_qtmonaco(monkeypatch): + monkeypatch.delitem(sys.modules, pylsp_module.PYLSP_PROVIDER_MODULE, raising=False) + with mock.patch.object(pylsp_module.importlib, "import_module") as import_module: + pylsp_module.stop_pylsp_server() + import_module.assert_not_called() + assert pylsp_module.PYLSP_PROVIDER_MODULE not in sys.modules + + +@pytest.mark.parametrize("running", [True, False]) +def test_stop_pylsp_server_stops_only_a_running_server(monkeypatch, running): + module = types.ModuleType(pylsp_module.PYLSP_PROVIDER_MODULE) + module.pylsp_server = FakeProvider(install_handlers=False) + module.pylsp_server.running = running + monkeypatch.setitem(sys.modules, pylsp_module.PYLSP_PROVIDER_MODULE, module) + + pylsp_module.stop_pylsp_server() + + assert module.pylsp_server.stop_calls == int(running) + + +def test_stop_server_then_terminate_reraises_with_default_action(monkeypatch): + calls = [] + monkeypatch.setattr(pylsp_module, "stop_pylsp_server", lambda: calls.append("stop")) + monkeypatch.setattr( + pylsp_module.signal, "signal", lambda signum, handler: calls.append((signum, handler)) + ) + monkeypatch.setattr(pylsp_module.signal, "raise_signal", lambda signum: calls.append(signum)) + + pylsp_module._stop_server_then_terminate(signal.SIGTERM, None) + + assert calls == ["stop", (signal.SIGTERM, signal.SIG_DFL), signal.SIGTERM] + + +@pytest.fixture +def captured_handlers(monkeypatch): + """Capture the handlers bec-app installs instead of installing them in the test process.""" + handlers = {} + monkeypatch.setattr( + main_app.signal, "signal", lambda signum, handler: handlers.__setitem__(signum, handler) + ) + return handlers + + +def test_bec_app_signal_handlers_close_windows_and_quit(qtbot, captured_handlers): + app = mock.MagicMock() + window = mock.MagicMock() + app.topLevelWidgets.return_value = [window] + + main_app.install_shutdown_signal_handlers(app) + assert set(captured_handlers) == {signal.SIGINT, signal.SIGTERM} + + captured_handlers[signal.SIGTERM](signal.SIGTERM, None) + app.quit.assert_not_called() # deferred to the event loop, not run inside the handler + qtbot.waitUntil(lambda: app.quit.called, timeout=1000) + window.close.assert_called_once() + + +def test_bec_app_second_signal_terminates_immediately(qtbot, captured_handlers, monkeypatch): + calls = [] + monkeypatch.setattr(main_app, "stop_pylsp_server", lambda: calls.append("stop")) + monkeypatch.setattr(main_app.signal, "raise_signal", lambda signum: calls.append(signum)) + app = mock.MagicMock() + app.topLevelWidgets.return_value = [] + + main_app.install_shutdown_signal_handlers(app) + captured_handlers[signal.SIGINT](signal.SIGINT, None) + captured_handlers[signal.SIGINT](signal.SIGINT, None) + + assert calls == ["stop", signal.SIGINT] + assert captured_handlers[signal.SIGINT] is signal.SIG_DFL + qtbot.waitUntil(lambda: app.quit.called, timeout=1000) diff --git a/tests/unit_tests/test_rpc_server.py b/tests/unit_tests/test_rpc_server.py index 001cd389..430eac82 100644 --- a/tests/unit_tests/test_rpc_server.py +++ b/tests/unit_tests/test_rpc_server.py @@ -76,7 +76,7 @@ def test_gui_server_shutdown_is_idempotent(gui_server): with ( patch.object(companion_app_module.shiboken6, "isValid", return_value=True), - patch.object(companion_app_module.pylsp_server, "is_running", return_value=False), + patch.object(companion_app_module, "stop_pylsp_server"), ): gui_server.shutdown() gui_server.shutdown() @@ -368,7 +368,7 @@ def test_gui_server_stop_dispatcher_disconnects_before_stopping_cli_server(gui_s 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), + patch.object(companion_app_module, "stop_pylsp_server"), ): gui_server.shutdown() assert calls == ["disconnect_all", "stop_cli_server"] diff --git a/tests/unit_tests/test_signal_termination.py b/tests/unit_tests/test_signal_termination.py new file mode 100644 index 00000000..89a1eea8 --- /dev/null +++ b/tests/unit_tests/test_signal_termination.py @@ -0,0 +1,273 @@ +""" +SIGINT (Ctrl-C) and SIGTERM must keep stopping BEC Widgets processes. + +``qtmonaco.pylsp_provider`` replaces both handlers at import time with one that only stops the +pylsp language server and returns, so a process that imported it (the GUI server module, the unit +suite at collection, any application with a Monaco editor such as bec-app's device manager) kept +running after Ctrl-C or ``kill``. Every scenario runs in a child process (offscreen, fake BEC +client on fakeredis) that reports READY, gets the signal and must exit within EXIT_TIMEOUT +without leaving its pylsp server behind. +""" + +from __future__ import annotations + +import json +import os +import signal +import subprocess +import sys +import textwrap +import time +from pathlib import Path + +import pytest + +READY_TIMEOUT = 90.0 +EXIT_TIMEOUT = 15.0 +REPO_ROOT = Path(__file__).resolve().parents[2] + +CHILD_BOOT = textwrap.dedent(r''' + import json, os, signal, sys, time + + def report(): + """Write the READY file: pid of the pylsp server (if any) and the current handlers.""" + provider = sys.modules.get("qtmonaco.pylsp_provider") + process = provider.pylsp_server.server_process if provider is not None else None + info = { + "pylsp_pid": process.pid if process is not None else None, + "sigint": repr(signal.getsignal(signal.SIGINT)), + "sigterm": repr(signal.getsignal(signal.SIGTERM)), + } + path = os.environ["SIGNAL_TEST_READY_FILE"] + with open(path + ".tmp", "w") as f: + json.dump(info, f) + os.replace(path + ".tmp", path) + + def report_from_event_loop(delay_ms=500): + from qtpy.QtCore import QTimer + QTimer.singleShot(delay_ms, report) + + def use_fake_bec(): + """Same fake BEC client as the packaged bec_dispatcher fixture (fakeredis, DMMock devices).""" + from unittest import mock + from bec_widgets.tests.utils import mock_client + from bec_widgets.utils import bec_dispatcher as dispatcher_module + + mock.patch.object(dispatcher_module, "BECClient", mock_client).start() + ''') + +# The unit suite collects test_rpc_server/test_companion_app, which import the GUI server module. +IMPORT_COMPANION_APP = """ +import bec_widgets.applications.companion_app # noqa: F401 +report() +time.sleep(120) +""" + +# A Python-driven process (like a test run) that created a Monaco editor; on KeyboardInterrupt +# it shuts the BEC client down like the bec_dispatcher fixture teardown does (its non-daemon threads would +# otherwise keep the interpreter alive). +MONACO_WIDGET_PYTHON_LOOP = """ +use_fake_bec() +from qtpy.QtWidgets import QApplication +app = QApplication([sys.argv[0]]) +from bec_widgets.utils.bec_dispatcher import BECDispatcher +from bec_widgets.widgets.editors.monaco.monaco_widget import MonacoWidget +editor = MonacoWidget() +report() +try: + time.sleep(120) +finally: + dispatcher = BECDispatcher() + dispatcher.disconnect_all() + dispatcher.client.shutdown() + dispatcher.stop_cli_server() +""" + +# A Qt application that created a Monaco editor (bec-app device manager, plugins, designer forms). +MONACO_WIDGET_QT_LOOP = """ +use_fake_bec() +from qtpy.QtWidgets import QApplication +app = QApplication([sys.argv[0]]) +from bec_widgets.widgets.editors.monaco.monaco_widget import MonacoWidget +editor = MonacoWidget() +editor.show() +report_from_event_loop() +sys.exit(app.exec()) +""" + +# bec-app as started by its console script (its device manager view holds a Monaco editor). +BEC_APP = """ +use_fake_bec() +from bec_widgets.applications import main_app +_mark = main_app.startup_profiler.mark +def mark(stage, **kwargs): + _mark(stage, **kwargs) + if stage == "interactive": + report_from_event_loop() +main_app.startup_profiler.mark = mark +sys.argv = [sys.argv[0]] +main_app.main() +""" + +# bec-gui-server; a Monaco editor is created after the server installed its shutdown handlers. +GUI_SERVER_WITH_MONACO = """ +use_fake_bec() +from bec_widgets.applications import companion_app +_ready = companion_app.GUIServer._notify_server_ready +editors = [] +def ready(self): + _ready(self) + from bec_widgets.widgets.editors.monaco.monaco_widget import MonacoWidget + editors.append(MonacoWidget()) + editors[-1].show() + report_from_event_loop() +companion_app.GUIServer._notify_server_ready = ready +sys.argv = ["bec-gui-server", "--id", "signal_test"] +companion_app.main() +""" + +# A pytest run whose collection imports the GUI server module (as tests/unit_tests does). +PYTEST_RUN_TEST_FILE = """ +import os, time +import bec_widgets.applications.companion_app # noqa: F401 + +def test_sleep(): + path = os.environ["SIGNAL_TEST_READY_FILE"] + with open(path + ".tmp", "w") as f: + f.write("{}") + os.replace(path + ".tmp", path) + time.sleep(120) +""" + + +def _pid_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: # pragma: no cover - pid reused by another user's process + return True + return True + + +def _child_env(tmp_path: Path) -> dict: + env = dict(os.environ) + env.update( + SIGNAL_TEST_READY_FILE=str(tmp_path / "ready.json"), + BECWIDGETS_PROFILE_DIR=str(tmp_path / "profiles"), + QT_QPA_PLATFORM="offscreen", + BEC_WIDGETS_OPENGL="0", + QTWEBENGINE_DISABLE_SANDBOX="1", + QTWEBENGINE_CHROMIUM_FLAGS="--disable-gpu", + OPHYD_CONTROL_LAYER="dummy", + ) + return env + + +def _signal_child(cmd: list[str], tmp_path: Path, sig: signal.Signals, cwd: Path) -> dict: + """ + Start ``cmd``, wait for its READY file, send ``sig`` and wait for the exit. + + Returns: + dict: ``returncode`` (None if the child survived EXIT_TIMEOUT and was SIGKILLed), + ``seconds`` until the exit, ``pylsp_leaked`` and the child's READY info. + """ + ready_file = tmp_path / "ready.json" + log_file = tmp_path / "child.log" + with open(log_file, "w") as log: + proc = subprocess.Popen( + cmd, cwd=cwd, env=_child_env(tmp_path), stdout=log, stderr=subprocess.STDOUT + ) + pylsp_pid = None + try: + deadline = time.monotonic() + READY_TIMEOUT + while not ready_file.exists(): + if proc.poll() is not None or time.monotonic() > deadline: + pytest.fail(f"child not ready (rc={proc.poll()}):\n{log_file.read_text()[-3000:]}") + time.sleep(0.1) + info = json.loads(ready_file.read_text()) + pylsp_pid = info.get("pylsp_pid") + start = time.monotonic() + proc.send_signal(sig) + try: + returncode = proc.wait(timeout=EXIT_TIMEOUT) + except subprocess.TimeoutExpired: + returncode = None + seconds = time.monotonic() - start + pylsp_leaked = False + if pylsp_pid is not None and returncode is not None: + time.sleep(0.5) + pylsp_leaked = _pid_alive(pylsp_pid) + return { + "returncode": returncode, + "seconds": seconds, + "pylsp_leaked": pylsp_leaked, + "info": info, + "log": log_file.read_text()[-3000:], + } + finally: + if proc.poll() is None: + proc.kill() + proc.wait() + if pylsp_pid is not None and _pid_alive(pylsp_pid): + os.kill(pylsp_pid, signal.SIGKILL) + + +def _python_child(script: str) -> list[str]: + return [sys.executable, "-c", CHILD_BOOT + textwrap.dedent(script)] + + +def _assert_exited(result: dict, sig: signal.Signals, expected_returncodes: tuple[int, ...]): + assert result["returncode"] is not None, ( + f"process ignored {sig.name} for {EXIT_TIMEOUT:.0f}s " + f"(handlers at READY: {result['info']})\n{result['log']}" + ) + assert result["returncode"] in expected_returncodes, result["log"] + assert not result["pylsp_leaked"], "pylsp language server outlived its process" + + +@pytest.mark.parametrize("sig", [signal.SIGTERM, signal.SIGINT], ids=lambda s: s.name) +def test_importing_companion_app_keeps_process_killable(tmp_path, sig): + result = _signal_child(_python_child(IMPORT_COMPANION_APP), tmp_path, sig, REPO_ROOT) + _assert_exited(result, sig, (-sig,)) + + +@pytest.mark.parametrize("sig", [signal.SIGTERM, signal.SIGINT], ids=lambda s: s.name) +def test_monaco_widget_keeps_python_process_killable(tmp_path, sig): + result = _signal_child(_python_child(MONACO_WIDGET_PYTHON_LOOP), tmp_path, sig, REPO_ROOT) + assert result["info"]["pylsp_pid"] is not None, "Monaco editor did not start pylsp" + _assert_exited(result, sig, (-sig,)) + + +def test_monaco_widget_qt_application_terminates_on_sigterm(tmp_path): + result = _signal_child( + _python_child(MONACO_WIDGET_QT_LOOP), tmp_path, signal.SIGTERM, REPO_ROOT + ) + assert result["info"]["pylsp_pid"] is not None, "Monaco editor did not start pylsp" + _assert_exited(result, signal.SIGTERM, (-signal.SIGTERM,)) + + +@pytest.mark.parametrize("sig", [signal.SIGTERM, signal.SIGINT], ids=lambda s: s.name) +def test_bec_app_quits_on_signal(tmp_path, sig): + result = _signal_child(_python_child(BEC_APP), tmp_path, sig, REPO_ROOT) + assert result["info"]["pylsp_pid"] is not None, "bec-app did not create a Monaco editor" + _assert_exited(result, sig, (0,)) + + +def test_gui_server_with_monaco_still_shuts_down_on_sigterm(tmp_path): + result = _signal_child( + _python_child(GUI_SERVER_WITH_MONACO), tmp_path, signal.SIGTERM, REPO_ROOT + ) + assert result["info"]["pylsp_pid"] is not None, "Monaco editor did not start pylsp" + _assert_exited(result, signal.SIGTERM, (0,)) + + +@pytest.mark.parametrize("sig", [signal.SIGTERM, signal.SIGINT], ids=lambda s: s.name) +def test_pytest_run_importing_companion_app_stops_on_signal(tmp_path, sig): + test_dir = tmp_path / "suite" + test_dir.mkdir() + (test_dir / "test_sleep.py").write_text(textwrap.dedent(PYTEST_RUN_TEST_FILE)) + cmd = [sys.executable, "-m", "pytest", "-p", "no:cacheprovider", "-q", "test_sleep.py"] + result = _signal_child(cmd, tmp_path, sig, test_dir) + # SIGINT: pytest reports the KeyboardInterrupt and exits with ExitCode.INTERRUPTED (2) + _assert_exited(result, sig, (-signal.SIGTERM,) if sig == signal.SIGTERM else (2,))