Pytest is an awesome testing package for Python and since long one of my favorite Python packages in general. It makes writing test really easy and its reporting capabilities for test failures are extremely helpful.

However, it currently (as of version 2.7) doesn’t help you very much with testing (asyncio) coroutines. So a naïve approach for testing coroutines would be:

# tests/test_coros.py

import asyncio

def test_coro():
    loop = asyncio.get_event_loop()

    @asyncio.coroutine
    def do_test():
        yield from asyncio.sleep(0.1, loop=loop)
        assert 0  # onoes!

    loop.run_until_complete(do_test())

This approach has several problems and a lot of overhead. The only interesting lines are the ones containing the yield from and assert statements.

It would be better if every test case had its own event loop instance that gets correctly closed no matter if the test passes or fails.

We also cannot simply yield within our test case because pytest would then think that our test case yields new test cases which is not the case here. So we have to create a separate coroutine which contains our actual test and fire up the even loop in order to execute it.

Our tests would look a lot cleaner and behave better if we could instead do something like this:

# tests/test_coros.py

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

It turns out that thanks to pytests flexible plug-in system, it’s possible to implement the desired behavior. Although, most of the required hooks are not documented very well or at all which also makes it relatively hard to find out which hooks you have to implement and how.

We create a local per-directory plug-in since this is a little bit easier than creating a “real”, external plug-in. Pytest looks in every test-directory for a file called conftest.py and applies the fixtures and hooks implemented there to all tests within that directory.

So lets start by writing a fixture that creates a new event loop instance for each test case and properly close it when the test is done:

# tests/conftest.py

import asyncio
import pytest


@pytest.yield_fixture
def loop():
    # Set-up
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop

    # Clean-up
    loop.close()


# tests/test_coros.py

def test_coro(loop):
    @asyncio.coroutine
    def do_test():
        yield from asyncio.sleep(0.1, loop=loop)
        assert 0  # onoes!

    loop.run_until_complete(do_test())

Before each test, pytest executes the loop fixture until the first yield statement. What gets yielded is then passed to the loop argument of our test case. When the test ends (successfully or not), pytest finishes the execution of the loop fixture and thus closes the loop properly. In the same way, you could write a fixture that creates a socket and closes it after each test (your socket fixture can depend on the loop fixture in the same way as our test does. Nice, isn’t it?).

But we are still not done yet. Let’s teach pytest how to execute our test coroutines. Therefore, we need to change how asyncio coroutines are collected (they should be collected like normal test functions, not like test generators) and how they get executed (via loop.run_until_complete()):

# tests/conftest.py

def pytest_pycollect_makeitem(collector, name, obj):
    """Collect asyncio coroutines as normal functions, not as generators."""
    if collector.funcnamefilter(name) and asyncio.iscoroutinefunction(obj):
        # We return a list of test function objects.  Depending on the
        # fixtures, one test function can result in multiple test items
        # (i.e., when it is decorated with "pytest.mark.parametrize()".
        return list(collector._genfunctions(name, obj))
    # else:
    #     We return None and pytest's default behavior gets applied to "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 None if its not a coroutine.  Pytest will handle the
        # test item in the normal way.
        return

    # Extract required arguments from all available fixtures:
    funcargs = pyfuncitem.funcargs  # dict with all fixtures
    argnames = pyfuncitem._fixtureinfo.argnames  # Args for this test
    testargs = {arg: funcargs[arg] for arg in argnames}

    # Create the generator object for this test (test will not yet execute!)
    coro = testfunction(**testargs)

    # Run the coro in the event loop and execute the test
    loop = testargs['loop'] if loop in testargs else asyncio.get_event_loop()
    loop.run_until_complete(coro)

    return True  # Signal pytest that we executed the test

This plug-in should work with pytest 2.4 or newer. I tested it with versions 2.6 and 2.7.

It seems like the pytest devs are currently trying to clean-up and refactor pytest’s plug-in system. Maybe, a future release will even contain similar functionality to our plug-in. It would be really nice if pytest supported testing coroutines out-of-the-box as they are becoming increasingly popular amongst various frameworks.

Update: There’s also pytest-asyncio . It was created after I posted the solution described above to Stack Overflow.