pytest-trio: Pytest plugin for trio

This is a pytest plugin to help you test projects that use Trio, a friendly library for concurrency and async I/O in Python. Features include:

• Async tests without the boilerplate: just write async def test_whatever(): ....
• Useful fixtures included: use autojump_clock for easy testing of code with timeouts, or nursery to easily set up background tasks.
• Write your own async fixtures: set up an async database connection or start a server inside a fixture, and then use it in your tests.
• If you have multiple async fixtures, pytest-trio will even do setup/teardown concurrently whenever possible. (Though honestly, we’re not sure whether this is a good idea or not and might remove it in the future. If it makes your tests harder to debug, or conversely provides you with big speedups, please let us know.)
• Integration with the fabulous Hypothesis library, so your async tests can use property-based testing: just use @given like you’re used to.
• Support for testing projects that use Trio exclusively and want to use pytest-trio everywhere, and also for testing projects that support multiple async libraries and only want to enable pytest-trio’s features for a subset of their test suite.

Vital statistics

• Install: pip install pytest-trio
• Documentation: https://pytest-trio.readthedocs.io
• Issue tracker, source code: https://github.com/python-trio/pytest-trio
• License: MIT or Apache 2, your choice
• Contributor guide: https://trio.readthedocs.io/en/latest/contributing.html
• Code of conduct: Contributors are requested to follow our code of conduct in all project spaces.

Quickstart

Enabling Trio mode and running your first async tests

Note

If you used cookiecutter-trio to set up your project, then pytest-trio and Trio mode are already configured! You can write async def test_whatever(): ... and it should just work. Feel free to skip to the next section.

Let’s make a temporary directory to work in, and write two trivial tests: one that we expect should pass, and one that we expect should fail:

# test_example.py
import trio

async def test_sleep():
    start_time = trio.current_time()
    await trio.sleep(1)
    end_time = trio.current_time()
    assert end_time - start_time >= 1

async def test_should_fail():
    assert False

If we run this under pytest normally, then we get a strange result:

$ pytest test_example.py

======================== test session starts =========================
platform linux -- Python 3.6.5, pytest-3.6.3, py-1.5.4, pluggy-0.6.0
rootdir: /tmp, inifile:
collected 2 items

test_example.py ..                                             [100%]

========================== warnings summary ==========================
test_example.py::test_sleep
  .../_pytest/python.py:196: RuntimeWarning: coroutine 'test_sleep' was never awaited
    testfunction(**testargs)

test_example.py::test_should_fail
  .../_pytest/python.py:196: RuntimeWarning: coroutine 'test_should_fail' was never awaited
    testfunction(**testargs)

-- Docs: http://doc.pytest.org/en/latest/warnings.html
================ 2 passed, 2 warnings in 0.02 seconds ================

So test_sleep passed, which is what we expected… but test_should_fail also passes, which is strange. And it says that the whole test run completed in 0.02 seconds, which is weird, because test_sleep should have taken at least second to run. And then there are these strange warnings at the bottom… what’s going on here?

The problem is that our tests are async, and pytest doesn’t know what to do with it. So it basically skips running them entirely, and then reports them as passed. This is not very helpful! If you see warnings like this, or if your tests seem to pass but your coverage reports claim that they weren’t run at all, then this might be the problem.

Here’s the fix:

1. Install pytest-trio: pip install pytest-trio

2. In your project root, create a file called pytest.ini with contents:

   [pytest]
   trio_mode = true
   

And we’re done! Let’s try running pytest again:

$ pip install pytest-trio

$ cat <<EOF >pytest.ini
[pytest]
trio_mode = true
EOF

$ pytest test_example.py
======================== test session starts =========================
platform linux -- Python 3.6.5, pytest-3.6.3, py-1.5.4, pluggy-0.6.0
rootdir: /tmp, inifile: pytest.ini
plugins: trio-0.4.2
collected 2 items

test_example.py .F                                             [100%]

============================== FAILURES ==============================
__________________________ test_should_fail __________________________

    async def test_should_fail():
>       assert False
E       assert False

test_example.py:7: AssertionError
================= 1 failed, 1 passed in 1.05 seconds =================

Notice that now it says plugins: trio, which means that pytest-trio is installed, and the results make sense: the good test passed, the bad test failed, no warnings, and it took just over 1 second, like we’d expect.

Trio’s magic autojump clock

Tests involving time are often slow and flaky. But we can fix that. Just add the autojump_clock fixture to your test, and it will run in a mode where Trio’s clock is virtualized and deterministic. Essentially, the clock doesn’t move, except that whenever all tasks are blocked waiting, it jumps forward until the next time when something will happen:

# Notice the 'autojump_clock' argument: that's all it takes!
async def test_sleep_efficiently_and_reliably(autojump_clock):
    start_time = trio.current_time()
    await trio.sleep(1)
    end_time = trio.current_time()
    assert start_time - end_time == 1

In the version of this test we saw before that used real time, at the end we had to use a >= comparison, in order to account for scheduler jitter and so forth. If there were a bug that caused trio.sleep() to take 10 seconds, our test wouldn’t have noticed. But now we’re using virtual time, so the call to await trio.sleep(1) takes exactly 1 virtual second, and the == test will pass every time. Before, we had to wait around for the test to complete; now, it completes essentially instantaneously. (Try it!) And, while here our example is super simple, it’s integration with Trio’s core scheduling logic allows this to work for arbitrarily complex programs (as long as they aren’t interacting with the outside world).

Async fixtures

We can write async fixtures:

@pytest.fixture
async def db_connection():
    return await some_async_db_library.connect(...)

async def test_example(db_connection):
    await db_connection.execute("SELECT * FROM ...")

If you need to run teardown code, you can use yield, just like a regular pytest fixture:

# DB connection that wraps each test in a transaction and rolls it
# back afterwards
@pytest.fixture
async def rollback_db_connection():
    # Setup code
    connection = await some_async_db_library.connect(...)
    await connection.execute("START TRANSACTION")

    # The value of this fixture
    yield connection

    # Teardown code, executed after the test is done
    await connection.execute("ROLLBACK")

If you need to support Python 3.5, which doesn’t allow yield inside an async def function, then you can define async fixtures using the async_generator library – just make sure to put the @pytest.fixture above the @async_generator.

Running a background server from a fixture

Here’s some code to implement an echo server. It’s supposed to take in arbitrary data, and then send it back out again:

async def echo_server_handler(stream):
    while True:
        data = await stream.receive_some(1000)
        if not data:
            break
        await stream.send_all(data)

# Usage: await trio.serve_tcp(echo_server_handler, ...)

Now we need to test it, to make sure it’s working correctly. In fact, since this is such complicated and sophisticated code, we’re going to write lots of tests for it. And they’ll all follow the same basic pattern: we’ll start the echo server running in a background task, then connect to it, send it some test data, and see how it responds. Here’s a first attempt:

# Let's cross our fingers and hope no-one else is using this port...
PORT = 14923

# Don't copy this -- we can do better
async def test_attempt_1():
    async with trio.open_nursery() as nursery:
        # Start server running in the background
        nursery.start_soon(
            partial(trio.serve_tcp, echo_server_handler, port=PORT)
        )

        # Connect to the server.
        echo_client = await trio.open_tcp_stream("127.0.0.1", PORT)
        # Send some test data, and check that it gets echoed back
        async with echo_client:
            for test_byte in [b"a", b"b", b"c"]:
                await echo_client.send_all(test_byte)
                assert await echo_client.receive_some(1) == test_byte

This will mostly work, but it has a few problems. The most obvious one is that when we run it, even if everything works perfectly, it will hang at the end of the test – we never shut down the server, so the nursery block will wait forever for it to exit.

To avoid this, we should cancel the nursery at the end of the test:

# Let's cross our fingers and hope no-one else is using this port...
PORT = 14923

# Don't copy this -- we can do better
async def test_attempt_2():
    async with trio.open_nursery() as nursery:
        try:
            # Start server running in the background
            nursery.start_soon(
                partial(trio.serve_tcp, echo_server_handler, port=PORT)
            )

            # Connect to the server.
            echo_client = await trio.open_tcp_stream("127.0.0.1", PORT)
            # Send some test data, and check that it gets echoed back
            async with echo_client:
                for test_byte in [b"a", b"b", b"c"]:
                    await echo_client.send_all(test_byte)
                    assert await echo_client.receive_some(1) == test_byte
        finally:
            nursery.cancel_scope.cancel()

In fact, this pattern is so common, that pytest-trio provides a handy nursery fixture to let you skip the boilerplate. Just add nursery to your test function arguments, and pytest-trio will open a nursery, pass it in to your function, and then cancel it for you afterwards:

# Let's cross our fingers and hope no-one else is using this port...
PORT = 14923

# Don't copy this -- we can do better
async def test_attempt_3(nursery):
    # Start server running in the background
    nursery.start_soon(
        partial(trio.serve_tcp, echo_server_handler, port=PORT)
    )

    # Connect to the server.
    echo_client = await trio.open_tcp_stream("127.0.0.1", PORT)
    # Send some test data, and check that it gets echoed back
    async with echo_client:
        for test_byte in [b"a", b"b", b"c"]:
            await echo_client.send_all(test_byte)
            assert await echo_client.receive_some(1) == test_byte

Next problem: we have a race condition. We spawn a background task to call serve_tcp, and then immediately try to connect to that server. Sometimes this will work fine. But it takes a little while for the server to start up and be ready to accept connections – so other times, randomly, our connection attempt will happen too quickly, and error out. After all – nursery.start_soon only promises that the task will be started soon, not that it’s actually happened. So this test will be flaky, and flaky tests are the worst.

Fortunately, Trio makes this easy to solve, by switching to using await nursery.start(...). You can read its docs for full details, but basically the idea is that both nursery.start_soon(...) and await nursery.start(...) create background tasks, but only start waits for the new task to finish getting itself set up. This requires some cooperation from the background task: it has to notify nursery.start when it’s ready. Fortunately, trio.serve_tcp() already knows how to cooperate with nursery.start, so we can write:

# Let's cross our fingers and hope no-one else is using this port...
PORT = 14923

# Don't copy this -- we can do better
async def test_attempt_4(nursery):
    # Start server running in the background
    # AND wait for it to finish starting up before continuing
    await nursery.start(
        partial(trio.serve_tcp, echo_server_handler, port=PORT)
    )

    # Connect to the server
    echo_client = await trio.open_tcp_stream("127.0.0.1", PORT)
    async with echo_client:
        for test_byte in [b"a", b"b", b"c"]:
            await echo_client.send_all(test_byte)
            assert await echo_client.receive_some(1) == test_byte

That solves our race condition. Next issue: hardcoding the port number like this is a bad idea, because port numbers are a machine-wide resource, so if we’re unlucky some other program might already be using it. What we really want to do is to tell serve_tcp() to pick a random port that no-one else is using. It turns out that this is easy: if you request port 0, then the operating system will pick an unused one for you automatically. Problem solved!

But wait… if the operating system is picking the port for us, how do we know figure out which one it picked, so we can connect to it later?

Well, there’s no way to predict the port ahead of time. But after serve_tcp() has opened a port, it can check and see what it got. So we need some way to pass this data back out of serve_tcp(). Fortunately, nursery.start handles this too: it lets the task pass out a piece of data after it’s started. And it just so happens that what serve_tcp() passes out is a list of SocketListener objects. And there’s a handy function called trio.testing.open_stream_to_socket_listener() that can take a SocketListener and make a connection to it.

Putting it all together:

from trio.testing import open_stream_to_socket_listener

# Don't copy this -- it finally works, but we can still do better!
async def test_attempt_5(nursery):
    # Start server running in the background
    # AND wait for it to finish starting up before continuing
    # AND find out where it's actually listening
    listeners = await nursery.start(
        partial(trio.serve_tcp, echo_server_handler, port=0)
    )

    # Connect to the server.
    # There might be multiple listeners (example: IPv4 and
    # IPv6), but we don't care which one we connect to, so we
    # just use the first.
    echo_client = await open_stream_to_socket_listener(listeners[0])
    async with echo_client:
        for test_byte in [b"a", b"b", b"c"]:
            await echo_client.send_all(test_byte)
            assert await echo_client.receive_some(1) == test_byte

Now, this works – but there’s still a lot of boilerplate. Remember, we need to write lots of tests for this server, and we don’t want to have to copy-paste all that stuff into every test. Let’s factor out the setup into a fixture:

@pytest.fixture
async def echo_client(nursery):
    listeners = await nursery.start(
        partial(trio.serve_tcp, echo_server_handler, port=0)
    )
    echo_client = await open_stream_to_socket_listener(listeners[0])
    async with echo_client:
        yield echo_client

And now in tests, all we have to do is request the echo_client fixture, and we get a background server and a client stream connected to it. So here’s our complete, final version:

# Final version -- copy this!
from functools import partial
import pytest
import trio
from trio.testing import open_stream_to_socket_listener

# The code being tested:
async def echo_server_handler(stream):
    while True:
        data = await stream.receive_some(1000)
        if not data:
            break
        await stream.send_all(data)

# The fixture:
@pytest.fixture
async def echo_client(nursery):
    listeners = await nursery.start(
        partial(trio.serve_tcp, echo_server_handler, port=0)
    )
    echo_client = await open_stream_to_socket_listener(listeners[0])
    async with echo_client:
        yield echo_client

# A test using the fixture:
async def test_final(echo_client):
    for test_byte in [b"a", b"b", b"c"]:
        await echo_client.send_all(test_byte)
        assert await echo_client.receive_some(1) == test_byte

No hangs, no race conditions, simple, clean, and reusable.

Reference

Trio mode

Most users will want to enable “Trio mode”. Without Trio mode:

• Pytest-trio only handles tests that have been decorated with @pytest.mark.trio
• Pytest-trio only handles fixtures if they’re async and used by a test that’s decorated with @pytest.mark.trio, or if they’re decorated with @pytest_trio.trio_fixture (instead of @pytest.fixture).

When Trio mode is enabled, two extra things happen:

• Async tests automatically have the trio mark added, so you don’t have to do it yourself.
• Async fixtures using @pytest.fixture automatically get converted to Trio fixtures. (The main effect of this is that it helps you catch mistakes where a you use an async fixture with a non-async test.)

There are two ways to enable Trio mode.

The first option is to use a pytest configuration file. The exact rules for how pytest finds configuration files are a bit complicated, but you want to end up with something like:

# pytest.ini
[pytest]
trio_mode = true

The second option is use a conftest.py file. Inside your tests directory, create a file called conftest.py, with the following contents:

# conftest.py
from pytest_trio.enable_trio_mode import *

This does exactly the same thing as setting trio_mode = true in pytest.ini, except for two things:

• Some people like to ship their tests as part of their library, so they (or their users) can test the final installed software by running pytest --pyargs PACKAGENAME. In this mode, pytest.ini files don’t work, but conftest.py files do.
• Enabling Trio mode in pytest.ini always enables it globally for your entire testsuite. Enabling it in conftest.py only enables it for test files that are in the same directory as the conftest.py, or its subdirectories.

If you have software that uses multiple async libraries, then you can use conftest.py to enable Trio mode for just the part of your testsuite that uses Trio; or, if you need even finer-grained control, you can leave Trio mode disabled and use @pytest.mark.trio explicitly on all your Trio tests.

Trio fixtures

Normally, pytest runs fixture code before starting the test, and teardown code afterwards. For technical reasons, we can’t wrap this whole process in trio.run() – only the test itself. As a workaround, pytest-trio introduces the concept of a “Trio fixture”, which acts like a normal fixture for most purposes, but actually does the setup and teardown inside the test’s call to trio.run().

The following fixtures are treated as Trio fixtures:

• Any function decorated with @pytest_trio.trio_fixture.
• Any async function decorated with @pytest.fixture, if Trio mode is enabled or this fixture is being requested by a Trio test.
• Any fixture which depends on a Trio fixture.

The most notable difference between regular fixtures and Trio fixtures is that regular fixtures can’t use Trio APIs, but Trio fixtures can. Most of the time you don’t need to worry about this, because you normally only call Trio APIs from async functions, and when Trio mode is enabled, all async fixtures are automatically Trio fixtures. However, if for some reason you do want to use Trio APIs from a synchronous fixture, then you’ll have to use @pytest_trio.trio_fixture:

# This fixture is not very useful
# But it is an example where @pytest.fixture doesn't work
@pytest_trio.trio_fixture
def trio_time():
    return trio.current_time()

Only Trio tests can use Trio fixtures. If you have a regular (synchronous) test that tries to use a Trio fixture, then that’s an error.

And finally, regular fixtures can be scoped to the test, class, module, or session, but Trio fixtures must be test scoped. Class, module, and session scope are not supported.

Concurrent setup/teardown

If your test uses multiple fixtures, then for speed, pytest-trio will try to run their setup and teardown code concurrently whenever this is possible while respecting the fixture dependencies.

Here’s an example, where a test depends on fix_b and fix_c, and these both depend on fix_a:

@trio_fixture
def fix_a():
    ...

@trio_fixture
def fix_b(fix_a):
    ...

@trio_fixture
def fix_c(fix_a):
    ...

@pytest.mark.trio
async def test_example(fix_b, fix_c):
    ...

When running test_example, pytest-trio will perform the following sequence of actions:

1. Set up fix_a
2. Set up fix_b and fix_c, concurrently.
3. Run the test.
4. Tear down fix_b and fix_c, concurrently.
5. Tear down fix_a.

We’re seeking feedback on whether this feature’s benefits outweigh its negatives.

Handling of ContextVars

The contextvars module lets you create ContextVar objects to represent task-local variables. Normally, in Trio, each task gets its own Context, so that changes to ContextVar objects are only visible inside the task that performs them. But pytest-trio overrides this, and for each test it uses a single Context which is shared by all fixtures and the test function itself.

The benefit of this is that you can set ContextVar values inside a fixture, and your settings will be visible in dependent fixtures and the test itself. For example, trio-asyncio uses a ContextVar to hold the current asyncio loop object, so this lets you open a loop inside a fixture and then use it inside other fixtures or the test itself.

The downside is that if two fixtures are run concurrently (see previous section), and both mutate the same ContextVar, then there will be a race condition and the the final value will be unpredictable. If you make one fixture depend on the other, then this will force an ordering and make the final value predictable again.

Built-in fixtures

These fixtures are automatically available to any code using pytest-trio.

autojump_clock

A trio.testing.MockClock, configured with rate=0, autojump_threshold=0.

mock_clock

A trio.testing.MockClock, with its default configuration (rate=0, autojump_threshold=inf).

What makes these particularly useful is that whenever pytest-trio runs a test, it checks the fixtures to see if one of them is a trio.abc.Clock object. If so, it passes that object to trio.run(). So if your test requests one of these fixtures, it automatically uses that clock.

If you implement your own Clock, and implement a fixture that returns it, then it will work the same way.

Of course, like any pytest fixture, you also get the actual object available. For example, you can call jump():

async def test_time_travel(mock_clock):
    assert trio.current_time() == 0
    mock_clock.jump(10)
    assert trio.current_time() == 10

nursery

A nursery created and managed by pytest-trio itself, which surrounds the test/fixture that requested it, and is automatically cancelled after the test/fixture completes. Basically, these are equivalent:

# Boring way
async def test_with_background_task():
    async with trio.open_nursery() as nursery:
        try:
            ...
        finally:
            nursery.cancel_scope.cancel()

# Fancy way
async def test_with_background_task(nursery):
    ...

For a fixture, the cancellation always happens after the fixture completes its teardown phase. (Or if it doesn’t have a teardown phase, then the cancellation happens after the teardown phase would have happened.)

This fixture is even more magical than most pytest fixtures, because if it gets requested several times within the same test, then it creates multiple nurseries, one for each fixture/test that requested it.

See Running a background server from a fixture for an example of how this can be used.

Integration with the Hypothesis library

There isn’t too much to say here, since the obvious thing just works:

from hypothesis import given
import hypothesis.strategies as st

@given(st.binary())
async def test_trio_and_hypothesis(data):
    ...

Under the hood, this requires some coordination between Hypothesis and pytest-trio. Hypothesis runs your test multiple times with different examples of random data. For each example, pytest-trio calls trio.run() again (so you get a fresh clean Trio environment), sets up any Trio fixtures, runs the actual test, and then tears down any Trio fixtures. Notice that this is a bit different than regular pytest fixtures, which are instantiated once and then re-used for all. Most of the time this shouldn’t matter (and is probably what you want anyway), but in some unusual cases it could surprise you. And this only applies to Trio fixtures – if a Trio test uses a mix of regular fixtures and Trio fixtures, then the regular fixtures will be reused, while the Trio fixtures will be repeatedly reinstantiated.

Also, pytest-trio only handles @given-based tests. If you want to write stateful tests for Trio-based libraries, then check out hypothesis-trio.

Release history

pytest-trio 0.5.0 (2018-08-26)

This is a major release, including a rewrite of large portions of the internals. We believe it should be backwards compatible with existing projects. Major new features include:

• “trio mode”: no more writing @pytest.mark.trio everywhere!
• it’s now safe to use nurseries inside fixtures (#55)
• new @trio_fixture decorator to explicitly mark a fixture as a trio fixture
• a number of easy-to-make mistakes are now caught and raise informative errors
• the nursery fixture is now 87% more magical

For more details, see the manual. Oh right, speaking of which: we finally have a manual! You should read it.

pytest-trio 0.4.2 (2018-06-29)

Features

• pytest-trio now integrates with Hypothesis to support @given on async tests using Trio. (#42)

pytest-trio 0.4.1 (2018-04-14)

No significant changes.

pytest-trio 0.4.0 (2018-04-14)

• Fix compatibility with trio 0.4.0 (#25)

pytest-trio 0.3.0 (2018-01-03)

Features

• Add nursery fixture and improve teardown handling for yield fixture (#25)

pytest-trio 0.2.0 (2017-12-15)

• Heavy improvements, add async yield fixture, fix bugs, add tests etc. (#17)

Deprecations and Removals

• Remove unused_tcp_port{,_factory} fixtures (#15)

pytest-trio 0.1.1 (2017-12-08)

Disable intersphinx for trio (cause crash in CI for the moment due to 404 in readthedoc).

pytest-trio 0.1.0 (2017-12-08)

Initial release.

Indices and tables

• Index
• Module Index
• Search Page
• Glossary