diff --git a/bec_widgets/widgets/editors/bec_console/bec_console.py b/bec_widgets/widgets/editors/bec_console/bec_console.py index 6a41677b..aa476744 100644 --- a/bec_widgets/widgets/editors/bec_console/bec_console.py +++ b/bec_widgets/widgets/editors/bec_console/bec_console.py @@ -3,11 +3,11 @@ from __future__ import annotations import enum from dataclasses import dataclass, field from uuid import uuid4 -from weakref import WeakValueDictionary +from weakref import WeakValueDictionary, ref import shiboken6 from bec_lib.logger import bec_logger -from qtpy.QtCore import Qt, Signal +from qtpy.QtCore import QCoreApplication, Qt, Signal from qtpy.QtGui import QMouseEvent from qtpy.QtWidgets import ( QApplication, @@ -65,16 +65,27 @@ class BecConsoleRegistry: """ self._consoles: WeakValueDictionary[str, BecConsole] = WeakValueDictionary() self._terminal_registry: dict[str, _TerminalOwnerInfo] = {} + self._cleanup_app: ref[QCoreApplication] | None = None @staticmethod def _is_valid_qobject(obj: object | None) -> bool: return obj is not None and shiboken6.isValid(obj) def _connect_app_cleanup(self) -> None: + """ + Clear the registry when the running application is about to quit. + + The connection is made at most once per application instance. ``UniqueConnection`` + cannot be used for that: Qt only supports it for slots of a QObject, and the registry is a + plain Python object, so such a connection is silently rejected. + """ app = QApplication.instance() if app is None: return - app.aboutToQuit.connect(self.clear, Qt.ConnectionType.UniqueConnection) + if self._cleanup_app is not None and self._cleanup_app() is app: + return + app.aboutToQuit.connect(self.clear) + self._cleanup_app = ref(app) @staticmethod def _apply_zoom_step(term: BecTerminal, direction: int) -> bool: diff --git a/tests/unit_tests/test_bec_console.py b/tests/unit_tests/test_bec_console.py index 6bfd5df7..74ba99b2 100644 --- a/tests/unit_tests/test_bec_console.py +++ b/tests/unit_tests/test_bec_console.py @@ -2,7 +2,7 @@ from unittest import mock import pytest import shiboken6 -from qtpy.QtCore import QEvent, QEventLoop, Qt +from qtpy.QtCore import SIGNAL, QEvent, QEventLoop, QObject, Qt, Signal from qtpy.QtGui import QHideEvent, QShowEvent from qtpy.QtTest import QTest from qtpy.QtWidgets import QApplication, QWidget @@ -120,6 +120,89 @@ def test_bec_console_registry_cleanup(console_widget: BecConsole): assert terminal_id not in _bec_console_registry._terminal_registry +class _FakeApp(QObject): + """Stand-in for the QApplication: its aboutToQuit can be emitted without quitting the + real test application, whose other aboutToQuit handlers would tear down shared state.""" + + aboutToQuit = Signal() + + +class _StubConsole: + """Minimal console exposing only what BecConsoleRegistry.register reads.""" + + def __init__(self, console_id: str, terminal_id: str): + self.console_id = console_id + self.terminal_id = terminal_id + self.persist_terminal_session = False + self.default_zoom_level = 0 + + +def _about_to_quit_receivers(app: QObject) -> int: + return app.receivers(SIGNAL("aboutToQuit()")) + + +def _patch_app_instance(app: QObject): + """Make the bec_console module (only) see ``app`` as the running application.""" + return mock.patch.object(bec_console_module, "QApplication", **{"instance.return_value": app}) + + +def test_bec_console_registry_clears_terminals_on_about_to_quit(qtbot): + fake_app = _FakeApp() + registry = bec_console_module.BecConsoleRegistry() + consoles = [ + _StubConsole("quit_console_1", "quit_t1"), + _StubConsole("quit_console_2", "quit_t2"), + ] + try: + with _patch_app_instance(fake_app): + for console in consoles: + registry.register(console) + terminals = [registry.get_terminal("quit_t1"), registry.get_terminal("quit_t2")] + assert all(term is not None for term in terminals) + + fake_app.aboutToQuit.emit() + + assert registry._terminal_registry == {} + process_deferred_deletes() + assert not any(shiboken6.isValid(term) for term in terminals) + # registering twice must not wire the cleanup twice + assert _about_to_quit_receivers(fake_app) == 1 + finally: + registry.clear() + process_deferred_deletes() + fake_app.deleteLater() + + +def test_bec_console_registry_reconnects_cleanup_for_new_app(qtbot): + """The once-only guard is per application: a new application instance gets wired too.""" + registry = bec_console_module.BecConsoleRegistry() + old_app, new_app = _FakeApp(), _FakeApp() + try: + for app in (old_app, new_app): + with _patch_app_instance(app): + registry._connect_app_cleanup() + registry._connect_app_cleanup() + assert _about_to_quit_receivers(app) == 1 + finally: + old_app.deleteLater() + new_app.deleteLater() + process_deferred_deletes() + + +def test_bec_console_registry_cleanup_connected_once_to_real_app(qtbot): + app = QApplication.instance() + registry = bec_console_module.BecConsoleRegistry() + before = _about_to_quit_receivers(app) + try: + registry._connect_app_cleanup() + registry._connect_app_cleanup() + assert _about_to_quit_receivers(app) == before + 1 + finally: + # never emit the real aboutToQuit here; just drop our connection again + if _about_to_quit_receivers(app) > before: + app.aboutToQuit.disconnect(registry.clear) + + def test_bec_shell_initialization(qtbot): widget = BECShell(gui_id="bec_shell") qtbot.addWidget(widget)