Chapter 15

Async tests, hooks and writing a plugin

Testing async code with pytest-asyncio, then extending pytest with hooks in conftest.py — a --runslow flag, a marker, a report header — and finally turning that into a plugin of your own and testing it with pytester.

60 minPython 3.12
  1. 1Encounter
  2. 2Understand
  3. 3Worked
  4. 4Predict
  5. 5Apply
  6. 6Stretch

The problem we are solving

Here is a small piece of asynchronous code — the kind that talks to a network and does not want to sit idle while it waits.

prices.py:

python
import asyncio

PRICES = {"pen": 15.0, "notebook": 60.0}


async def fetch_price(item):
    await asyncio.sleep(0.01)  # stands in for a network call
    if item not in PRICES:
        raise KeyError(item)
    return PRICES[item]

You test it the way you have tested everything so far: a function whose name starts with test_. It is an async function, because await is only allowed inside one. And to be sure the test is really checking something, the expected price is deliberately wrong:

test_prices.py:

python
from prices import fetch_price


async def test_pen_price():
    assert await fetch_price("pen") == 999.0

pytest -q:

text
F                                                                        [100%]
=================================== FAILURES ===================================
________________________________ test_pen_price ________________________________
async def functions are not natively supported.
You need to install a suitable plugin for your async framework, for example:
  - anyio
  - pytest-asyncio
  - pytest-tornasync
  - pytest-trio
  - pytest-twisted
=========================== short test summary info ============================
FAILED test_prices.py::test_pen_price - Failed: async def functions are not n...
1 failed in 0.11s

Read the failure carefully. It says nothing about 15.0 or 999.0. The assertion never ran. Calling an async def function does not run its body — it only creates a coroutine, and something has to drive an event loop to run it. Plain pytest does not do that. (Older versions of pytest skipped such a test with a warning, so it looked green while testing nothing. pytest 9 fails it, which is much better.)

That is the first half of this chapter: getting pytest to run async code. The second half is the deeper idea behind the fix — pytest itself is a set of hooks, and a plugin is just code that fills some of them. Once you see that, you can write your own: a --runslow flag, a marker, a line in the report header, and a test suite for the plugin itself.

By the end of this chapter you can

  • Explain why a bare async def test_... does not run, and recognise the failure
  • Run async tests with pytest-asyncio, using @pytest.mark.asyncio or asyncio_mode = "auto"
  • Write async fixtures, including ones with teardown after yield
  • Add a command-line option with pytest_addoption and read it with request.config.getoption
  • Skip slow tests unless --runslow is given, using pytest_collection_modifyitems
  • Register a marker in pytest_configure and add a line with pytest_report_header
  • Say where pytest finds hooks: conftest.py, -p, and pytest11 entry points
  • Test a plugin with the pytester fixture

Prerequisites: coverage and CI.


Before you write the test

This chapter tests two different kinds of thing — an async function and a pytest plugin — and both need some thinking before the first line of test code.

What is promised. Write the contract down in plain words first, because the tests are that contract, translated.

  • fetch_price(item) is a coroutine function. Awaited with a known item, it gives that item's price. Awaited with an unknown item, it raises KeyError. Many calls awaited together should overlap, not queue.
  • The slow-test plugin promises: without --runslow, every test marked slow is skipped, with a reason; with --runslow, everything runs; the header says which; and the slow marker is registered, so it raises no warning.

What must be in place. Async tests need a runner, which pytest does not have: install pytest-asyncio and decide on strict or auto mode before writing tests, because the decorators you write depend on it. The code under test has to be importable — prices.py at the project root, and later the plugin module too (pythonpath = ["."]). Testing the plugin needs the pytester fixture switched on. And nothing should need a real network: asyncio.sleep stands in for one.

The plan. One row per behaviour, each with its input and what you expect to see:

| Case | Input | Expected | | --- | --- | --- | | known item | await fetch_price("pen") | 15.0 | | unknown item | await fetch_price("stapler") | KeyError | | forgot await | fetch_price("pen") in a plain test | a coroutine object, not a price | | many at once | 400 calls through asyncio.gather | all 400 back, in about the time of one | | default run | pytest with one slow test | 1 passed, 1 skipped | | flag given | pytest --runslow | 2 passed | | header | pytest | a line slow tests: skipped (use --runslow) | | marker | pytest --markers | slow listed, no PytestUnknownMarkWarning |

What not to test. Not asyncio.sleep, nor the event loop — Python tests those. Not pytest's own skip machinery: you test that your hook attaches the skip, by counting outcomes, not how pytest prints an s. Not exact timings, either: "400 calls finish in under a second" is a stable claim on any machine, while "in 0.013 seconds" is not.

The rest of the chapter works through that table, row by row.


Running async tests with pytest-asyncio

The failure message lists several plugins. For code built on asyncio, the usual choice is pytest-asyncio:

text
pip install pytest-asyncio

Installing it is not enough by itself. Run the same test again and the output is word for word the same: async def functions are not natively supported. By default pytest-asyncio works in strict mode — it only touches tests you have explicitly handed to it. You hand one over with a marker:

python
import pytest

from prices import fetch_price


@pytest.mark.asyncio
async def test_pen_price():
    assert await fetch_price("pen") == 999.0
text
F                                                                        [100%]
=================================== FAILURES ===================================
________________________________ test_pen_price ________________________________

    @pytest.mark.asyncio
    async def test_pen_price():
>       assert await fetch_price("pen") == 999.0
E       assert 15.0 == 999.0

test_prices.py:8: AssertionError
=========================== short test summary info ============================
FAILED test_prices.py::test_pen_price - assert 15.0 == 999.0
1 failed in 0.10s

Now it fails for the right reason. assert 15.0 == 999.0 — the coroutine ran, await produced a real number, and the assertion compared it. Seeing a test fail with the wrong value is the proof that it is really running. Fix the number, and add a test for the error path while you are there:

python
import pytest

from prices import fetch_price


@pytest.mark.asyncio
async def test_pen_price():
    assert await fetch_price("pen") == 15.0


@pytest.mark.asyncio
async def test_unknown_item():
    with pytest.raises(KeyError):
        await fetch_price("stapler")
text
..                                                                       [100%]
2 passed in 0.10s

pytest.raises works inside an async test exactly as it does elsewhere; the await simply goes inside the with block.

There is one more wrong way worth seeing, because it is the most common one: avoiding the whole question by writing an ordinary def test and calling the async function without await.

python
from prices import fetch_price


def test_pen_price():
    assert fetch_price("pen") == 15.0
text
F                                                                        [100%]
=================================== FAILURES ===================================
________________________________ test_pen_price ________________________________

    def test_pen_price():
>       assert fetch_price("pen") == 15.0
E       AssertionError: assert <coroutine object fetch_price at 0x70659d650e80> == 15.0
E        +  where <coroutine object fetch_price at 0x70659d650e80> = fetch_price('pen')

test_noawait.py:5: AssertionError
=========================== short test summary info ============================
FAILED test_noawait.py::test_pen_price - AssertionError: assert <coroutine ob...
1 failed in 0.08s
sys:1: RuntimeWarning: coroutine 'fetch_price' was never awaited

The report says exactly what happened: fetch_price('pen') returned a coroutine object, and a coroutine is not equal to 15.0. The last line is Python adding that the coroutine was thrown away without ever running. This one at least fails loudly. The dangerous variant is assert fetch_price("pen") with no comparison — a coroutine object is truthy, so that test reports 1 passed, 1 warning while checking nothing. Any test of async code must itself be async def and must await.

asyncio_mode = "auto" and async fixtures

Writing @pytest.mark.asyncio above every test gets old quickly. In a project where async is the normal case, switch the mode in pyproject.toml:

toml
[tool.pytest.ini_options]
asyncio_mode = "auto"

In auto mode, pytest-asyncio takes every async def test and every async def fixture, with no marker. The header of a full pytest run tells you which mode is in force:

text
asyncio: mode=Mode.AUTO, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function

Fixtures can be async too. Here a cart is set up, handed to the test, and closed afterwards — with an await on both sides of the yield:

python
import pytest

from prices import fetch_price


class Cart:
    def __init__(self):
        self.items = []
        self.closed = False

    async def add(self, item):
        self.items.append((item, await fetch_price(item)))

    async def close(self):
        self.closed = True


@pytest.fixture
async def cart():
    c = Cart()
    yield c
    await c.close()


async def test_add(cart):
    await cart.add("pen")
    assert cart.items == [("pen", 15.0)]


async def test_two_items(cart):
    await cart.add("pen")
    await cart.add("notebook")
    assert [price for _, price in cart.items] == [15.0, 60.0]
text
..                                                                       [100%]
2 passed in 0.11s

The yield works as it did in the fixtures chapter: everything before it is setup, everything after it is teardown, and teardown runs even when the test fails.

In strict mode, the same fixture needs its own decorator, @pytest_asyncio.fixture, from import pytest_asyncio. With a plain @pytest.fixture there, pytest does not know how to run it and stops at setup:

text
E                                                                        [100%]
==================================== ERRORS ====================================
__________________________ ERROR at setup of test_add __________________________
'test_add' requested an async fixture 'cart', with no plugin or hook that handled it. This is an error, as pytest does not natively support it.
See: https://docs.pytest.org/en/stable/deprecations.html#sync-test-depending-on-async-fixture
=========================== short test summary info ============================
ERROR test_cart.py::test_add - Failed: 'test_add' requested an async fixture ...
1 error in 0.09s

So the choice is simple. Strict mode: mark async tests with @pytest.mark.asyncio and async fixtures with @pytest_asyncio.fixture. Auto mode: plain async def everywhere. Pick one per project and write it in the config so nobody has to guess.

How did pytest-asyncio do that? It did not change pytest. It implemented a few of pytest's hooks — functions with fixed names that pytest calls at fixed moments, such as "a test is about to be run". That is all a plugin is. The rest of this chapter writes a plugin of your own.

Hooks in conftest.py

A hook is a function named pytest_<something> that pytest calls at a particular moment of a run. You do not register it; you define it in a place pytest looks, and the name does the rest. The first such place is the conftest.py file you already use for shared fixtures.

A command-line option: pytest_addoption

Suppose the same tests should run against your laptop and against a staging server. The address belongs on the command line, not in the code:

conftest.py:

python
import pytest


def pytest_addoption(parser):
    parser.addoption(
        "--shop-url", default="http://localhost:8000", help="where the shop under test runs"
    )


@pytest.fixture
def shop_url(request):
    return request.config.getoption("--shop-url")

test_shop.py:

python
def test_shop_url(shop_url):
    print("testing against", shop_url)
    assert shop_url.startswith("http")

pytest -q -s, then pytest -q -s --shop-url https://staging.example.com:

text
testing against http://localhost:8000
.
1 passed in 0.07s
testing against https://staging.example.com
.
1 passed in 0.07s

pytest_addoption runs once, before the command line is parsed, and parser.addoption takes the same arguments as Python's argparse. Then request.config is the run's configuration object, and getoption reads the value back. The option also appears in pytest --help:

text
--shop-url=SHOP_URL   where the shop under test runs

Wrapping the option in a fixture is the usual pattern: tests ask for shop_url and never touch the configuration directly.

Skipping slow tests: pytest_collection_modifyitems

A common need: some tests take seconds, and you do not want them on every save — only before a push, or in CI. Mark them slow, and skip them unless a --runslow flag is given.

The tempting first attempt is to put the check inside each slow test: read the option at the top, call pytest.skip() if it is missing. It works, and it is the wrong way. The rule now lives in every slow test instead of one place; the day someone writes a new slow test and forgets those two lines, it runs on every save and nobody notices why the suite got slower. And the test body has to start running before it can decide not to run, so its fixtures are already set up — perhaps a database, perhaps a server.

The right place is before any test starts. After pytest has collected every test, and before any of them runs, it calls pytest_collection_modifyitems(config, items). items is the list of collected tests, and the hook may change it — one rule, in one place, applied to every test that carries the marker.

conftest.py:

python
import pytest


def pytest_addoption(parser):
    parser.addoption(
        "--runslow", action="store_true", default=False, help="also run tests marked slow"
    )


def pytest_configure(config):
    config.addinivalue_line("markers", "slow: a test that takes seconds, not milliseconds")


def pytest_report_header(config):
    if config.getoption("--runslow"):
        return "slow tests: included"
    return "slow tests: skipped (use --runslow)"


def pytest_collection_modifyitems(config, items):
    if config.getoption("--runslow"):
        return
    skip_slow = pytest.mark.skip(reason="needs --runslow")
    for item in items:
        if "slow" in item.keywords:
            item.add_marker(skip_slow)

test_reports.py:

python
import time

import pytest


def test_quick_total():
    assert sum([15.0, 60.0]) == 75.0


@pytest.mark.slow
def test_full_year_report():
    time.sleep(1)  # pretend this crunches a year of sales
    assert True

pytest -q, then pytest -q --runslow:

text
.s                                                                       [100%]
1 passed, 1 skipped in 0.10s
..                                                                       [100%]
2 passed in 1.09s

Without the flag, the slow test is an s and the run takes a tenth of a second. With it, both run and the second costs its full second. action="store_true" makes --runslow a switch: absent means False, present means True. Inside a hook there is no request, so the hook reads the value straight from config.

item.keywords holds the names of the test's markers (among other things), so "slow" in item.keywords asks "does this test carry @pytest.mark.slow?". item.add_marker(skip_slow) attaches a skip to it, exactly as if you had written @pytest.mark.skip(reason="needs --runslow") above the test yourself.

Registering the marker and adding a header line

Two smaller hooks in that file deserve a word.

pytest_configure(config) runs once, after the command line is parsed. config.addinivalue_line("markers", ...) registers the slow marker — the same as a markers = [...] line in pyproject.toml, but carried by the plugin itself. Without it, every @pytest.mark.slow produces PytestUnknownMarkWarning, and under --strict-markers it becomes an error. Now pytest --markers lists it first:

text
@pytest.mark.slow: a test that takes seconds, not milliseconds

pytest_report_header(config) returns a string (or a list of strings) that pytest prints at the top of a run. It is the right place to state anything a reader of a CI log needs to know. pytest -rs:

text
============================= test session starts ==============================
platform linux -- Python 3.12.3, pytest-9.1.1, pluggy-1.6.0
slow tests: skipped (use --runslow)
rootdir: /home/you/shop
collected 2 items

test_reports.py .s                                                       [100%]

=========================== short test summary info ============================
SKIPPED [1] test_reports.py:10: needs --runslow
========================= 1 passed, 1 skipped in 0.07s =========================

A skipped test with a clear reason is honest: the summary says what did not run, and why.

How pytest finds hooks

Every one of these hooks lives in a module. pytest collects such modules — plugins — from several places:

  1. Its own built-in plugins. Most of pytest is written as plugins: -k, markers, tmp_path, the terminal report.
  2. Installed packages that declare a pytest11 entry point. This is how pytest-asyncio, pytest-cov and the rest load themselves as soon as they are installed. The plugins: line of the header lists them.
  3. Modules named with -p on the command line, or in addopts.
  4. conftest.py files, at the root of your tests and in any subdirectory (a subdirectory's conftest.py applies only to tests below it). A few early hooks, such as pytest_addoption, are only honoured in a conftest.py at the root.

So the conftest.py above can be lifted out as is. Rename it to pytest_slowtests.py, and it is a plugin. Load it with -p and the module name:

text
pytest -p pytest_slowtests

The module has to be importable. If it sits at the project root, tell pytest so with pythonpath, and to avoid typing -p every time, put it in addopts:

toml
[tool.pytest.ini_options]
pythonpath = ["."]
addopts = "-p pytest_slowtests"

To share it across projects, make it a package and declare the entry point. The group must be named pytest11; the name on the left is the plugin's name, and the value on the right is the module:

toml
[build-system]
requires = ["setuptools>=69"]
build-backend = "setuptools.build_meta"

[project]
name = "pytest-slowtests"
version = "0.1.0"
dependencies = ["pytest>=8"]

[project.entry-points.pytest11]
slowtests = "pytest_slowtests"

Install it (pip install ./pytest-slowtests) into a fresh environment, and in a project with no conftest.py at all:

text
============================= test session starts ==============================
platform linux -- Python 3.12.3, pytest-9.1.1, pluggy-1.6.0
slow tests: skipped (use --runslow)
rootdir: /home/you/other-project
plugins: slowtests-0.1.0
collected 2 items

test_reports.py .s                                                       [100%]

========================= 1 passed, 1 skipped in 0.01s =========================

plugins: slowtests-0.1.0 — pytest found it through the entry point. The opposite also works: -p no:slowtests turns an installed plugin off for one run, which is useful when you suspect a plugin of causing trouble.

Testing a plugin with pytester

A plugin changes how pytest behaves, so its tests have to run pytest and look at what happened. pytest ships a fixture for exactly that, pytester. It is off by default; switch it on in the test module (or in the root conftest.py):

python
pytest_plugins = ["pytester"]

pytester gives each test an empty temporary directory, a fresh little project. pytester.makepyfile(...) writes a test file into it, pytester.runpytest(...) runs pytest there with the arguments you give, and the result has assert_outcomes(passed=..., skipped=..., failed=...) to check the counts, and result.stdout.fnmatch_lines([...]) to check lines of output (* may be used as a wildcard). The full version is in the complete example below.


A complete example

One small project holding everything from this chapter: async code, async tests in auto mode, the slow-test plugin, and tests for the plugin.

text
shop/
├── pyproject.toml
├── prices.py
├── pytest_slowtests.py
└── tests/
    ├── test_plugin.py
    └── test_prices.py

prices.py is the one from the start of the chapter, and pytest_slowtests.py is the conftest.py from the --runslow section, renamed. pyproject.toml:

toml
[tool.pytest.ini_options]
pythonpath = ["."]
testpaths = ["tests"]
addopts = "-p pytest_slowtests"
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"

tests/test_prices.py:

python
import asyncio

import pytest

from prices import PRICES, fetch_price


async def test_pen_price():
    assert await fetch_price("pen") == 15.0


async def test_unknown_item():
    with pytest.raises(KeyError):
        await fetch_price("stapler")


@pytest.mark.slow
async def test_every_price_at_once():
    names = list(PRICES) * 200
    prices = await asyncio.gather(*(fetch_price(n) for n in names))
    assert len(prices) == 400

tests/test_plugin.py:

python
pytest_plugins = ["pytester"]

SAMPLE = """
import pytest


def test_quick():
    assert True


@pytest.mark.slow
def test_slow():
    assert True
"""


def run(pytester, *args):
    # The inner run is a brand-new project in a temporary directory.
    pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function\n")
    pytester.makepyfile(SAMPLE)
    return pytester.runpytest("-p", "pytest_slowtests", *args)


def test_slow_is_skipped_by_default(pytester):
    result = run(pytester)
    result.assert_outcomes(passed=1, skipped=1)


def test_runslow_runs_everything(pytester):
    result = run(pytester, "--runslow")
    result.assert_outcomes(passed=2)


def test_header_mentions_the_flag(pytester):
    result = run(pytester)
    result.stdout.fnmatch_lines(["slow tests: skipped (use --runslow)"])

pytest -v (header shortened):

text
slow tests: skipped (use --runslow)
configfile: pyproject.toml
testpaths: tests
asyncio: mode=Mode.AUTO, debug=False, asyncio_default_fixture_loop_scope=function, asyncio_default_test_loop_scope=function
collecting ... collected 6 items

tests/test_plugin.py::test_slow_is_skipped_by_default PASSED             [ 16%]
tests/test_plugin.py::test_runslow_runs_everything PASSED                [ 33%]
tests/test_plugin.py::test_header_mentions_the_flag PASSED               [ 50%]
tests/test_prices.py::test_pen_price PASSED                              [ 66%]
tests/test_prices.py::test_unknown_item PASSED                           [ 83%]
tests/test_prices.py::test_every_price_at_once SKIPPED (needs --runslow) [100%]

========================= 5 passed, 1 skipped in 0.17s =========================

And pytest -q --runslow:

text
......                                                                   [100%]
6 passed in 0.40s

Three things are worth noticing.

First, the plugin works on async tests too. test_every_price_at_once is an async def with a slow marker, and the skip is applied to it like any other test. Hooks see collected items; whether the function behind one is async is pytest-asyncio's business, not theirs.

Second, asyncio.gather ran 400 calls of fetch_price, each sleeping 0.01 seconds, and the whole run took well under a second — not four. They waited at the same time. That is the reason async code exists, and the test proves it behaves that way.

Third, the makeini line in run. The inner run inside pytester is a separate project with its own configuration, and it loads every installed plugin — including pytest-asyncio, which warns when asyncio_default_fixture_loop_scope is unset. Giving the inner project that one setting keeps the outer report clean. The general lesson: a pytester run sees your installed plugins, but not your pyproject.toml.

When the plugin breaks, these tests say so precisely. Here the skip check was mistyped as "slw", so nothing was skipped:

text
>       result.assert_outcomes(passed=1, skipped=1)
E       AssertionError: assert {'passed': 2,...rors': 0, ...} == {'passed': 1,...rors': 0, ...}
E         
E         Omitting 4 identical items, use -vv to show
E         Differing items:
E         {'passed': 2} != {'passed': 1}
E         {'skipped': 0} != {'skipped': 1}
E         Use -v to get more diff

Below that, under Captured stdout call, pytest prints the whole output of the inner run, so you can see exactly what it did.


When it breaks

async def functions are not natively supported. The test is an async def, and nothing is running it. Either pytest-asyncio is not installed, or it is installed in strict mode and the test lacks @pytest.mark.asyncio. Add the marker, or set asyncio_mode = "auto".

requested an async fixture 'cart', with no plugin or hook that handled it An async def fixture is decorated with plain @pytest.fixture in strict mode. Use @pytest_asyncio.fixture, or switch to auto mode.

AssertionError: assert <coroutine object fetch_price at 0x...> == 15.0, followed by RuntimeWarning: coroutine 'fetch_price' was never awaited A normal def test called an async function without await, and compared the coroutine object itself. Make the test async def and write await fetch_price("pen").

PytestUnknownMarkWarning: Unknown pytest.mark.slow - is this a typo? The marker is used but never registered. Register it in pytest_configure with config.addinivalue_line("markers", ...), or in the markers setting.

PluginValidationError: unknown hook 'pytest_addoptions' in plugin <module 'conftest' ...> A function starts with pytest_ but is not a hook pytest knows — here, a stray s. pytest checks every pytest_ name in a plugin, which is exactly what catches the typo. Check the spelling against the hook reference.

ValueError: no option named '--runslow' config.getoption asked for an option that no pytest_addoption added. Either the name is different, or the pytest_addoption hook lives in a conftest.py pytest did not load early enough — keep it at the root of your tests.

pytest: error: unrecognized arguments: --run-slow The option on the command line does not match the one registered. pytest --help lists every option your plugins have added, under "custom options".

ImportError: Error importing plugin "pytest_slowtests": No module named 'pytest_slowtests' -p takes a module name, and that module could not be imported. Add pythonpath = ["."] to the config, or install the plugin as a package.