From 924b0052c6f2407dd4458b6cc88195075cca4ca9 Mon Sep 17 00:00:00 2001 From: Florian Bruhin Date: Thu, 5 Nov 2015 06:37:35 +0100 Subject: [PATCH] bdd: Match historic messages in wait_for. Fixes #1083. --- tests/integration/quteprocess.py | 25 +++++-------- tests/integration/test_testprocess.py | 17 +++++++++ tests/integration/testprocess.py | 53 +++++++++++++++++---------- 3 files changed, 59 insertions(+), 36 deletions(-) diff --git a/tests/integration/quteprocess.py b/tests/integration/quteprocess.py index c5d839dc2..8e95ef39e 100644 --- a/tests/integration/quteprocess.py +++ b/tests/integration/quteprocess.py @@ -205,24 +205,17 @@ class QuteProc(testprocess.Process): def mark_expected(self, category=None, loglevel=None, message=None): """Mark a given logging message as expected.""" - found_message = False + line = self.wait_for(category=category, loglevel=loglevel, + message=message) + line.expected = True - # Search existing messages - for item in self._data: - if category is not None and item.category != category: - continue - elif loglevel is not None and item.loglevel != loglevel: - continue - elif message is not None and item.message != message: - continue - item.expected = True - found_message = True + def wait_for(self, timeout=15000, **kwargs): + """Override testprocess.wait_for to check past messages. - # If there is none, wait for the message - if not found_message: - line = self.wait_for(category=category, loglevel=loglevel, - message=message) - line.expected = True + self._data is cleared after every test to provide at least some + isolation. + """ + return super().wait_for(timeout, **kwargs) def get_session(self): """Save the session and get the parsed session data.""" diff --git a/tests/integration/test_testprocess.py b/tests/integration/test_testprocess.py index 13b705fe1..1442802f1 100644 --- a/tests/integration/test_testprocess.py +++ b/tests/integration/test_testprocess.py @@ -20,6 +20,7 @@ """Test testprocess.Process.""" import sys +import time import contextlib import datetime @@ -110,3 +111,19 @@ class TestWaitFor: pyproc.start() with pytest.raises(testprocess.WaitForTimeout): pyproc.wait_for(data="foobar", timeout=100) + + def test_existing_message(self, pyproc): + """Test with a message which already passed when waiting.""" + pyproc.code = "print('foobar')" + pyproc.start() + time.sleep(0.5) # to make sure the message is printed + pyproc.wait_for(data="foobar") + + def test_existing_message_previous_test(self, pyproc): + """Make sure the message of a previous test gets ignored.""" + pyproc.code = "print('foobar')" + pyproc.start() + time.sleep(0.5) # to make sure the message is printed + pyproc.after_test() + with pytest.raises(testprocess.WaitForTimeout): + pyproc.wait_for(data="foobar", timeout=100) diff --git a/tests/integration/testprocess.py b/tests/integration/testprocess.py index c7b39b4a3..67cd6095b 100644 --- a/tests/integration/testprocess.py +++ b/tests/integration/testprocess.py @@ -157,32 +157,54 @@ class Process(QObject): """Check if the process is currently running.""" return self.proc.state() == QProcess.Running - def wait_for(self, timeout=15000, **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. + def _match_data(self, value, expected): + """Helper for wait_for to match a given value. The behavior of this method is slightly different depending on the types of the filtered values: + - If expected is None, the filter always matches. - If the value is a string or bytes object and the expected value is too, the pattern is treated as a fnmatch glob pattern. - If the value is a string or bytes object and the expected value is a compiled regex, it is used for matching. - If the value is any other type, == is used. + Return: + A bool + """ + regex_type = type(re.compile('')) + if expected is None: + return True + elif isinstance(expected, regex_type): + return expected.match(value) + elif isinstance(value, (bytes, str)): + return fnmatch.fnmatchcase(value, expected) + else: + return value == expected + + def wait_for(self, timeout=15000, **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. + Return: The matched line. """ + # Search existing messages + for item in self._data: + matches = [] - # 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? + for key, expected in kwargs.items(): + value = getattr(item, key) + matches.append(self._match_data(value, expected)) - regex_type = type(re.compile('')) + if all(matches): + return item + # If there is none, wait for the message spy = QSignalSpy(self.new_data) elapsed_timer = QElapsedTimer() elapsed_timer.start() @@ -200,17 +222,8 @@ class Process(QObject): matches = [] for key, expected in kwargs.items(): - if expected is None: - continue - value = getattr(line, key) - - if isinstance(expected, regex_type): - matches.append(expected.match(value)) - elif isinstance(value, (bytes, str)): - matches.append(fnmatch.fnmatchcase(value, expected)) - else: - matches.append(value == expected) + matches.append(self._match_data(value, expected)) if all(matches): return line