Share code between SignalBlocker and MultiSignalBlocker.

Author: The-Compiler

pytestqt/_tests/test_wait_signal.py

@@ -25,10 +25,7 @@ def explicit_wait(qtbot, signal, timeout, multiple):
     """
     func = qtbot.waitSignals if multiple else qtbot.waitSignal
     blocker = func(signal, timeout)
-    if multiple:
-        assert not blocker.signals_triggered
-    else:
-        assert not blocker.signal_triggered
+    assert not blocker.signal_triggered
     blocker.wait()
     return blocker
 
@@ -87,7 +84,7 @@ def test_signal_triggered(qtbot, wait_function, emit_delay, timeout,
 
 @pytest.mark.parametrize(
     ('wait_function', 'emit_delay_1', 'emit_delay_2', 'timeout',
-     'expected_signals_triggered'),
+     'expected_signal_triggered'),
     [
         (explicit_wait, 500, 600, 2000, True),
         (explicit_wait, 500, 600, None, True),
@@ -103,7 +100,7 @@ def test_signal_triggered(qtbot, wait_function, emit_delay, timeout,
 )
 def test_signal_triggered_multiple(qtbot, wait_function, emit_delay_1,
                                    emit_delay_2, timeout,
-                                   expected_signals_triggered):
+                                   expected_signal_triggered):
     """
     Testing for a signal in different conditions, ensuring we are obtaining
     the expected results.
@@ -129,7 +126,7 @@ def test_signal_triggered_multiple(qtbot, wait_function, emit_delay_1,
     assert not blocker._loop.isRunning()
 
     # ensure that either signal was triggered or timeout occurred
-    assert blocker.signals_triggered == expected_signals_triggered
+    assert blocker.signal_triggered == expected_signal_triggered
 
     # Check that we exited by the earliest parameter; timeout = None means
     # wait forever, so ensure we waited at most 4 times emit-delay
@@ -160,4 +157,4 @@ def test_explicit_emit_multiple(qtbot):
         signaller.signal.emit()
         signaller.signal_2.emit()
 
-    assert waiting.signals_triggered
+    assert waiting.signal_triggered

pytestqt/plugin.py

@@ -319,25 +319,29 @@ def waitSignals(self, signals=None, timeout=1000):
     wait_signals = waitSignals  # pep-8 alias
 
 
-class SignalBlocker(object):
+class AbstractSignalBlocker(object):
 
     """
-    Returned by :meth:`QtBot.waitSignal` method.
+    Base class for :class:`SignalBlocker` and :class:`MultiSignalBlocker`.
 
-    .. automethod:: wait
-    .. automethod:: connect
+    Provides :meth:`wait` and a context manager protocol, but no means to add
+    new signals and to detect when the signals should be considered "done".
+    This needs to be implemented by subclasses.
+
+    Subclasses also need to provide ``self._signals`` which should evaluate to
+    ``False`` if no signals were configured.
 
     :ivar int timeout: maximum time to wait for a signal to be triggered. Can
         be changed before :meth:`wait` is called.
 
-    :ivar bool signal_triggered: set to ``True`` if a signal was triggered, or
-        ``False`` if timeout was reached instead. Until :meth:`wait` is called,
-        this is set to ``None``.
+    :ivar bool signal_triggered: set to ``True`` if a signal (or all signals in
+        case of :class:MultipleSignalBlocker:) was triggered, or ``False`` if
+        timeout was reached instead. Until :meth:`wait` is called, this is set
+        to ``None``.
     """
 
     def __init__(self, timeout=1000):
         self._loop = QtCore.QEventLoop()
-        self._signals = []
         self.timeout = timeout
         self.signal_triggered = False
 
@@ -350,12 +354,33 @@ def wait(self):
         """
         if self.signal_triggered:
             return
-        if self.timeout is None and len(self._signals) == 0:
+        if self.timeout is None and not self._signals:
             raise ValueError("No signals or timeout specified.")
         if self.timeout is not None:
             QtCore.QTimer.singleShot(self.timeout, self._loop.quit)
         self._loop.exec_()
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self.wait()
+
+
+class SignalBlocker(AbstractSignalBlocker):
+
+    """
+    Returned by :meth:`QtBot.waitSignal` method.
+
+    .. automethod:: wait
+    .. automethod:: connect
+
+    """
+
+    def __init__(self, timeout=1000):
+        super(SignalBlocker, self).__init__(timeout)
+        self._signals = []
+
     def connect(self, signal):
         """
         Connects to the given signal, making :meth:`wait()` return once this signal
@@ -373,49 +398,19 @@ def _quit_loop_by_signal(self):
         self.signal_triggered = True
         self._loop.quit()
 
-    def __enter__(self):
-        return self
 
-    def __exit__(self, type, value, traceback):
-        self.wait()
-
-
-class MultiSignalBlocker(object):
+class MultiSignalBlocker(AbstractSignalBlocker):
 
     """
     Returned by :meth:`QtBot.waitSignals` method.
 
     .. automethod:: wait
     .. automethod:: add_signal
-
-    :ivar int timeout: maximum time to wait for a signal to be triggered. Can
-        be changed before :meth:`wait` is called.
-
-    :ivar bool signals_triggered: set to ``True`` if all signals were
-        triggered, or ``False`` if timeout was reached instead. Until
-        :meth:`wait` is called, this is set to ``None``.
     """
 
     def __init__(self, timeout=1000):
-        self._loop = QtCore.QEventLoop()
+        super(MultiSignalBlocker, self).__init__(timeout)
         self._signals = {}
-        self.timeout = timeout
-        self.signals_triggered = False
-
-    def wait(self):
-        """
-        Waits until either a connected signal is triggered or timeout is reached.
-
-        :raise ValueError: if no signals are connected and timeout is None; in
-            this case it would wait forever.
-        """
-        if self.signals_triggered:
-            return
-        if self.timeout is None and not self._signals:
-            raise ValueError("No signals or timeout specified.")
-        if self.timeout is not None:
-            QtCore.QTimer.singleShot(self.timeout, self._loop.quit)
-        self._loop.exec_()
 
     def add_signal(self, signal):
         """
@@ -436,15 +431,9 @@ def _signal_emitted(self, signal):
         """
         self._signals[signal] = True
         if all(self._signals.values()):
-            self.signals_triggered = True
+            self.signal_triggered = True
             self._loop.quit()
 
-    def __enter__(self):
-        return self
-
-    def __exit__(self, type, value, traceback):
-        self.wait()
-
 
 @contextmanager
 def capture_exceptions():