Add waitSignals.

Author: The-Compiler

pytestqt/_tests/test_wait_signal.py

@@ -7,6 +7,7 @@
 class Signaller(QtCore.QObject):
 
     signal = Signal()
+    signal_2 = Signal()
 
 
 def test_signal_blocker_exception(qtbot):
@@ -18,21 +19,26 @@ def test_signal_blocker_exception(qtbot):
         qtbot.waitSignal(None, None).wait()
 
 
-def explicit_wait(qtbot, signal, timeout):
+def explicit_wait(qtbot, signal, timeout, multiple):
     """
     Explicit wait for the signal using blocker API.
     """
-    blocker = qtbot.waitSignal(signal, timeout)
-    assert not blocker.signal_triggered
+    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
     blocker.wait()
     return blocker
 
 
-def context_manager_wait(qtbot, signal, timeout):
+def context_manager_wait(qtbot, signal, timeout, multiple):
     """
     Waiting for signal using context manager API.
     """
-    with qtbot.waitSignal(signal, timeout) as blocker:
+    func = qtbot.waitSignals if multiple else qtbot.waitSignal
+    with func(signal, timeout) as blocker:
         pass
     return blocker
 
@@ -55,14 +61,15 @@ def test_signal_triggered(qtbot, wait_function, emit_delay, timeout,
     the expected results.
     """
     signaller = Signaller()
+
     timer = QtCore.QTimer()
     timer.setSingleShot(True)
     timer.timeout.connect(signaller.signal.emit)
     timer.start(emit_delay)
 
     # block signal until either signal is emitted or timeout is reached
     start_time = time.time()
-    blocker = wait_function(qtbot, signaller.signal, timeout)
+    blocker = wait_function(qtbot, signaller.signal, timeout, multiple=False)
 
     # Check that event loop exited.
     assert not blocker._loop.isRunning()
@@ -78,6 +85,60 @@ def test_signal_triggered(qtbot, wait_function, emit_delay, timeout,
     assert time.time() - start_time < (max_wait_ms / 1000.0)
 
 
+@pytest.mark.parametrize(
+    ('wait_function', 'emit_delay_1', 'emit_delay_2', 'timeout',
+     'expected_signals_triggered'),
+    [
+        (explicit_wait, 500, 600, 2000, True),
+        (explicit_wait, 500, 600, None, True),
+        (context_manager_wait, 500, 600, 2000, True),
+        (context_manager_wait, 500, 600, None, True),
+        (explicit_wait, 2000, 2000, 500, False),
+        (explicit_wait, 500, 2000, 1000, False),
+        (explicit_wait, 2000, 500, 1000, False),
+        (context_manager_wait, 2000, 2000, 500, False),
+        (context_manager_wait, 500, 2000, 1000, False),
+        (context_manager_wait, 2000, 500, 1000, False),
+    ]
+)
+def test_signal_triggered_multiple(qtbot, wait_function, emit_delay_1,
+                                   emit_delay_2, timeout,
+                                   expected_signals_triggered):
+    """
+    Testing for a signal in different conditions, ensuring we are obtaining
+    the expected results.
+    """
+    signaller = Signaller()
+
+    timer = QtCore.QTimer()
+    timer.setSingleShot(True)
+    timer.timeout.connect(signaller.signal.emit)
+    timer.start(emit_delay_1)
+
+    timer2 = QtCore.QTimer()
+    timer2.setSingleShot(True)
+    timer2.timeout.connect(signaller.signal_2.emit)
+    timer2.start(emit_delay_2)
+
+    # block signal until either signal is emitted or timeout is reached
+    start_time = time.time()
+    blocker = wait_function(qtbot, [signaller.signal, signaller.signal_2],
+                            timeout, multiple=True)
+
+    # Check that event loop exited.
+    assert not blocker._loop.isRunning()
+
+    # ensure that either signal was triggered or timeout occurred
+    assert blocker.signals_triggered == expected_signals_triggered
+
+    # Check that we exited by the earliest parameter; timeout = None means
+    # wait forever, so ensure we waited at most 4 times emit-delay
+    if timeout is None:
+        timeout = max(emit_delay_1, emit_delay_2) * 4
+    max_wait_ms = max(emit_delay_1, emit_delay_2, timeout)
+    assert time.time() - start_time < (max_wait_ms / 1000.0)
+
+
 def test_explicit_emit(qtbot):
     """
     Make sure an explicit emit() inside a waitSignal block works.
@@ -87,3 +148,16 @@ def test_explicit_emit(qtbot):
         signaller.signal.emit()
 
     assert waiting.signal_triggered
+
+
+def test_explicit_emit_multiple(qtbot):
+    """
+    Make sure an explicit emit() inside a waitSignal block works.
+    """
+    signaller = Signaller()
+    with qtbot.waitSignals([signaller.signal, signaller.signal_2],
+                           timeout=5000) as waiting:
+        signaller.signal.emit()
+        signaller.signal_2.emit()
+
+    assert waiting.signals_triggered

pytestqt/plugin.py

@@ -72,6 +72,7 @@ class QtBot(object):
     **Signals**
 
     .. automethod:: waitSignal
+    .. automethod:: waitSignals
 
     **Raw QTest API**
 
@@ -236,7 +237,7 @@ def stopForInteraction(self):
     def waitSignal(self, signal=None, timeout=1000):
         """
         .. versionadded:: 1.2
-        
+
         Stops current test until a signal is triggered.
 
         Used to stop the control flow of a test until a signal is emitted, or
@@ -273,6 +274,50 @@ def waitSignal(self, signal=None, timeout=1000):
 
     wait_signal = waitSignal  # pep-8 alias
 
+    def waitSignals(self, signals=None, timeout=1000):
+        """
+        .. versionadded:: 1.4
+
+        Stops current test until all given signals are triggered.
+
+        Used to stop the control flow of a test until all given signals are
+        emitted, or a number of milliseconds, specified by ``timeout``, has
+        elapsed.
+
+        Best used as a context manager::
+
+           with qtbot.waitSignals(signal, timeout=1000):
+               long_function_that_calls_signal()
+
+        Also, you can use the :class:`MultiSignalBlocker` directly if the
+        context manager form is not convenient::
+
+           blocker = qtbot.waitSignals(signal, timeout=1000)
+           blocker.connect(other_signal)
+           long_function_that_calls_signal()
+           blocker.wait()
+
+        :param list signals:
+            A list of :class:`Signal`s to wait for. Set to ``None`` to just use
+            timeout.
+        :param int timeout:
+            How many milliseconds to wait before resuming control flow.
+        :returns:
+            ``MultiSignalBlocker`` object. Call ``MultiSignalBlocker.wait()``
+            to wait.
+
+        .. note::
+           Cannot have both ``signals`` and ``timeout`` equal ``None``, or
+           else you will block indefinitely. We throw an error if this occurs.
+        """
+        blocker = MultiSignalBlocker(timeout=timeout)
+        if signals is not None:
+            for signal in signals:
+                blocker.add_signal(signal)
+        return blocker
+
+    wait_signals = waitSignals  # pep-8 alias
+
 
 class SignalBlocker(object):
 
@@ -335,6 +380,72 @@ def __exit__(self, type, value, traceback):
         self.wait()
 
 
+class MultiSignalBlocker(object):
+
+    """
+    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()
+        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):
+        """
+        Adds the given signal to the list of signals which :meth:`wait()` waits
+        for.
+
+        :param signal: QtCore.Signal
+        """
+        self._signals[signal] = False
+        signal.connect(functools.partial(self._signal_emitted, signal))
+
+    def _signal_emitted(self, signal):
+        """
+        Called when a given signal is emitted.
+
+        If all expected signals have been emitted, quits the event loop and
+        marks that we finished because signals.
+        """
+        self._signals[signal] = True
+        if all(self._signals.values()):
+            self.signals_triggered = True
+            self._loop.quit()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self.wait()
+
+
 @contextmanager
 def capture_exceptions():
     """
@@ -435,4 +546,4 @@ def pytest_configure(config):
 
 
 def pytest_report_header():
-    return ['qt-api: %s' % QT_API]
+    return ['qt-api: %s' % QT_API]