diff --git a/docs/signals.rst b/docs/signals.rst index b0294311..dd8372d2 100644 --- a/docs/signals.rst +++ b/docs/signals.rst @@ -114,6 +114,16 @@ of the blocker: Signals without arguments will set ``args`` to an empty list. If the time out is reached instead, ``args`` will be ``None``. +Getting all arguments of non-matching arguments +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +.. versionadded:: 2.1 + +When using the ``check_params_cb`` parameter, it may happen that the provided signal is received multiple times with +different parameter values, which may or may not match the requirements of the callback. +``all_args`` then contains the list of signal parameters (as tuple) in the order they were received. + + waitSignals ----------- @@ -174,7 +184,7 @@ evaluation takes place). order parameter -^^^^^^^^^^^^^^^^^^^^^ +^^^^^^^^^^^^^^^ .. versionadded:: 2.0 @@ -194,6 +204,17 @@ A third option is to set ``order="simple"`` which is like "strict", but signals in-between the provided ones, e.g. if the expected signals are ``[a, b, c]`` and the sender actually emits ``[a, a, b, a, c]``, the test completes successfully (it would fail with ``order="strict"``). +Getting emitted signals and arguments +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +.. versionadded:: 2.1 + +To determine which of the expected signals were emitted during a ``wait()`` you can use +``blocker.all_signals_and_args`` which contains a list of +:class:`wait_signal.SignalAndArgs ` objects, indicating the signals (and their arguments) +in the order they were received. + + Making sure a given signal is not emitted ----------------------------------------- diff --git a/pytestqt/qtbot.py b/pytestqt/qtbot.py index 5fba0c4f..8e9d9287 100644 --- a/pytestqt/qtbot.py +++ b/pytestqt/qtbot.py @@ -215,8 +215,12 @@ def waitSignal(self, signal=None, timeout=1000, raising=None, check_params_cb=No .. versionadded:: 1.4 The *raising* parameter. + .. versionadded:: 2.0 + The *check_params_cb* parameter. + :param Signal signal: - A signal to wait for. Set to ``None`` to just use timeout. + A signal to wait for, or a tuple ``(signal, signal_name_as_str)`` to improve the error message that is part + of ``SignalTimeoutError``. Set to ``None`` to just use timeout. :param int timeout: How many milliseconds to wait before resuming control flow. :param bool raising: @@ -225,7 +229,7 @@ def waitSignal(self, signal=None, timeout=1000, raising=None, check_params_cb=No 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. + Optional ``callable`` 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: @@ -274,8 +278,9 @@ def waitSignals(self, signals=None, timeout=1000, raising=None, check_params_cbs 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. Alternatively: a list of (``Signal, str``) tuples of the form + ``(signal, signal_name_as_str)`` to improve the error message that is part of ``SignalTimeoutError``. + Set to ``None`` to just use timeout. :param int timeout: How many milliseconds to wait before resuming control flow. :param bool raising: diff --git a/pytestqt/wait_signal.py b/pytestqt/wait_signal.py index 87572ce6..3cc656c2 100644 --- a/pytestqt/wait_signal.py +++ b/pytestqt/wait_signal.py @@ -1,4 +1,6 @@ import functools +from collections import namedtuple + from pytestqt.qt_compat import qt_api @@ -20,6 +22,8 @@ def __init__(self, timeout=1000, raising=True): self.timeout = timeout self.signal_triggered = False self.raising = raising + self._signals = None # will be initialized by inheriting implementations + self._timeout_message = "" if timeout is None: self._timer = None else: @@ -44,8 +48,7 @@ def wait(self): self._timer.start() self._loop.exec_() if not self.signal_triggered and self.raising: - raise SignalTimeoutError("Didn't get signal after %sms." % - self.timeout) + raise SignalTimeoutError(self._timeout_message) def _quit_loop_by_timeout(self): try: @@ -54,11 +57,79 @@ def _quit_loop_by_timeout(self): self._loop.quit() def _cleanup(self): + # store timeout message before the data to construct it is lost + self._timeout_message = self._get_timeout_error_message() if self._timer is not None: _silent_disconnect(self._timer.timeout, self._quit_loop_by_timeout) self._timer.stop() self._timer = None + def _get_timeout_error_message(self): + """Subclasses have to implement this, returning an appropriate error message for a SignalTimeoutError.""" + raise NotImplementedError # pragma: no cover + + def _extract_pyqt_signal_name(self, potential_pyqt_signal): + signal_name = potential_pyqt_signal.signal # type: str + if not isinstance(signal_name, str): + raise TypeError("Invalid 'signal' attribute in {}. " + "Expected str but got {}".format(signal_name, type(signal_name))) + # strip magic number "2" that PyQt prepends to the signal names + signal_name = signal_name.lstrip('2') + return signal_name + + def _extract_signal_from_signal_tuple(self, potential_signal_tuple): + if isinstance(potential_signal_tuple, tuple): + if len(potential_signal_tuple) != 2: + raise ValueError("Signal tuple must have length of 2 (first element is the signal, " + "the second element is the signal's name).") + signal_tuple = potential_signal_tuple + signal_name = signal_tuple[1] + if not isinstance(signal_name, str): + raise TypeError("Invalid type for provided signal name, " + "expected str but got {}".format(type(signal_name))) + if not signal_name: + raise ValueError("The provided signal name may not be empty") + return signal_name + return "" + + def determine_signal_name(self, potential_signal_tuple): + """ + Attempts to determine the signal's name. If the user provided the signal name as 2nd value of the tuple, this + name has preference. Bad values cause a ``ValueError``. + Otherwise it attempts to get the signal from the ``signal`` attribute of ``signal`` (which only exists for + PyQt signals). + :returns: str name of the signal, an empty string if no signal name can be determined, or raises an error + in case the user provided an invalid signal name manually + """ + signal_name = self._extract_signal_from_signal_tuple(potential_signal_tuple) + + if not signal_name: + try: + signal_name = self._extract_pyqt_signal_name(potential_signal_tuple) + except AttributeError: + # not a PyQt signal + # -> no signal name could be determined + signal_name = "" + + return signal_name + + def get_callback_name(self, callback): + """Attempts to extract the name of the callback. Returns empty string in case of failure.""" + try: + name = callback.__name__ + except AttributeError: + try: + name = callback.func.__name__ # e.g. for callbacks wrapped with functools.partial() + except AttributeError: + name = "" + return name + + @staticmethod + def get_signal_from_potential_signal_tuple(signal_tuple): + if isinstance(signal_tuple, tuple): + return signal_tuple[0] + return signal_tuple + def __enter__(self): return self @@ -103,7 +174,9 @@ def __init__(self, timeout=1000, raising=True, check_params_cb=None): super(SignalBlocker, self).__init__(timeout, raising=raising) self._signals = [] self.args = None + self.all_args = [] self.check_params_callback = check_params_cb + self.signal_name = "" def connect(self, signal): """ @@ -113,16 +186,19 @@ def connect(self, signal): More than one signal can be connected, in which case **any** one of them will make ``wait()`` return. - :param signal: QtCore.Signal + :param signal: QtCore.Signal or tuple (QtCore.Signal, str) """ - signal.connect(self._quit_loop_by_signal) - self._signals.append(signal) + self.signal_name = self.determine_signal_name(potential_signal_tuple=signal) + actual_signal = self.get_signal_from_potential_signal_tuple(signal) + actual_signal.connect(self._quit_loop_by_signal) + self._signals.append(actual_signal) 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: + self.all_args.append(args) if not self.check_params_callback(*args): return # parameter check did not pass try: @@ -138,6 +214,64 @@ def _cleanup(self): _silent_disconnect(signal, self._quit_loop_by_signal) self._signals = [] + def get_params_as_str(self): + if not self.all_args: + return "" + + if len(self.all_args[0]) == 1: + # we have a list of tuples with 1 element each (i.e. the signal has 1 parameter), it doesn't make sense + # to return something like "[(someParam,), (someParam,)]", it's just ugly. Instead return something like + # "[someParam, someParam]" + args_list = [arg[0] for arg in self.all_args] + else: + args_list = self.all_args + + return str(args_list) + + def _get_timeout_error_message(self): + if self.check_params_callback is not None: + return ("Signal {signal_name} emitted with parameters {params} " + "within {timeout} ms, but did not satisfy " + "the {cb_name} callback").format(signal_name=self.signal_name, params=self.get_params_as_str(), + timeout=self.timeout, + cb_name=self.get_callback_name(self.check_params_callback)) + else: + return "Signal {signal_name} not emitted after {timeout} ms".format(signal_name=self.signal_name, + timeout=self.timeout) + + +class SignalAndArgs: + def __init__(self, signal_name, args): + self.signal_name = signal_name + self.args = args + + def _get_readable_signal_with_optional_args(self): + args = repr(self.args) if self.args else "" + + # remove signal parameter signature, e.g. turn "some_signal(QString,int)" to "some_signal", because we're adding + # the actual parameters anyways + signal_name = self.signal_name + signal_name = signal_name.partition('(')[0] + + return signal_name + args + + def __str__(self): + return self._get_readable_signal_with_optional_args() + + def __eq__(self, other): + if isinstance(other, self.__class__): + return self.__dict__ == other.__dict__ + else: + return False + + +# Returns e.g. "3rd" for 3, or "21st" for 21 +get_ordinal_str = lambda n: "%d%s" % (n, {1: "st", 2: "nd", 3: "rd"}.get(n if n < 20 else n % 10, "th")) + + +class NoMatchingIndexFoundError(Exception): + pass + class MultiSignalBlocker(_AbstractSignalBlocker): """ @@ -154,107 +288,226 @@ class MultiSignalBlocker(_AbstractSignalBlocker): def __init__(self, timeout=1000, raising=True, check_params_cbs=None, order="none"): super(MultiSignalBlocker, self).__init__(timeout, raising=raising) - self.order = order - self.check_params_callbacks = check_params_cbs + 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 + self._actual_signal_and_args_at_violation = None + self._signal_names = {} # maps from the unique Signal to the name of the signal (as string) + self.all_signals_and_args = [] # list of SignalAndArgs instances def add_signals(self, signals): """ Adds the given signal to the list of signals which :meth:`wait()` waits for. - :param list signals: list of QtCore.Signal`s + :param list signals: list of QtCore.Signal`s or tuples (QtCore.Signal, str) """ - # determine uniqueness of signals, creating a map that maps from a unique signal to a list of indices + self._determine_unique_signals(signals) + self._create_signal_emitted_indices(signals) + self._connect_unique_signals() + + def _get_timeout_error_message(self): + if not self._are_signal_names_available(): + error_message = self._get_degenerate_error_message() + else: + error_message = self._get_expected_and_actual_signals_message() + if self._strict_order_violated: + error_message = self._get_order_violation_message() + error_message + + return error_message + + def _determine_unique_signals(self, signals): + # create 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) + signals_as_str = [str(self.get_signal_from_potential_signal_tuple(signal)) for signal in signals] + signal_str_to_unique_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 + signal = self.get_signal_from_potential_signal_tuple(signals[index]) + potential_tuple = signals[index] + if signal_str not in signal_str_to_unique_signal: + unique_signal_tuple = potential_tuple + signal_str_to_unique_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) + unique_signal = signal_str_to_unique_signal[signal_str] + self._signals_map[unique_signal].append(index) + unique_signal_tuple = signals[index] + + self._determine_and_save_signal_name(unique_signal_tuple) + def _determine_and_save_signal_name(self, unique_signal_tuple): + signal_name = self.determine_signal_name(unique_signal_tuple) + if signal_name: # might be an empty string if no name could be determined + unique_signal = self.get_signal_from_potential_signal_tuple(unique_signal_tuple) + self._signal_names[unique_signal] = signal_name + + def _create_signal_emitted_indices(self, signals): for signal in signals: self._signals_emitted.append(False) + def _connect_unique_signals(self): for unique_signal in self._signals_map: - slot = functools.partial(self._signal_emitted, unique_signal) + slot = functools.partial(self._unique_signal_emitted, unique_signal) self._slots.append(slot) unique_signal.connect(slot) self._signals.append(unique_signal) - def _signal_emitted(self, signal, *args): + def _unique_signal_emitted(self, unique_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. """ - if self.order == "none": + self._record_emitted_signal_if_possible(unique_signal, *args) + + self._check_signal_match(unique_signal, *args) + + if self._all_signals_emitted(): + self.signal_triggered = True + try: + self._cleanup() + finally: + self._loop.quit() + + def _record_emitted_signal_if_possible(self, unique_signal, *args): + if self._are_signal_names_available(): + self.all_signals_and_args.append( + SignalAndArgs(signal_name=self._signal_names[unique_signal], args=args)) + + def _check_signal_match(self, unique_signal, *args): + 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: + try: + successful_index = self._get_first_matching_index(unique_signal, *args) 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 + except NoMatchingIndexFoundError: # none found + pass + elif self._order == "simple": + if self._check_signal_matches_expected_index(unique_signal, *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): + if self._check_signal_matches_expected_index(unique_signal, *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! + else: + if self._are_signal_names_available(): + self._actual_signal_and_args_at_violation = SignalAndArgs( + signal_name=self._signal_names[unique_signal], args=args) + + def _all_signals_emitted(self): + return not self._strict_order_violated and all(self._signals_emitted) + + def _get_first_matching_index(self, unique_signal, *args): + successfully_emitted = False + successful_index = -1 + potential_indices = self._get_unemitted_signal_indices(unique_signal) + for potential_index in potential_indices: + if not self._violates_callback_at_index(potential_index, *args): + successful_index = potential_index + successfully_emitted = True + break + if not successfully_emitted: + raise NoMatchingIndexFoundError + + return successful_index + + def _check_signal_matches_expected_index(self, unique_signal, *args): + potential_indices = self._get_unemitted_signal_indices(unique_signal) + if potential_indices: + if self._signal_expected_index == potential_indices[0]: + if not self._violates_callback_at_index(self._signal_expected_index, *args): + return True + return False + + def _violates_callback_at_index(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. + Checks if there's a callback at the provided index that is violates due to invalid parameters. Returns False if + there is no callback for that index, or if a callback exists but it wasn't violated (returned True). + Returns True otherwise. """ - if self.check_params_callbacks: - callback_func = self.check_params_callbacks[index] + if self._check_params_callbacks: + callback_func = self._check_params_callbacks[index] if callback_func: if not callback_func(*args): - return False - return True + return True + return False 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 _are_signal_names_available(self): + if self._signal_names: + return True + return False + + def _get_degenerate_error_message(self): + received_signals = sum(self._signals_emitted) + total_signals = len(self._signals_emitted) + return ("Received {actual} of the {total} expected signals. " + "To improve this error message, provide the names of the signals " + "in the waitSignals() call.").format(actual=received_signals, total=total_signals) + + def _get_expected_and_actual_signals_message(self): + if not self.all_signals_and_args: + emitted_signals = "None" + else: + emitted_signal_string_list = [str(_) for _ in self.all_signals_and_args] + emitted_signals = self._format_as_array(emitted_signal_string_list) + + missing_signal_strings = [] + for missing_signal_index in self._get_missing_signal_indices(): + missing_signal_strings.append(self._get_signal_string_representation_for_index(missing_signal_index)) + missing_signals = self._format_as_array(missing_signal_strings) + + return "Emitted signals: {}. Missing: {}".format(emitted_signals, missing_signals) + + @staticmethod + def _format_as_array(list_of_strings): + return "[{}]".format(', '.join(list_of_strings)) + + def _get_order_violation_message(self): + expected_signal_as_str = self._get_signal_string_representation_for_index(self._signal_expected_index) + actual_signal_as_str = str(self._actual_signal_and_args_at_violation) + return ("Signal order violated! Expected {expected} as {ordinal} signal, " + "but received {actual} instead. ").format(expected=expected_signal_as_str, + ordinal=get_ordinal_str(self._signal_expected_index + 1), + actual=actual_signal_as_str) + + def _get_missing_signal_indices(self): + return [index for index, value in enumerate(self._signals_emitted) if not self._signals_emitted[index]] + + def _get_signal_string_representation_for_index(self, index): + """Returns something like or (callback: )""" + signal = self._get_signal_for_index(index) + signal_str_repr = self._signal_names[signal] + + if self._check_params_callbacks: + potential_callback = self._check_params_callbacks[index] + if potential_callback: + callback_name = self.get_callback_name(potential_callback) + if callback_name: + signal_str_repr += " (callback: {})".format(callback_name) + + return signal_str_repr + + def _get_signal_for_index(self, index): + for signal in self._signals_map: + if index in self._signals_map[signal]: + return signal + def _cleanup(self): super(MultiSignalBlocker, self)._cleanup() for i in range(len(self._signals)): diff --git a/tests/test_wait_signal.py b/tests/test_wait_signal.py index b1159ba6..aeb017e4 100644 --- a/tests/test_wait_signal.py +++ b/tests/test_wait_signal.py @@ -2,9 +2,10 @@ import fnmatch import pytest +import sys from pytestqt.qt_compat import qt_api -from pytestqt.wait_signal import SignalEmittedError +from pytestqt.wait_signal import SignalEmittedError, SignalTimeoutError, SignalAndArgs def test_signal_blocker_exception(qtbot): @@ -228,6 +229,7 @@ class Signaller(qt_api.QtCore.QObject): signal_2 = qt_api.Signal() signal_args = qt_api.Signal(str, int) signal_args_2 = qt_api.Signal(str, int) + signal_single_arg = qt_api.Signal(int) assert timer @@ -403,6 +405,57 @@ def test_signal_identity(signaller): assert str(x) != str(z) +def test_invalid_signal(qtbot): + """Tests that a TypeError is raised when providing a signal object that actually is not a Qt signal at all.""" + + class NotReallyASignal: + def __init__(self): + self.signal = False + + with pytest.raises(TypeError): + with qtbot.waitSignal(signal=NotReallyASignal(), raising=False): + pass + + +def test_invalid_signal_tuple_length(qtbot, signaller): + """ + Test that a ValueError is raised when not providing a signal+name tuple with exactly 2 elements + as signal parameter. + """ + with pytest.raises(ValueError): + signal_tuple_with_invalid_length = (signaller.signal, "signal()", "does not belong here") + with qtbot.waitSignal(signal=signal_tuple_with_invalid_length, raising=False): + pass + + +def test_provided_empty_signal_name(qtbot, signaller): + """Test that a ValueError is raised when providing a signal+name tuple where the name is an empty string.""" + with pytest.raises(ValueError): + invalid_signal_tuple = (signaller.signal, "") + with qtbot.waitSignal(signal=invalid_signal_tuple, raising=False): + pass + + +def test_provided_invalid_signal_name_type(qtbot, signaller): + """Test that a TypeError is raised when providing a signal+name tuple where the name is not actually string.""" + with pytest.raises(TypeError): + invalid_signal_tuple = (signaller.signal, 12345) # 12345 is not a signal name + with qtbot.waitSignal(signal=invalid_signal_tuple, raising=False): + pass + + +def test_signalandargs_equality(): + signal_args1 = SignalAndArgs(signal_name="signal", args=(1,2)) + signal_args2 = SignalAndArgs(signal_name="signal", args=(1, 2)) + assert signal_args1 == signal_args2 + + +def test_signalandargs_inequality(): + signal_args1_1 = SignalAndArgs(signal_name="signal", args=(1,2)) + signal_args1_2 = "foo" + assert signal_args1_1 != signal_args1_2 + + def get_waitsignals_cases_all(order): """ Returns the list of tuples (emitted-signal-list, expected-signal-list, expect_signal_triggered) for the @@ -640,6 +693,385 @@ def test_signals_and_callbacks_length_mismatch(self, qtbot, signaller): pass +class TestAllArgs: + """ + Tests blocker.all_args (waitSignal() blocker) which is filled with the args of the emitted signals in case + the signal has args and the user provided a callable for the check_params_cb argument of waitSignal(). + """ + + def test_no_signal_without_args(self, qtbot, signaller): + """When not emitting any signal and expecting one without args, all_args has to be empty.""" + with qtbot.waitSignal(signal=signaller.signal, timeout=200, check_params_cb=None, raising=False) as blocker: + pass # don't emit anything + assert blocker.all_args == [] + + def test_one_signal_without_args(self, qtbot, signaller): + """When emitting an expected signal without args, all_args has to be empty.""" + with qtbot.waitSignal(signal=signaller.signal, timeout=200, check_params_cb=None, raising=False) as blocker: + signaller.signal.emit() + assert blocker.all_args == [] + + def test_one_signal_with_args_matching(self, qtbot, signaller): + """ + When emitting an expected signals with args that match the expected one (satisfy the cb), all_args must + contain these args. + """ + + def cb(str_param, int_param): + return True + + with qtbot.waitSignal(signal=signaller.signal_args, timeout=200, check_params_cb=cb, raising=False) as blocker: + signaller.signal_args.emit('1', 1) + assert blocker.all_args == [('1', 1)] + + def test_two_signals_with_args_partially_matching(self, qtbot, signaller): + """ + When emitting an expected signals with non-matching args followed by emitting it again with matching args, + all_args must contain both of these args. + """ + + def cb(str_param, int_param): + return str_param == '1' and int_param == 1 + + with qtbot.waitSignal(signal=signaller.signal_args, timeout=200, check_params_cb=cb, raising=False) as blocker: + signaller.signal_args.emit('2', 2) + signaller.signal_args.emit('1', 1) + assert blocker.all_args == [('2', 2), ('1', 1)] + + +def get_mixed_signals_with_guaranteed_name(signaller): + """ + Returns a list of signals with the guarantee that the signals have names (i.e. the names are + manually provided in case of using PySide, where the signal names cannot be determined at run-time). + """ + if qt_api.pytest_qt_api == 'pyside': + signals = [(signaller.signal, "signal()"), (signaller.signal_args, "signal_args(QString,int)"), + (signaller.signal_args, "signal_args(QString,int)")] + else: + signals = [signaller.signal, signaller.signal_args, signaller.signal_args] + return signals + + +class TestAllSignalsAndArgs: + """ + Tests blocker.all_signals_and_args (waitSignals() blocker) is a list of SignalAndArgs objects, one for each + received expected signal (irrespective of the order parameter). + """ + + def test_empty_when_no_signal(self, qtbot, signaller): + """Tests that all_signals_and_args is empty when no expected signal is emitted.""" + signals = get_mixed_signals_with_guaranteed_name(signaller) + with qtbot.waitSignals(signals=signals, timeout=200, check_params_cbs=None, order="none", + raising=False) as blocker: + pass + assert blocker.all_signals_and_args == [] + + def test_empty_when_no_signal_name_available(self, qtbot, signaller): + """ + Tests that all_signals_and_args is empty even though expected signals are emitted, but signal names aren't + available. + """ + if qt_api.pytest_qt_api != 'pyside': + pytest.skip("test only makes sense for PySide, whose signals don't contain a name!") + + with qtbot.waitSignals(signals=[signaller.signal, signaller.signal_args, signaller.signal_args], + timeout=200, check_params_cbs=None, order="none", raising=False) as blocker: + signaller.signal.emit() + signaller.signal_args.emit('1', 1) + assert blocker.all_signals_and_args == [] + + def test_non_empty_on_timeout_no_cb(self, qtbot, signaller): + """ + Tests that all_signals_and_args contains the emitted signals. No callbacks for arg-evaluation are provided. The + signals are emitted out of order, causing a timeout. + """ + signals = get_mixed_signals_with_guaranteed_name(signaller) + with qtbot.waitSignals(signals=signals, timeout=200, check_params_cbs=None, order="simple", + raising=False) as blocker: + signaller.signal_args.emit('1', 1) + signaller.signal.emit() + assert not blocker.signal_triggered + assert blocker.all_signals_and_args == [ + SignalAndArgs(signal_name='signal_args(QString,int)', args=('1', 1)), + SignalAndArgs(signal_name='signal()', args=()) + ] + + def test_non_empty_no_cb(self, qtbot, signaller): + """ + Tests that all_signals_and_args contains the emitted signals. No callbacks for arg-evaluation are provided. The + signals are emitted in order. + """ + signals = get_mixed_signals_with_guaranteed_name(signaller) + with qtbot.waitSignals(signals=signals, timeout=200, check_params_cbs=None, order="simple", + raising=False) as blocker: + signaller.signal.emit() + signaller.signal_args.emit('1', 1) + signaller.signal_args.emit('2', 2) + assert blocker.signal_triggered + assert blocker.all_signals_and_args == [ + SignalAndArgs(signal_name='signal()', args=()), + SignalAndArgs(signal_name='signal_args(QString,int)', args=('1', 1)), + SignalAndArgs(signal_name='signal_args(QString,int)', args=('2', 2)) + ] + + +PY_2 = sys.version_info[0] == 2 + + +class TestWaitSignalSignalTimeoutErrorMessage: + """Tests that the messages of SignalTimeoutError are formatted correctly, for waitSignal() calls.""" + + def test_without_callback_and_args(self, qtbot, signaller): + """ + In a situation where a signal without args is expected but not emitted, tests that the SignalTimeoutError + message contains the name of the signal (without arguments). + """ + if qt_api.pytest_qt_api == 'pyside': + signal = (signaller.signal, "signal()") + else: + signal = signaller.signal + + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignal(signal=signal, timeout=200, check_params_cb=None, raising=True): + pass # don't emit any signals + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + assert ex_msg == "Signal signal() not emitted after 200 ms" + + def test_unable_to_get_callback_name(self, qtbot, signaller): + """ + Test that for complicated callbacks which aren't callables, but e.g. double-wrapped partials, the test code + is sometimes unable to determine the name of the callback. + Note that this behavior changes with Python 3.5, where a functools.partial() is smart enough to detect wrapped + calls. + """ + if sys.version_info >= (3,5): + pytest.skip("Only on Python 3.4 and lower double-wrapped functools.partial callbacks are a problem") + + if qt_api.pytest_qt_api == 'pyside': + signal = (signaller.signal_single_arg, "signal_single_arg(int)") + else: + signal = signaller.signal_single_arg + + def callback(int_param, unused_param1, unused_param2): + return int_param == 1337 + + wrapped_callback = functools.partial(callback, unused_param2=1) + double_wrapped_callback = functools.partial(wrapped_callback, unused_param1=1) + + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignal(signal=signal, timeout=200, raising=True, + check_params_cb=double_wrapped_callback): + signaller.signal_single_arg.emit(1) + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + assert ex_msg == ("Signal signal_single_arg(int) emitted with parameters [1] within 200 ms, " + "but did not satisfy the callback") + + def test_with_single_arg(self, qtbot, signaller): + """ + In a situation where a signal with one argument is expected but the emitted instances have values that are + rejected by a callback, tests that the SignalTimeoutError message contains the name of the signal and the + list of non-accepted arguments. + """ + if qt_api.pytest_qt_api == 'pyside': + signal = (signaller.signal_single_arg, "signal_single_arg(int)") + else: + signal = signaller.signal_single_arg + + def arg_validator(int_param): + return int_param == 1337 + + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignal(signal=signal, timeout=200, check_params_cb=arg_validator, raising=True): + signaller.signal_single_arg.emit(1) + signaller.signal_single_arg.emit(2) + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + assert ex_msg == ("Signal signal_single_arg(int) emitted with parameters [1, 2] within 200 ms, " + "but did not satisfy the arg_validator callback") + + def test_with_multiple_args(self, qtbot, signaller): + """ + In a situation where a signal with two arguments is expected but the emitted instances have values that are + rejected by a callback, tests that the SignalTimeoutError message contains the name of the signal and the + list of tuples of the non-accepted arguments. + """ + if qt_api.pytest_qt_api == 'pyside': + signal = (signaller.signal_args, "signal_args(QString,int)") + else: + signal = signaller.signal_args + + def arg_validator(str_param, int_param): + return str_param == "1337" and int_param == 1337 + + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignal(signal=signal, timeout=200, check_params_cb=arg_validator, raising=True): + signaller.signal_args.emit("1", 1) + signaller.signal_args.emit('2', 2) + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + parameters = "[('1', 1), ('2', 2)]" + if PY_2: + parameters = "[(u'1', 1), (u'2', 2)]" + if qt_api.pytest_qt_api == 'pyqt4': + parameters = "[(PyQt4.QtCore.QString(u'1'), 1), (PyQt4.QtCore.QString(u'2'), 2)]" + assert ex_msg == ("Signal signal_args(QString,int) emitted with parameters {} " + "within 200 ms, but did not satisfy the arg_validator callback").format(parameters) + + +class TestWaitSignalsSignalTimeoutErrorMessage: + """Tests that the messages of SignalTimeoutError are formatted correctly, for waitSignals() calls.""" + + @pytest.mark.parametrize("order", ["none", "simple", "strict"]) + def test_no_signal_emitted_with_some_callbacks(self, qtbot, signaller, order): + """ + Tests that the SignalTimeoutError message contains that none of the expected signals were emitted, and lists + the expected signals correctly, with the name of the callbacks where applicable. + """ + + def my_callback(str_param, int_param): + return True + + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignals(signals=get_mixed_signals_with_guaranteed_name(signaller), timeout=200, + check_params_cbs=[None, None, my_callback], order=order, raising=True): + pass # don't emit any signals + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + assert ex_msg == ("Emitted signals: None. Missing: " + "[signal(), signal_args(QString,int), signal_args(QString,int) (callback: my_callback)]") + + @pytest.mark.parametrize("order", ["none", "simple", "strict"]) + def test_no_signal_emitted_no_callbacks(self, qtbot, signaller, order): + """ + Tests that the SignalTimeoutError message contains that none of the expected signals were emitted, and lists + the expected signals correctly (without any callbacks). + """ + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignals(signals=get_mixed_signals_with_guaranteed_name(signaller), timeout=200, + check_params_cbs=None, order=order, raising=True): + pass # don't emit any signals + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + assert ex_msg == ("Emitted signals: None. Missing: " + "[signal(), signal_args(QString,int), signal_args(QString,int)]") + + def test_none_order_one_signal_emitted(self, qtbot, signaller): + """ + When expecting 3 signals but only one of them is emitted, test that the SignalTimeoutError message contains + the emitted signal and the 2 missing expected signals. order is set to "none". + """ + + def my_callback_1(str_param, int_param): + return str_param == "1" and int_param == 1 + + def my_callback_2(str_param, int_param): + return str_param == "2" and int_param == 2 + + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignals(signals=get_mixed_signals_with_guaranteed_name(signaller), timeout=200, + check_params_cbs=[None, my_callback_1, my_callback_2], order="none", raising=True): + signaller.signal_args.emit("1", 1) + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + signal_args = "'1', 1" + if PY_2: + signal_args = "u'1', 1" + if qt_api.pytest_qt_api == 'pyqt4': + signal_args = "PyQt4.QtCore.QString(u'1'), 1" + assert ex_msg == ("Emitted signals: [signal_args({})]. Missing: " + "[signal(), signal_args(QString,int) (callback: my_callback_2)]").format(signal_args) + + def test_simple_order_first_signal_emitted(self, qtbot, signaller): + """ + When expecting 3 signals in a simple order but only the first one is emitted, test that the + SignalTimeoutError message contains the emitted signal and the 2nd+3rd missing expected signals. + """ + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignals(signals=get_mixed_signals_with_guaranteed_name(signaller), timeout=200, + check_params_cbs=None, order="simple", raising=True): + signaller.signal.emit() + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + assert ex_msg == ("Emitted signals: [signal]. Missing: " + "[signal_args(QString,int), signal_args(QString,int)]") + + def test_simple_order_second_signal_emitted(self, qtbot, signaller): + """ + When expecting 3 signals in a simple order but only the second one is emitted, test that the + SignalTimeoutError message contains the emitted signal and all 3 missing expected signals. + """ + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignals(signals=get_mixed_signals_with_guaranteed_name(signaller), timeout=200, + check_params_cbs=None, order="simple", raising=True): + signaller.signal_args.emit("1", 1) + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + signal_args = "'1', 1" + if PY_2: + signal_args = "u'1', 1" + if qt_api.pytest_qt_api == 'pyqt4': + signal_args = "PyQt4.QtCore.QString(u'1'), 1" + assert ex_msg == ("Emitted signals: [signal_args({})]. Missing: " + "[signal(), signal_args(QString,int), signal_args(QString,int)]").format(signal_args) + + def test_strict_order_violation(self, qtbot, signaller): + """ + When expecting 3 signals in a strict order but only the second and then the first one is emitted, test that the + SignalTimeoutError message contains the order violation, the 2 emitted signals and all 3 missing expected + signals. + """ + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignals(signals=get_mixed_signals_with_guaranteed_name(signaller), timeout=200, + check_params_cbs=None, order="strict", raising=True): + signaller.signal_args.emit("1", 1) + signaller.signal.emit() + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + signal_args = "'1', 1" + if PY_2: + signal_args = "u'1', 1" + if qt_api.pytest_qt_api == 'pyqt4': + signal_args = "PyQt4.QtCore.QString(u'1'), 1" + assert ex_msg == ("Signal order violated! Expected signal() as 1st signal, " + "but received signal_args({}) instead. Emitted signals: [signal_args({}), signal]. " + "Missing: [signal(), signal_args(QString,int), signal_args(QString,int)]").format(signal_args, + signal_args) + + def test_degenerate_error_msg(self, qtbot, signaller): + """ + Tests that the SignalTimeoutError message is degenerate when using PySide signals for which no name is provided + by the user. This degenerate messages doesn't contain the signals' names, and includes a hint to the user how + to fix the situation. + """ + if qt_api.pytest_qt_api != 'pyside': + pytest.skip("test only makes sense for PySide, whose signals don't contain a name!") + + with pytest.raises(SignalTimeoutError) as excinfo: + with qtbot.waitSignals(signals=[signaller.signal, signaller.signal_args, signaller.signal_args], + timeout=200, check_params_cbs=None, order="none", + raising=True): + signaller.signal.emit() + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + assert ex_msg == ("Received 1 of the 3 expected signals. " + "To improve this error message, provide the names of the signals " + "in the waitSignals() call.") + + def test_self_defined_signal_name(self, qtbot, signaller): + """ + Tests that the waitSignals implementation prefers the user-provided signal names over the names that can + be determined at runtime from the signal objects themselves. + """ + + def my_cb(str_param, int_param): + return True + + with pytest.raises(SignalTimeoutError) as excinfo: + signals = [(signaller.signal, "signal_without_args"), (signaller.signal_args, "signal_with_args")] + callbacks = [None, my_cb] + with qtbot.waitSignals(signals=signals, timeout=200, check_params_cbs=callbacks, order="none", + raising=True): + pass + ex_msg = TestWaitSignalsSignalTimeoutErrorMessage.get_exception_message(excinfo) + assert ex_msg == ("Emitted signals: None. " + "Missing: [signal_without_args, signal_with_args (callback: my_cb)]") + + @staticmethod + def get_exception_message(excinfo): + return excinfo.value.args[0] + + class TestAssertNotEmitted: """Tests for qtbot.assertNotEmitted."""