Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Added signal blocker, as discussed in #12 #13

Merged
merged 1 commit into from Jul 2, 2014
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
46 changes: 46 additions & 0 deletions pytestqt/_tests/test_wait_signal.py
@@ -0,0 +1,46 @@
import pytest
import time

from pytestqt.qt_compat import QtCore, Signal


def test_signal_blocker_exception(qtbot):
with pytest.raises(ValueError):
qtbot.waitSignal(None, None).wait()


class Signaller(QtCore.QObject):

signal = Signal()


def test_wait_signal_context_manager(qtbot, monkeypatch):
signaller = Signaller()

# Emit a signal after half a second, and block the signal with a timeout
# of 2 seconds.
QtCore.QTimer.singleShot(500, signaller.signal.emit)
with qtbot.waitSignal(signaller.signal, 2000) as blocker:
saved_loop = blocker.loop
start_time = time.time()

# Check that event loop exited.
assert not saved_loop.isRunning()
# Check that it didn't exit by a timeout.
assert time.time() - start_time < 2 # Less than 2 seconds elapsed


def test_wait_signal_function(qtbot, monkeypatch):
signaller = Signaller()

# Emit a signal after half a second, and block the signal with a timeout
# of 2 seconds.
QtCore.QTimer.singleShot(500, signaller.signal.emit)
blocker = qtbot.waitSignal(signaller.signal, 2000)
start_time = time.time()
blocker.wait()

# Check that event loop exited.
assert not blocker.loop.isRunning()
# Check that it didn't exit by a timeout.
assert time.time() - start_time < 2 # Less than 2 seconds elapsed
66 changes: 64 additions & 2 deletions pytestqt/plugin.py
Expand Up @@ -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):
Expand Down Expand Up @@ -61,6 +60,7 @@ class QtBot(object):
.. automethod:: addWidget
.. automethod:: waitForWindowShown
.. automethod:: stopForInteraction
.. automethod:: waitSignal

**Raw QTest API**

Expand Down Expand Up @@ -212,6 +212,68 @@ 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()

Can also be used to return blocker object::

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:

def __init__(self, timeout=1000):
self.loop = QtCore.QEventLoop()
self._signals = []
self.timeout = timeout

def wait(self):
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.loop.exec_()

def connect(self, signal):
signal.connect(self.loop.quit)
self._signals.append(signal)

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):
"""
Expand Down