WIP: Inherit signal blockers from QObject

Author: The-Compiler

src/pytestqt/wait_signal_impl.py

@@ -47,14 +47,15 @@ class _AbstractSignalBlocker(qt_api.QtCore.QObject):
     """
 
     def __init__(self, timeout=5000, raising=True):
+        super().__init__()
         self._loop = qt_api.QtCore.QEventLoop()
         self.timeout = timeout
         self.signal_triggered = False
         self.raising = raising
         self._signals = None  # will be initialized by inheriting implementations
         self._timeout_message = ""
 
-        self._timer = qt_api.QtCore.QTimer(self._loop)
+        self._timer = qt_api.QtCore.QTimer(self)
         self._timer.setSingleShot(True)
         if timeout is not None:
             self._timer.setInterval(timeout)
@@ -84,13 +85,15 @@ def wait(self):
         if not self.signal_triggered and self.raising:
             raise TimeoutError(self._timeout_message)
 
+    @qt_api.Slot()
     def _quit_loop_by_timeout(self):
         try:
             self._cleanup()
         finally:
             self._loop.quit()
 
     def _cleanup(self):
+        # assert self._timer.thread() == qt_api.QtCore.QThread.currentThread()
         # store timeout message before the data to construct it is lost
         self._timeout_message = self._get_timeout_error_message()
         self._timer.stop()
@@ -234,6 +237,7 @@ def connect(self, signal):
         actual_signal.connect(self._quit_loop_by_signal)
         self._signals.append(actual_signal)
 
+    @qt_api.Slot()
     def _quit_loop_by_signal(self, *args):
         """
         quits the event loop and marks that we finished because of a signal.
@@ -251,6 +255,8 @@ def _quit_loop_by_signal(self, *args):
 
     def _cleanup(self):
         super()._cleanup()
+        # FIXME move to _AbstractSignalBlocker once we got MultiSignalBlocker correct
+        assert self._timer.thread() == qt_api.QtCore.QThread.currentThread()
         for signal in self._signals:
             _silent_disconnect(signal, self._quit_loop_by_signal)
         self._signals = []
@@ -567,7 +573,7 @@ def _cleanup(self):
         del self._slots[:]
 
 
-class CallbackBlocker:
+class CallbackBlocker(qt_api.QtCore.QObject):
     """
     .. versionadded:: 3.1
 
@@ -595,14 +601,15 @@ class CallbackBlocker:
     """
 
     def __init__(self, timeout=5000, raising=True):
+        super().__init__()
         self.timeout = timeout
         self.raising = raising
         self.args = None
         self.kwargs = None
         self.called = False
         self._loop = qt_api.QtCore.QEventLoop()
 
-        self._timer = qt_api.QtCore.QTimer(self._loop)
+        self._timer = qt_api.QtCore.QTimer(self)
         self._timer.setSingleShot(True)
         if timeout is not None:
             self._timer.setInterval(timeout)
@@ -631,13 +638,15 @@ def assert_called_with(self, *args, **kwargs):
         assert self.args == list(args)
         assert self.kwargs == kwargs
 
+    @qt_api.Slot()
     def _quit_loop_by_timeout(self):
         try:
             self._cleanup()
         finally:
             self._loop.quit()
 
     def _cleanup(self):
+        assert self._timer.thread() == qt_api.QtCore.QThread.currentThread()
         self._timer.stop()
 
     def __call__(self, *args, **kwargs):

tests/test_wait_signal.py

@@ -1367,17 +1367,17 @@ def test_timeout_not_raising(self, qtbot):
 
 
 @pytest.mark.parametrize(
-    "check_stderr, count",
+    "check_warnings, count",
     [
-        # Checking stderr messages
+        # Checking for warnings
         pytest.param(
-            True,  # check stderr
+            True,  # check warnings
             200,  # gets output reliably even with only few runs (often the first)
             id="stderr",
         ),
         # Triggering AttributeError
         pytest.param(
-            False,  # don't check stderr
+            False,  # don't check warnings
             # Hopefully enough to trigger the AttributeError race condition reliably.
             # With 500 runs, only 1 of 5 Windows PySide6 CI jobs triggered it (but all
             # Ubuntu/macOS jobs did).  With 1500 runs, Windows jobs still only triggered
@@ -1392,7 +1392,11 @@ def test_timeout_not_raising(self, qtbot):
 )
 @pytest.mark.parametrize("multi_blocker", [True, False])
 def test_signal_raised_from_thread(
-    pytester: pytest.Pytester, check_stderr: bool, multi_blocker: bool, count: int
+    monkeypatch: pytest.MonkeyPatch,
+    pytester: pytest.Pytester,
+    check_warnings: bool,
+    multi_blocker: bool,
+    count: int,
 ) -> None:
     """Wait for a signal with a thread.
 
@@ -1409,7 +1413,7 @@ class Worker(qt_api.QtCore.QObject):
 
 
         @pytest.mark.parametrize("_", range({count}))
-        def test_thread(qtbot, capfd, _):
+        def test_thread(qtbot, _):
             worker = Worker()
             thread = qt_api.QtCore.QThread()
             worker.moveToThread(thread)
@@ -1425,25 +1429,23 @@ def test_thread(qtbot, capfd, _):
             finally:
                 thread.quit()
                 thread.wait()
-
-            if {check_stderr}:  # check_stderr
-                out, err = capfd.readouterr()
-                assert not err
     """
     )
-
+    if check_warnings:
+        monkeypatch.setenv("QT_FATAL_WARNINGS", "1")
     res = pytester.runpytest_subprocess("-x", "-s")
-    outcomes = res.parseoutcomes()
 
-    if outcomes.get("failed", 0) and check_stderr and qt_api.pytest_qt_api == "pyside6":
-        # The test succeeds on PyQt (unsure why!), but we can't check
-        # qt_api.pytest_qt_api at import time, so we can't use
-        # pytest.mark.xfail conditionally.
-        pytest.xfail(
-            "Qt error: QObject::killTimer: "
-            "Timers cannot be stopped from another thread"
-        )
+    qtimer_message = "QObject::killTimer: Timers cannot be stopped from another thread"
+    if (
+        qtimer_message in res.stderr.str()
+        and multi_blocker
+        and check_warnings
+        and qt_api.pytest_qt_api == "pyside6"
+    ):
+        # We haven't fixed MultiSignalBlocker yet...
+        pytest.xfail(f"Qt error: {qtimer_message}")
 
+    outcomes = res.parseoutcomes()
     res.assert_outcomes(passed=outcomes["passed"])  # no failed/error