py.test mixing fixtures and asyncio coroutines

Question

I am building some tests for python3 code using py.test. The code accesses a Postgresql Database using aiopg (Asyncio based interface to postgres).

My main expectations:

• Every test case should have access to a new asyncio event loop.

• A test that runs too long will stop with a timeout exception.

• Every test case should have access to a database connection.

• I don't want to repeat myself when writing the test cases.

Using py.test fixtures I can get pretty close to what I want, but I still have to repeat myself a bit in every asynchronous test case.

This is how my code looks like:

@pytest.fixture(scope='function')
def tloop(request):
    # This fixture is responsible for getting a new event loop
    # for every test, and close it when the test ends.
    ...

def run_timeout(cor,loop,timeout=ASYNC_TEST_TIMEOUT):
    """
    Run a given coroutine with timeout.
    """
    task_with_timeout = asyncio.wait_for(cor,timeout)
    try:
        loop.run_until_complete(task_with_timeout)
    except futures.TimeoutError:
        # Timeout:
        raise ExceptAsyncTestTimeout()


@pytest.fixture(scope='module')
def clean_test_db(request):
    # Empty the test database.
    ...

@pytest.fixture(scope='function')
def udb(request,clean_test_db,tloop):
    # Obtain a connection to the database using aiopg
    # (That's why we need tloop here).
    ...


# An example for a test:
def test_insert_user(tloop,udb):
    @asyncio.coroutine
    def insert_user():
        # Do user insertion here ...
        yield from udb.insert_new_user(...
        ...

    run_timeout(insert_user(),tloop)

I can live with the solution that I have so far, but it can get cumbersome to define an inner coroutine and add the run_timeout line for every asynchronous test that I write.

I want my tests to look somewhat like this:

@some_magic_decorator
def test_insert_user(udb):
    # Do user insertion here ...
    yield from udb.insert_new_user(...
    ...

I attempted to create such a decorator in some elegant way, but failed. More generally, if my test looks like:

@some_magic_decorator
def my_test(arg1,arg2,...,arg_n):
    ...

Then the produced function (After the decorator is applied) should be:

def my_test_wrapper(tloop,arg1,arg2,...,arg_n):
    run_timeout(my_test(),tloop)

Note that some of my tests use other fixtures (besides udb for example), and those fixtures must show up as arguments to the produced function, or else py.test will not invoke them.

I tried using both wrapt and decorator python modules to create such a magic decorator, however it seems like both of those modules help me create a function with a signature identical to my_test, which is not a good solution in this case.

This can probably solved using eval or a similar hack, but I was wondering if there is something elegant that I'm missing here.

Answer 1

I’m currently trying to solve a similar problem. Here’s what I’ve come up with so far. It seems to work but needs some clean-up:

# tests/test_foo.py
import asyncio

@asyncio.coroutine
def test_coro(loop):
    yield from asyncio.sleep(0.1)
    assert 0

# tests/conftest.py
import asyncio


@pytest.yield_fixture
def loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop
    loop.close()


def pytest_pycollect_makeitem(collector, name, obj):
    """Collect asyncio coroutines as normal functions, not as generators."""
    if asyncio.iscoroutinefunction(obj):
        return list(collector._genfunctions(name, obj))


def pytest_pyfunc_call(pyfuncitem):
    """If ``pyfuncitem.obj`` is an asyncio coroutinefunction, execute it via
    the event loop instead of calling it directly."""
    testfunction = pyfuncitem.obj

    if not asyncio.iscoroutinefunction(testfunction):
        return

    # Copied from _pytest/python.py:pytest_pyfunc_call()
    funcargs = pyfuncitem.funcargs
    testargs = {}
    for arg in pyfuncitem._fixtureinfo.argnames:
        testargs[arg] = funcargs[arg]
    coro = testfunction(**testargs)  # Will no execute the test yet!

    # Run the coro in the event loop
    loop = testargs.get('loop', asyncio.get_event_loop())
    loop.run_until_complete(coro)

    return True  # TODO: What to return here?

So I basically let pytest collect asyncio coroutines like normal functions. I also intercept text exectuion for functions. If the to-be-tested function is a coroutine, I execute it in the event loop. It works with or without a fixture creating a new event loop instance per test.

Edit: According to Ronny Pfannschmidt, something like this will be added to pytest after the 2.7 release. :-)