diff --git a/docs/index.rst b/docs/index.rst index de8bed0f..87d986fb 100644 --- a/docs/index.rst +++ b/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 ` +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 ========== diff --git a/pytestqt/_tests/test_wait_signal.py b/pytestqt/_tests/test_wait_signal.py new file mode 100644 index 00000000..76daf4e4 --- /dev/null +++ b/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) diff --git a/pytestqt/plugin.py b/pytestqt/plugin.py index 33c7f7b4..f412d6c7 100644 --- a/pytestqt/plugin.py +++ b/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): """