fix(bec_console): clear the console registry when the application quits

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:

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)