Skip to content

waitSignal and waitSignals() evaluate signal parameters #141

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.

Sign up for GitHub

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

Merged
merged 3 commits into from
Jul 28, 2016
Merged

waitSignal and waitSignals() evaluate signal parameters #141

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b8507e3
waitSignal and waitSignals() have a callback/callbacks parameter that…
Jul 14, 2016
a8cee4c
waitSignal and waitSignals() have a callback/callbacks parameter that…
Jul 14, 2016
2e21419
Fixed various smaller issues:
Jul 18, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 32 additions & 8 deletions pytestqt/qtbot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,7 +234,7 @@ def stopForInteraction(self):

stop = stopForInteraction

def waitSignal(self, signal=None, timeout=1000, raising=None):
def waitSignal(self, signal=None, timeout=1000, raising=None, check_params_cb=None):
"""
.. versionadded:: 1.2

Expand Down Expand Up @@ -270,6 +270,10 @@ def waitSignal(self, signal=None, timeout=1000, raising=None):
should be raised if a timeout occurred.
This defaults to ``True`` unless ``qt_wait_signal_raising = false``
is set in the config.
:param Callable check_params_cb:
Optional callable(*parameters) that compares the provided signal parameters to some expected parameters.
It has to match the signature of ``signal`` (just like a slot function would) and return ``True`` if
parameters match, ``False`` otherwise.
:returns:
``SignalBlocker`` object. Call ``SignalBlocker.wait()`` to wait.

Expand All @@ -285,14 +289,14 @@ def waitSignal(self, signal=None, timeout=1000, raising=None):
raising = True
else:
raising = _parse_ini_boolean(raising_val)
blocker = SignalBlocker(timeout=timeout, raising=raising)
blocker = SignalBlocker(timeout=timeout, raising=raising, check_params_cb=check_params_cb)
if signal is not None:
blocker.connect(signal)
return blocker

wait_signal = waitSignal # pep-8 alias

def waitSignals(self, signals=None, timeout=1000, raising=None):
def waitSignals(self, signals=None, timeout=1000, raising=None, check_params_cbs=None, order="none"):
"""
.. versionadded:: 1.4

Expand All @@ -315,15 +319,28 @@ def waitSignals(self, signals=None, timeout=1000, raising=None):
blocker.wait()

:param list signals:
A list of :class:`Signal` objects to wait for. Set to ``None`` to
just use timeout.
A list of :class:`Signal` objects to wait for. Set to ``None`` to just use
timeout.
:param int timeout:
How many milliseconds to wait before resuming control flow.
:param bool raising:
If :class:`QtBot.SignalTimeoutError <pytestqt.plugin.SignalTimeoutError>`
should be raised if a timeout occurred.
This defaults to ``True`` unless ``qt_wait_signal_raising = false``
is set in the config.
:param list check_params_cbs:
optional list of callables that compare the provided signal parameters to some expected parameters.
Each callable has to match the signature of the corresponding signal in ``signals`` (just like a slot
function would) and return ``True`` if parameters match, ``False`` otherwise.
Instead of a specific callable, ``None`` can be provided, to disable parameter checking for the
corresponding signal.
If the number of callbacks doesn't match the number of signals ``ValueError`` will be raised.
:param str order: determines the order in which to expect signals
* ``"none"``: no order is enforced
* ``"strict"``: signals have to be emitted strictly in the provided order
(e.g. fails when expecting signals [a, b] and [a, a, b] is emitted)
* ``"simple"``: like "strict", but signals may be emitted in-between the provided ones, e.g. expected
``signals`` == [a, b, c] and actually emitted signals = [a, a, b, a, c] works (would fail with "strict")
:returns:
``MultiSignalBlocker`` object. Call ``MultiSignalBlocker.wait()``
to wait.
Expand All @@ -334,12 +351,19 @@ def waitSignals(self, signals=None, timeout=1000, raising=None):

.. note:: This method is also available as ``wait_signals`` (pep-8 alias)
"""
if order not in ["none", "simple", "strict"]:
raise ValueError("order has to be set to 'none', 'simple' or 'strict'")

if raising is None:
raising = self._request.config.getini('qt_wait_signal_raising')
blocker = MultiSignalBlocker(timeout=timeout, raising=raising)

if check_params_cbs:
if len(check_params_cbs) != len(signals):
raise ValueError("Number of callbacks ({}) does not "
"match number of signals ({})!".format(len(check_params_cbs), len(signals)))
blocker = MultiSignalBlocker(timeout=timeout, raising=raising, order=order, check_params_cbs=check_params_cbs)
if signals is not None:
for signal in signals:
blocker._add_signal(signal)
blocker.add_signals(signals)
return blocker

wait_signals = waitSignals # pep-8 alias
Expand Down
116 changes: 95 additions & 21 deletions pytestqt/wait_signal.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@


class _AbstractSignalBlocker(object):

"""
Base class for :class:`SignalBlocker` and :class:`MultiSignalBlocker`.

Expand Down Expand Up @@ -71,7 +70,6 @@ def __exit__(self, type, value, traceback):


class SignalBlocker(_AbstractSignalBlocker):

"""
Returned by :meth:`pytestqt.qtbot.QtBot.waitSignal` method.

Expand Down Expand Up @@ -101,10 +99,11 @@ class SignalBlocker(_AbstractSignalBlocker):
.. automethod:: connect
"""

def __init__(self, timeout=1000, raising=True):
def __init__(self, timeout=1000, raising=True, check_params_cb=None):
super(SignalBlocker, self).__init__(timeout, raising=raising)
self._signals = []
self.args = None
self.check_params_callback = check_params_cb

def connect(self, signal):
"""
Expand All @@ -123,6 +122,9 @@ def _quit_loop_by_signal(self, *args):
"""
quits the event loop and marks that we finished because of a signal.
"""
if self.check_params_callback:
if not self.check_params_callback(*args):
return # parameter check did not pass
Copy link
Member

@nicoddemus nicoddemus Jul 15, 2016 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if we should store the received args, so we can provide a better error message?

Consider:

def is_disconnected(text):
    return text == 'END'

with qtbot.waitSignal(conn.text_received, callback=is_disconnected):
    conn.request_disconnect()

conn.text_received is emitted with ['WAIT_CLIENTS', 'PROCESSING'] and eventually times out.

In this situation all the user will see is a TimeoutError and won't have any hints at what might have gone wrong. I suppose the user would then have to add print statements at is_disconnected to try to figure out the problem.

I think if we store the received arguments when there is a callback, we can reuse that later in the exception message:

TimeoutError:  `text_received` emitted with ['WAIT_CLIENTS', 'PROCESSING'] 
but did not satisfy the `is_disconnected` callback.

That idea also implies that the same arguments should be available for inspection later in case raising is False, probably as a list of *args:

blocker.all_args == ['WAIT_CLIENTS', 'PROCESSING']

What do you guys think?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds good.
Another question is: should this also be implemented for MultiSignalBlocker? It would generally be interesting to get a string-dump (in human readable form) of all recorded signals and their parameters, including the points of time where things went wrong (e.g. in case of strict order when the order was violated). Similarly, if raising == False, blocker.recorded_signals (or something similar) could contain a list of recorded signals together with their parameters.

try:
self.signal_triggered = True
self.args = list(args)
Expand All @@ -138,7 +140,6 @@ def _cleanup(self):


class MultiSignalBlocker(_AbstractSignalBlocker):

"""
Returned by :meth:`pytestqt.qtbot.QtBot.waitSignals` method, blocks until
all signals connected to it are triggered or the timeout is reached.
Expand All @@ -151,48 +152,121 @@ class MultiSignalBlocker(_AbstractSignalBlocker):
.. automethod:: wait
"""

def __init__(self, timeout=1000, raising=True):
def __init__(self, timeout=1000, raising=True, check_params_cbs=None, order="none"):
super(MultiSignalBlocker, self).__init__(timeout, raising=raising)
self._signals = {}
self._slots = {}

def _add_signal(self, signal):
self.order = order
self.check_params_callbacks = check_params_cbs
self._signals_emitted = [] # list of booleans, indicates whether the signal was already emitted
self._signals_map = {} # maps from a unique Signal to a list of indices where to expect signal instance emits
self._signals = [] # list of all Signals (for compatibility with _AbstractSignalBlocker)
self._slots = [] # list of slot functions
self._signal_expected_index = 0 # only used when forcing order
self._strict_order_violated = False

def add_signals(self, signals):
"""
Adds the given signal to the list of signals which :meth:`wait()` waits
for.

:param signal: QtCore.Signal
:param list signals: list of QtCore.Signal`s
"""
self._signals[signal] = False
slot = functools.partial(self._signal_emitted, signal)
self._slots[signal] = slot
signal.connect(slot)
# determine uniqueness of signals, creating a map that maps from a unique signal to a list of indices
# (positions) where this signal is expected (in case order matters)
signals_as_str = [str(signal) for signal in signals]
signal_str_to_signal = {} # maps from a signal-string to one of the signal instances (the first one found)
for index, signal_str in enumerate(signals_as_str):
signal = signals[index]
if signal_str not in signal_str_to_signal:
signal_str_to_signal[signal_str] = signal
self._signals_map[signal] = [index] # create a new list
else:
# append to existing list
first_signal_that_occurred = signal_str_to_signal[signal_str]
self._signals_map[first_signal_that_occurred].append(index)

def _signal_emitted(self, signal):
for signal in signals:
self._signals_emitted.append(False)

for unique_signal in self._signals_map:
slot = functools.partial(self._signal_emitted, unique_signal)
self._slots.append(slot)
unique_signal.connect(slot)
self._signals.append(unique_signal)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unrelated whitespace change

def _signal_emitted(self, signal, *args):
"""
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()):
if self.order == "none":
# perform the test for every matching index (stop after the first one that matches)
successfully_emitted = False
successful_index = -1
potential_indices = self._get_unemitted_signal_indices(signal)
for potential_index in potential_indices:
if self._check_callback(potential_index, *args):
successful_index = potential_index
successfully_emitted = True
break

if successfully_emitted:
self._signals_emitted[successful_index] = True
elif self.order == "simple":
potential_indices = self._get_unemitted_signal_indices(signal)
if potential_indices:
if self._signal_expected_index == potential_indices[0]:
if self._check_callback(self._signal_expected_index, *args):
self._signals_emitted[self._signal_expected_index] = True
self._signal_expected_index += 1
else: # self.order == "strict"
if not self._strict_order_violated:
# only do the check if the strict order has not been violated yet
self._strict_order_violated = True # assume the order has been violated this time
potential_indices = self._get_unemitted_signal_indices(signal)
if potential_indices:
if self._signal_expected_index == potential_indices[0]:
if self._check_callback(self._signal_expected_index, *args):
self._signals_emitted[self._signal_expected_index] = True
self._signal_expected_index += 1
self._strict_order_violated = False # order has not been violated after all!

if not self._strict_order_violated and all(self._signals_emitted):
try:
self.signal_triggered = True
self._cleanup()
finally:
self._loop.quit()

def _check_callback(self, index, *args):
"""
Checks if there's a callback that evaluates the validity of the parameters. Returns False if there is one
and its evaluation revealed that the parameters were invalid. Returns True otherwise.
"""
if self.check_params_callbacks:
callback_func = self.check_params_callbacks[index]
if callback_func:
if not callback_func(*args):
return False
return True

def _get_unemitted_signal_indices(self, signal):
"""Returns the indices for the provided signal for which NO signal instance has been emitted yet."""
return [index for index in self._signals_map[signal] if self._signals_emitted[index] == False]

def _cleanup(self):
super(MultiSignalBlocker, self)._cleanup()
for signal, slot in self._slots.items():
for i in range(len(self._signals)):
signal = self._signals[i]
slot = self._slots[i]
_silent_disconnect(signal, slot)
self._signals.clear()
self._slots.clear()
del self._signals_emitted[:]
self._signals_map.clear()
del self._slots[:]


class SignalEmittedSpy(object):

"""
.. versionadded:: 1.11

Expand Down
Loading