fix(FPS): qtimer cleanup leaking

bec_widgets/widgets/figure/plots/plot_base.py

@@ -452,13 +452,14 @@ class BECPlotBase(BECConnector, pg.GraphicsLayout):
 
             self.fps_monitor.sigFpsUpdate.connect(self.update_fps_label)
 
-    def unhook_fps_monitor(self):
+    def unhook_fps_monitor(self, delete_label=True):
         """Unhook the FPS monitor from the plot."""
         if self.fps_monitor is not None:
             # Remove Monitor
             self.fps_monitor.cleanup()
             self.fps_monitor.deleteLater()
             self.fps_monitor = None
+        if self.fps_label is not None and delete_label:
             # Remove Label
             self.removeItem(self.fps_label)
             self.fps_label.deleteLater()
@@ -490,6 +491,7 @@ class BECPlotBase(BECConnector, pg.GraphicsLayout):
     def cleanup_pyqtgraph(self):
         """Cleanup pyqtgraph items."""
         self.unhook_crosshair()
+        self.unhook_fps_monitor(delete_label=False)
         self.tick_item.cleanup()
         self.arrow_item.cleanup()
         item = self.plot_item

tests/unit_tests/conftest.py

@@ -1,5 +1,8 @@
+from unittest import mock
+
 import pytest
 from pytestqt.exceptions import TimeoutError as QtBotTimeoutError
+from qtpy.QtCore import QTimer
 from qtpy.QtWidgets import QApplication
 
 from bec_widgets.cli.rpc_register import RPCRegister
@@ -7,6 +10,22 @@ from bec_widgets.qt_utils import error_popups
 from bec_widgets.utils import bec_dispatcher as bec_dispatcher_module
 
 
+class TestableQTimer(QTimer):
+    _instances = []
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        TestableQTimer._instances.append(self)
+
+    @classmethod
+    def check_all_stopped(cls, qtbot):
+        try:
+            qtbot.waitUntil(lambda: all(not timer.isActive() for timer in cls._instances))
+        except QtBotTimeoutError as exc:
+            raise TimeoutError("Failed to stop all timers") from exc
+        cls._instances = []
+
+
 @pytest.hookimpl(tryfirst=True, hookwrapper=True)
 def pytest_runtest_makereport(item, call):
     # execute all other hooks to obtain the report object
@@ -18,13 +37,16 @@ def pytest_runtest_makereport(item, call):
 
 @pytest.fixture(autouse=True)
 def qapplication(qtbot, request):  # pylint: disable=unused-argument
-    yield
+    with mock.patch("qtpy.QtCore.QTimer", new=TestableQTimer):
+        yield
 
-    # if the test failed, we don't want to check for open widgets as
-    # it simply pollutes the output
-    if request.node.stash._storage.get("failed"):
-        print("Test failed, skipping cleanup checks")
-        return
+        # if the test failed, we don't want to check for open widgets as
+        # it simply pollutes the output
+        if request.node.stash._storage.get("failed"):
+            print("Test failed, skipping cleanup checks")
+            return
+
+        TestableQTimer.check_all_stopped(qtbot)
 
     qapp = QApplication.instance()
     qapp.processEvents()