tests: Don't wait for the same line twice.

We need to search for lines in the history because we could miss something
otherwise, but for subsequent wait_for calls, we really don't want to wait for
the same thing again.

This should make test_backforward.py more stable as it *actually* waits when
going back now. Before, it did produce failures such as this one on OS X:

____________________________ test_going_backforward ____________________________
[..]
    @bdd.then(bdd.parsers.parse("The requests should be:\n{pages}"))
    def list_of_loaded_pages(httpbin, pages):
        requests = [httpbin.Request('GET', '/' + path.strip())
                    for path in pages.split('\n')]
>       assert httpbin.get_requests() == requests
E       assert [Request(verb...rward/1.txt')] == [Request(verb=...rward/2.txt')]
E         At index 3 diff: Request(verb='GET', path='/data/backforward/1.txt') != Request(verb='GET', path='/data/backforward/2.txt')
E         Full diff:
E         [Request(verb='GET', path='/data/backforward/1.txt'),
E         Request(verb='GET', path='/data/backforward/2.txt'),
E         Request(verb='GET', path='/data/backforward/1.txt'),
E         -  Request(verb='GET', path='/data/backforward/1.txt')]
E         ?                                              ^
E         +  Request(verb='GET', path='/data/backforward/2.txt')]
E         ?                                              ^
tests/integration/features/conftest.py:85: AssertionError
This commit is contained in:
Florian Bruhin 2015-11-06 06:57:38 +01:00
parent f5f74b7ddc
commit 25ecd9068c
2 changed files with 32 additions and 6 deletions

View File

@ -112,9 +112,25 @@ class TestWaitFor:
def test_existing_message_previous_test(self, pyproc):
"""Make sure the message of a previous test gets ignored."""
pyproc.code = "time.sleep(0.1); print('foobar')"
pyproc.code = "print('foobar')"
pyproc.start()
pyproc.wait_for(data="foobar")
# We can't use wait_for here, as that'd actually test what the next
# test does.
time.sleep(0.5)
pyproc.after_test()
with pytest.raises(testprocess.WaitForTimeout):
pyproc.wait_for(data="foobar", timeout=100)
def test_existing_message_already_waited(self, pyproc):
"""Make sure an existing message doesn't stop waiting twice.
wait_for checks existing messages (see above), but we don't want it to
automatically proceed if we already *did* use wait_for on one of the
existing messages, as that makes it likely it's not what we actually
want.
"""
pyproc.code = "time.sleep(0.1); print('foobar')"
pyproc.start()
pyproc.wait_for(data="foobar")
with pytest.raises(testprocess.WaitForTimeout):
pyproc.wait_for(data="foobar", timeout=100)

View File

@ -52,10 +52,12 @@ class Line:
Attributes:
data: The raw data passed to the constructor.
waited_for: If Process.wait_for was used on this line already.
"""
def __init__(self, data):
self.data = data
self.waited_for = False
def __repr__(self):
return '{}({!r})'.format(self.__class__.__name__, self.data)
@ -209,15 +211,19 @@ class Process(QObject):
The matched line.
"""
# Search existing messages
for item in self._data:
for line in self._data:
matches = []
for key, expected in kwargs.items():
value = getattr(item, key)
value = getattr(line, key)
matches.append(self._match_data(value, expected))
if all(matches):
return item
if all(matches) and not line.waited_for:
# If we waited for this line, chances are we don't mean the
# same thing the next time we use wait_for and it matches
# this line again.
line.waited_for = True
return line
# If there is none, wait for the message
spy = QSignalSpy(self.new_data)
@ -241,4 +247,8 @@ class Process(QObject):
matches.append(self._match_data(value, expected))
if all(matches):
# If we waited for this line, chances are we don't mean the
# same thing the next time we use wait_for and it matches
# this line again.
line.waited_for = True
return line