Source code
Revision control
Copy as Markdown
Other Tools
# -*- coding: utf-8 -*-
"""
coroutine_tests
~~~~~~~~~~~~~~~
This file gives access to a coroutine-based test class. This allows each test
case to be defined as a pair of interacting coroutines, sending data to each
other by yielding the flow of control.
The advantage of this method is that we avoid the difficulty of using threads
in Python, as well as the pain of using sockets and events to communicate and
organise the communication. This makes the tests entirely deterministic and
makes them behave identically on all platforms, as well as ensuring they both
succeed and fail quickly.
"""
import itertools
import functools
import pytest
class CoroutineTestCase(object):
"""
A base class for tests that use interacting coroutines.
The run_until_complete method takes a number of coroutines as arguments.
Each one is, in order, passed the output of the previous coroutine until
one is exhausted. If a coroutine does not initially yield data (that is,
its first action is to receive data), the calling code should prime it by
using the 'server' decorator on this class.
"""
def run_until_complete(self, *coroutines):
"""
Executes a set of coroutines that communicate between each other. Each
one is, in order, passed the output of the previous coroutine until
one is exhausted. If a coroutine does not initially yield data (that
is, its first action is to receive data), the calling code should prime
it by using the 'server' decorator on this class.
Once a coroutine is exhausted, the method performs a final check to
ensure that all other coroutines are exhausted. This ensures that all
assertions in those coroutines got executed.
"""
looping_coroutines = itertools.cycle(coroutines)
data = None
for coro in looping_coroutines:
try:
data = coro.send(data)
except StopIteration:
break
for coro in coroutines:
try:
next(coro)
except StopIteration:
continue
else:
pytest.fail("Coroutine %s not exhausted" % coro)
def server(self, func):
"""
A decorator that marks a test coroutine as a 'server' coroutine: that
is, one whose first action is to consume data, rather than one that
initially emits data. The effect of this decorator is simply to prime
the coroutine.
"""
@functools.wraps(func)
def wrapper(*args, **kwargs):
c = func(*args, **kwargs)
next(c)
return c
return wrapper