From dfc1782bbf447aed92f68031154af8cef4d131f7 Mon Sep 17 00:00:00 2001 From: Florian Bruhin Date: Thu, 22 Oct 2015 22:23:55 +0200 Subject: [PATCH] Add testprocess.wait_for to react to messages. --- tests/integration/quteprocess.py | 47 ++++++++++++--------------- tests/integration/testprocess.py | 55 +++++++++++++++++++++++++++++++- 2 files changed, 74 insertions(+), 28 deletions(-) diff --git a/tests/integration/quteprocess.py b/tests/integration/quteprocess.py index 6f45c8afa..8fc115115 100644 --- a/tests/integration/quteprocess.py +++ b/tests/integration/quteprocess.py @@ -61,12 +61,16 @@ class LogLine: """, re.VERBOSE) def __init__(self, line): + self._line = line match = self.LOG_RE.match(line) if match is None: raise NoLineMatch(line) self.__dict__.update(match.groupdict()) self.expected = False + def __repr__(self): + return 'LogLine({!r})'.format(self._line) + class QuteProc(testprocess.Process): @@ -77,9 +81,6 @@ class QuteProc(testprocess.Process): _httpbin: The HTTPBin webserver. """ - executing_command = pyqtSignal() - setting_done = pyqtSignal() - url_loaded = pyqtSignal() got_error = pyqtSignal() def __init__(self, httpbin, parent=None): @@ -107,26 +108,12 @@ class QuteProc(testprocess.Process): ": LoadStatus.success") - url_loaded_pattern = re.compile( - r"load status for : LoadStatus.success") - if (log_line.category == 'ipc' and log_line.message.startswith("Listening as ")): self._ipc_socket = log_line.message.split(' ', maxsplit=2)[2] elif (log_line.category == 'webview' and log_line.message == start_okay_message): self.ready.emit() - elif (log_line.category == 'commands' and - log_line.module == 'command' and log_line.function == 'run' and - log_line.message.startswith('Calling ')): - self.executing_command.emit() - elif (log_line.category == 'config' and log_line.message.startswith( - 'Config option changed: ')): - self.setting_done.emit() - elif (log_line.category == 'webview' and - url_loaded_pattern.match(log_line.message)): - self.url_loaded.emit() elif log_line.loglevel in ['WARNING', 'ERROR']: self.got_error.emit() @@ -156,23 +143,29 @@ class QuteProc(testprocess.Process): def send_cmd(self, command): assert self._ipc_socket is not None - with self._wait_signal(self.executing_command): - ipc.send_to_running_instance(self._ipc_socket, [command], - target_arg='') + + ipc.send_to_running_instance(self._ipc_socket, [command], + target_arg='') + self.wait_for(category='commands', module='command', function='run', + message='Calling *') # Wait a bit in cause the command triggers any error. time.sleep(0.5) def set_setting(self, sect, opt, value): - with self._wait_signal(self.setting_done): - self.send_cmd(':set "{}" "{}" "{}"'.format(sect, opt, value)) + self.send_cmd(':set "{}" "{}" "{}"'.format(sect, opt, value)) + self.wait_for(category='config', message='Config option changed: *') def open_path(self, path, new_tab=False): + url_loaded_pattern = re.compile( + r"load status for : LoadStatus.success") + url = 'http://localhost:{}/{}'.format(self._httpbin.port, path) - with self._wait_signal(self.url_loaded): - if new_tab: - self.send_cmd(':open -t ' + url) - else: - self.send_cmd(':open ' + url) + if new_tab: + self.send_cmd(':open -t ' + url) + else: + self.send_cmd(':open ' + url) + self.wait_for(category='webview', message=url_loaded_pattern) def mark_expected(self, category=None, loglevel=None, msg=None): """Mark a given logging message as expected.""" diff --git a/tests/integration/testprocess.py b/tests/integration/testprocess.py index 1eef2eef2..070b971e9 100644 --- a/tests/integration/testprocess.py +++ b/tests/integration/testprocess.py @@ -19,8 +19,12 @@ """Base class for a subprocess run for tests..""" +import re +import fnmatch + import pytestqt.plugin # pylint: disable=import-error -from PyQt5.QtCore import pyqtSlot, pyqtSignal, QProcess, QObject +from PyQt5.QtCore import pyqtSlot, pyqtSignal, QProcess, QObject, QElapsedTimer +from PyQt5.QtTest import QSignalSpy class InvalidLine(Exception): @@ -37,6 +41,11 @@ class ProcessExited(Exception): pass +class WaitForTimeout(Exception): + + """Raised when wait_for didn't get the expected message.""" + + class Process(QObject): """Abstraction over a running test subprocess process. @@ -147,3 +156,47 @@ class Process(QObject): def is_running(self): """Check if the process is currently running.""" return self.proc.state() == QProcess.Running + + def wait_for(self, timeout=5000, **kwargs): + """Wait until a given value is found in the data. + + Keyword arguments to this function get interpreted as attributes of the + searched data. Every given argument is treated as a pattern which + the attribute has to match against. + + If a string is passed, the patterns are treated as a fnmatch glob + pattern. Alternatively, a compiled regex can be passed to match against + that. + """ + + # FIXME make this a context manager which inserts a marker in + # self._data in __enter__ and checks if the signal already did arrive + # after marker in __exit__, and if not, waits? + + regex_type = type(re.compile('')) + + spy = QSignalSpy(self.new_data) + elapsed_timer = QElapsedTimer() + elapsed_timer.start() + + while True: + got_signal = spy.wait(timeout) + if not got_signal or elapsed_timer.hasExpired(timeout): + raise WaitForTimeout + + for args in spy: + assert len(args) == 1 + line = args[0] + + matches = [] + + for key, expected in kwargs.items(): + value = getattr(line, key) + + if isinstance(expected, regex_type): + matches.append(expected.match(value)) + else: + matches.append(fnmatch.fnmatchcase(value, expected)) + + if all(matches): + return