Wait signal

docs/index.rst

@@ -116,14 +116,37 @@ created earlier::
     assert window.filesTable.item(0, 0).text() == 'video1.avi'
     assert window.filesTable.item(1, 0).text() == 'video2.avi'    
 
-
-And that's it for this quick tutorial!    
+
+
+Waiting for threads, processes, etc.
+====================================
+
+If your program has long running cumputations running in other threads or
+processes, you can use :meth:`qtbot.waitSignal <pytestqt.plugin.QtBot.waitSignal>`
+to block a test until a signal is emitted (such as ``QThread.finished``) or a
+timeout is reached. This makes it easy to write tests that wait until a
+computation running in another thread or process is completed before
+ensuring the results are correct::
+
+    def test_long_computation(qtbot):
+        app = Application()
+
+        # Watch for the app.worker.finished signal, then start the worker.
+        with qtbot.waitSignal(app.worker.finished, timeout=10000) as blocker:
+            blocker.connect(app.worker.failed)  # Can add other signals to blocker
+            app.worker.start()
+            # Test will wait here until either signal is emitted, or 10 seconds has elapsed
+
+        assert blocker.signal_triggered  # Assuming the work took less than 10 seconds
+        assert_application_results(app)
+
 
 QtBot
 =====
 
 .. module:: pytestqt.plugin
 .. autoclass:: QtBot
+.. autoclass:: SignalBlocker
 
 Versioning
 ==========

pytestqt/_tests/test_wait_signal.py

@@ -0,0 +1,75 @@
+import pytest
+import time
+
+from pytestqt.qt_compat import QtCore, Signal
+
+
+class Signaller(QtCore.QObject):
+
+    signal = Signal()
+
+
+def test_signal_blocker_exception(qtbot):
+    """
+    Make sure waitSignal without signals and timeout doesn't hang, but raises
+    ValueError instead.
+    """
+    with pytest.raises(ValueError):
+        qtbot.waitSignal(None, None).wait()
+
+
+def explicit_wait(qtbot, signal, timeout):
+    """
+    Explicit wait for the signal using blocker API.
+    """
+    blocker = qtbot.waitSignal(signal, timeout)
+    assert blocker.signal_triggered is None
+    blocker.wait()
+    return blocker
+
+
+def context_manager_wait(qtbot, signal, timeout):
+    """
+    Waiting for signal using context manager API.
+    """
+    with qtbot.waitSignal(signal, timeout) as blocker:
+        pass
+    return blocker
+
+
+@pytest.mark.parametrize(
+    ('wait_function', 'emit_delay', 'timeout', 'expected_signal_triggered'),
+    [
+        (explicit_wait, 500, 2000, True),
+        (explicit_wait, 500, None, True),
+        (context_manager_wait, 500, 2000, True),
+        (context_manager_wait, 500, None, True),
+        (explicit_wait, 2000, 500, False),
+        (context_manager_wait, 2000, 500, False),
+    ]
+)
+def test_signal_triggered(qtbot, wait_function, emit_delay, timeout,
+                          expected_signal_triggered):
+    """
+    Testing for a signal in different conditions, ensuring we are obtaining
+    the expected results.
+    """
+    signaller = Signaller()
+    QtCore.QTimer.singleShot(emit_delay, signaller.signal.emit)
+
+    # block signal until either signal is emitted or timeout is reached
+    start_time = time.time()
+    blocker = wait_function(qtbot, signaller.signal, timeout)
+
+    # Check that event loop exited.
+    assert not blocker._loop.isRunning()
+
+    # ensure that either signal was triggered or timeout occurred
+    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
+    if timeout is None:
+        timeout = emit_delay * 4
+    max_wait_ms = max(emit_delay, timeout)
+    assert time.time() - start_time < (max_wait_ms / 1000.0)

pytestqt/plugin.py

@@ -4,8 +4,7 @@
 
 import pytest
 
-from pytestqt.qt_compat import QtGui
-from pytestqt.qt_compat import QtTest
+from pytestqt.qt_compat import QtCore, QtGui, QtTest
 
 
 def _inject_qtest_methods(cls):
@@ -61,6 +60,7 @@ class QtBot(object):
     .. automethod:: addWidget
     .. automethod:: waitForWindowShown
     .. automethod:: stopForInteraction
+    .. automethod:: waitSignal
 
     **Raw QTest API**
 
@@ -212,6 +212,105 @@ def stopForInteraction(self):
 
     stop = stopForInteraction
 
+    def waitSignal(self, signal=None, timeout=1000):
+        """
+        Stops current test until a signal is triggered.
+
+        Used to stop the control flow of a test until a signal is emitted, or
+        a number of milliseconds, specified by ``timeout``, has elapsed.
+
+        Best used as a context manager::
+
+           with qtbot.waitSignal(signal, timeout=1000):
+               long_function_that_calls_signal()
+
+        Also, you can use the :class:`SignalBlocker` directly if the context
+        manager form is not convenient::
+
+           blocker = qtbot.waitSignal(signal, timeout=1000)
+           blocker.connect(other_signal)
+           long_function_that_calls_signal()
+           blocker.wait()
+
+        :param Signal signal:
+            A signal to wait for. Set to ``None`` to just use timeout.
+        :param int timeout:
+            How many milliseconds to wait before resuming control flow.
+        :returns:
+            ``SignalBlocker`` object. Call ``SignalBlocker.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 = SignalBlocker(timeout=timeout)
+        if signal is not None:
+            blocker.connect(signal)
+        return blocker
+
+
+class SignalBlocker(object):
+    """
+    Returned by :meth:`QtBot.waitSignal` method.
+
+    .. automethod:: wait
+    .. automethod:: connect
+
+    :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``.
+    """
+
+    def __init__(self, timeout=1000):
+        self._loop = QtCore.QEventLoop()
+        self._signals = []
+        self.timeout = timeout
+        self.signal_triggered = None
+
+    def wait(self):
+        """
+        Waits until either condition 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.timeout is None and len(self._signals) == 0:
+            raise ValueError("No signals or timeout specified.")
+        if self.timeout is not None:
+            QtCore.QTimer.singleShot(self.timeout, self._loop.quit)
+        self.signal_triggered = False
+        self._loop.exec_()
+
+    def connect(self, signal):
+        """
+        Connects to the given signal, making :meth:`wait()` return once this signal
+        is emitted.
+
+        :param signal: QtCore.Signal
+        """
+        signal.connect(self._quit_loop_by_signal)
+        self._signals.append(signal)
+
+
+    def _quit_loop_by_signal(self):
+        """
+        quits the event loop and marks that we finished because of a signal.
+        """
+        self.signal_triggered = True
+        self._loop.quit()
+
+    def __enter__(self):
+        # Return self for testing purposes. Generally not needed.
+        return self
+
+    def __exit__(self, type, value, traceback):
+        self.wait()
+
 
 def pytest_configure(config):
     """