Python · Testing with pytest · Advanced

Async Testing with pytest-asyncio

5 tasks

Test async Python code and async HTTP clients with pytest-asyncio.

Why pytest cannot run async tests — and how pytest-asyncio fixes it

#
## The problem: async def functions return coroutines, not results When Python encounters `async def test_something():` and pytest calls it, the function body does not execute. Calling an `async def` function returns a *coroutine object* — a suspended computation that only runs when placed inside an event loop and explicitly awaited. Standard pytest calls your test function like a regular callable: ```python # What pytest effectively does (simplified): result = test_function() ``` If `test_function` is `async def`, `result` is a coroutine object. No exception is raised, so pytest reports **PASSED** — and the test body never ran, no assertion was checked, nothing was verified. ```python # This test always passes and always lies. async def test_user_count(): count = await get_user_count() # Never executed. assert count == 42 # Never executed. ``` This is the most dangerous failure mode in async testing: **silent success on a test that does nothing.** Python emits `RuntimeWarning: coroutine 'test_user_count' was never awaited`, but pytest may suppress it, and the PASSED status in the output looks perfectly normal. ## The fix: pytest-asyncio `pytest-asyncio` is a plugin that teaches pytest how to run async test functions. It wraps each `async def test_*` in an event loop, runs the coroutine to completion, and propagates any exceptions as test failures. ```bash pip install pytest-asyncio ``` Add one line to `pytest.ini`: ```ini [pytest] asyncio_mode = auto ``` That is the complete setup. Every `async def test_*` function is now automatically run inside an event loop. ## asyncio_mode: choosing your discipline | Mode | How it works | |---|---| | `auto` | Every `async def test_*` is automatically treated as an async test | | `strict` | Each async test must be explicitly marked with `@pytest.mark.asyncio` | | `loose` | Deprecated — use `auto` | **Use `asyncio_mode = auto` for new projects.** In `strict` mode, forgetting `@pytest.mark.asyncio` on a test takes you straight back to the original bug: the coroutine is never awaited, the test silently passes, nothing was checked. `strict` is useful when adding pytest-asyncio to a large existing codebase where you want to opt in gradually. ## Your first async test With `asyncio_mode = auto` configured: ```python # test_compute.py import asyncio async def slow_double(x): await asyncio.sleep(0.01) # simulate async I/O return x * 2 async def test_slow_double(): result = await slow_double(5) assert result == 10 ``` ``` $ pytest test_compute.py -v test_compute.py::test_slow_double PASSED ``` No decorators needed. The event loop ran, the `await` worked, the assertion ran. ## Async fixtures `@pytest.fixture` works on `async def` functions exactly as on regular functions. With `asyncio_mode = auto`, pytest-asyncio automatically awaits async fixtures before injecting them into tests. The standard pattern is an `async with` block inside a `yield` fixture: ```python import pytest import httpx @pytest.fixture async def http_client(): async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client: yield client # `async with` ensures client.aclose() is called after every test, # even if the test raises an exception. async def test_products(http_client): resp = await http_client.get('/api/products/') assert resp.status_code == 200 ``` The fixture lifecycle is identical to sync fixtures: - Code before `yield` — setup (the `AsyncClient` is created and the connection pool is ready) - `yield client` — the client is injected into the test - After `yield` — teardown (the `async with` exit closes the connection pool) ## Sharing a client across tests: scope For a session-scoped async fixture (one client for the entire test run), you must also align the event loop scope. Add to `pytest.ini`: ```ini [pytest] asyncio_mode = auto asyncio_default_fixture_loop_scope = session ``` ```python @pytest.fixture(scope='session') async def http_client(): async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client: yield client ``` Without `asyncio_default_fixture_loop_scope = session`, each test runs in its own event loop, and session-scoped async fixtures get confused about which loop they belong to — you will see `RuntimeError: Task attached to a different loop`. ## Authenticated session fixture A session fixture can also log in once and reuse the token across all tests: ```python @pytest.fixture(scope='session') async def auth_client(): async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client: resp = await client.post('/api/auth/token/', json={ 'username': 'testuser', 'password': 'TestUser2024!', }) resp.raise_for_status() token = resp.json()['token'] client.headers['Authorization'] = f'Token {token}' yield client ``` One login. One connection pool. Every test that requests `auth_client` gets the same authenticated session. ## Common pitfalls **1. Missing `asyncio_mode = auto` in config** Without it, `async def test_*` functions return coroutine objects. Tests appear to pass because no exception is raised — but the test body never ran. Always configure `pytest.ini`. **2. `asyncio.run()` inside an async test** ```python # Wrong — you are already inside a running event loop async def test_bad(): result = asyncio.run(some_coroutine()) # RuntimeError: This event loop is already running ``` Inside a pytest-asyncio test, you are already inside a running event loop. Use `await` directly, not `asyncio.run()`. **3. Missing `await` on an async call** ```python async def test_products(http_client): resp = http_client.get('/api/products/') # Missing await! assert resp.status_code == 200 # AttributeError: 'coroutine' object has no attribute 'status_code' ``` Every async call inside a test must be awaited. If you see `AttributeError` on a response attribute, check for a missing `await`. **4. Scope mismatch on session async fixtures** A `scope='session'` async fixture without `asyncio_default_fixture_loop_scope = session` in config raises `RuntimeError` about event loop conflicts or a `ScopeMismatch` error. Both the fixture `scope` and the config setting must be set together.

Async testing in practice: from simple coroutines to real HTTP

#
## Project setup ```bash pip install pytest-asyncio httpx ``` ```ini # pytest.ini [pytest] asyncio_mode = auto asyncio_default_fixture_loop_scope = session ``` ## Example 1: Testing a simple async function ```python # test_async_basic.py import asyncio async def fetch_status(delay: float) -> str: await asyncio.sleep(delay) return 'ready' async def test_fetch_returns_ready(): result = await fetch_status(0.01) assert result == 'ready' async def test_fetch_returns_string(): result = await fetch_status(0) assert isinstance(result, str) ``` `asyncio.sleep(0)` yields to the event loop for one cycle without actually waiting — the minimal async operation, useful for stubs and fixtures that must be `async def` without introducing real delays. ## Example 2: Async fixture that creates and closes an HTTP client ```python # conftest.py import pytest import httpx BASE_URL = 'https://apilearn.tukas.dev' @pytest.fixture(scope='session') async def client(): async with httpx.AsyncClient(base_url=BASE_URL, timeout=10.0) as c: yield c # Connection pool is closed here, after the last test in the session. ``` ```python # test_products.py async def test_products_status(client): resp = await client.get('/api/products/') assert resp.status_code == 200 async def test_products_structure(client): resp = await client.get('/api/products/') data = resp.json() assert 'results' in data assert isinstance(data['results'], list) ``` The `client` fixture is created once for the session. Both tests share the same `AsyncClient` instance and underlying connection pool — no reconnection overhead between tests. ## Example 3: Authenticated async client ```python # conftest.py (continued) @pytest.fixture(scope='session') async def auth_client(client): resp = await client.post('/api/auth/token/', json={ 'username': 'testuser', 'password': 'TestUser2024!', }) resp.raise_for_status() token = resp.json()['token'] client.headers['Authorization'] = f'Token {token}' yield client # Teardown: remove the auth header so tests using plain `client` are not affected. del client.headers['Authorization'] ``` ```python # test_profile.py async def test_profile(auth_client): resp = await auth_client.get('/api/users/profile/') assert resp.status_code == 200 assert 'username' in resp.json() ``` `auth_client` depends on `client` — fixture chaining works identically in async fixtures. The teardown removes the `Authorization` header so that tests using the unauthenticated `client` fixture are not accidentally authenticated. ## Example 4: Sequential async operations ```python # test_sequential.py async def get_first_product_slug(client) -> str: resp = await client.get('/api/products/') resp.raise_for_status() products = resp.json()['results'] return products[0]['slug'] async def test_first_product_has_slug(client): slug = await get_first_product_slug(client) assert isinstance(slug, str) assert len(slug) > 0 ``` Sequential `await` calls look like ordinary synchronous Python. The event loop handles concurrency between tests, but within a single test, execution proceeds line by line. ## Example 5: Testing async exceptions `pytest.raises` works inside async tests exactly as in sync tests: ```python # test_validation.py import pytest async def require_https(url: str) -> str: if not url.startswith('https://'): raise ValueError(f'HTTPS required, got: {url!r}') return url async def test_rejects_http(): with pytest.raises(ValueError, match='HTTPS required'): await require_https('http://example.com') async def test_accepts_https(): result = await require_https('https://example.com') assert result == 'https://example.com' ``` `pytest.raises` is not async-aware — it catches synchronous exceptions. The `await require_https(...)` runs the coroutine; if it raises, the exception propagates out of the `await` expression and is caught by `pytest.raises` normally. No special async version is needed. ## Running the full suite ``` $ pytest -v test_async_basic.py::test_fetch_returns_ready PASSED test_async_basic.py::test_fetch_returns_string PASSED test_products.py::test_products_status PASSED test_products.py::test_products_structure PASSED test_profile.py::test_profile PASSED test_sequential.py::test_first_product_has_slug PASSED test_validation.py::test_rejects_http PASSED test_validation.py::test_accepts_https PASSED ```

pytest-asyncio reference card

#
## Installation and config ```bash pip install pytest-asyncio httpx ``` ```ini # pytest.ini [pytest] asyncio_mode = auto asyncio_default_fixture_loop_scope = session ``` ## asyncio_mode values | Value | Behaviour | |---|---| | `auto` | All `async def test_*` run as async tests automatically | | `strict` | Each test needs `@pytest.mark.asyncio` | | `loose` | Deprecated | ## Async test (no decorator needed with `auto`) ```python async def test_something(): result = await some_coroutine() assert result == expected ``` ## Async fixture patterns ```python # Function-scoped (default): new client per test @pytest.fixture async def http_client(): async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client: yield client # Session-scoped: one client for all tests @pytest.fixture(scope='session') async def http_client(): async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client: yield client ``` Requires `asyncio_default_fixture_loop_scope = session` in `pytest.ini` for session scope. ## Authenticated session fixture ```python @pytest.fixture(scope='session') async def auth_client(http_client): resp = await http_client.post('/api/auth/token/', json={'username': 'testuser', 'password': 'TestUser2024!'}) token = resp.json()['token'] http_client.headers['Authorization'] = f'Token {token}' yield http_client del http_client.headers['Authorization'] ``` ## Testing exceptions ```python async def test_raises(): with pytest.raises(SomeError, match='expected message'): await function_that_raises() ``` ## httpx.AsyncClient quick reference | Call | Description | |---|---| | `await client.get(url, params={...})` | GET request | | `await client.post(url, json={...})` | POST with JSON body | | `resp.status_code` | HTTP status integer | | `resp.json()` | Parse JSON body | | `resp.headers['X-Header']` | Read response header | | `resp.raise_for_status()` | Raise on 4xx/5xx | ## Common errors | Error | Cause | Fix | |---|---|---| | Test always passes, body never runs | Missing `asyncio_mode = auto` | Add to `pytest.ini` | | `RuntimeError: This event loop is already running` | Called `asyncio.run()` inside async test | Use `await` directly | | `AttributeError: 'coroutine' object has no attribute ...` | Missing `await` before async call | Add `await` | | `RuntimeError: Task attached to a different loop` | Session fixture without loop scope config | Set `asyncio_default_fixture_loop_scope = session` |
01

Set up pytest-asyncio and write your first async test

#

Configure `pytest.ini` with `asyncio_mode = auto`. Write an async function `slow_double(x)` that: - Uses `await asyncio.sleep(0)` to yield to the event loop - Returns `x * 2` Write `test_slow_double()` that awaits `slow_double(7)` and asserts the result equals `14`. **Important:** verify that your `pytest.ini` has `asyncio_mode = auto` before running. Without it the test will silently pass without running its body.

# pytest.ini (create in project root):
# [pytest]
# asyncio_mode = auto

# test_first_async.py
import asyncio


async def slow_double(x):
    # await asyncio.sleep(0) to yield to the event loop, then return x * 2
    pass


async def test_slow_double():
    result = ...
    assert result == 14
Solution
# pytest.ini
# [pytest]
# asyncio_mode = auto

# test_first_async.py
import asyncio


async def slow_double(x):
    await asyncio.sleep(0)
    return x * 2


async def test_slow_double():
    result = await slow_double(7)
    assert result == 14
02

Write an async httpx fixture

#

Install `httpx` and write a **function-scoped** async fixture `http_client` that: - Creates an `httpx.AsyncClient` with `base_url='https://apilearn.tukas.dev'` - Yields the client to the test - Closes the client after the test finishes (use `async with` for automatic teardown) Write `test_client_works(http_client)` that GETs `/api/products/` and asserts the status code is 200.

# test_http_fixture.py
import pytest
import httpx


@pytest.fixture
async def http_client():
    # Use async with httpx.AsyncClient(...) as client: yield client
    ...


async def test_client_works(http_client):
    resp = ...
    assert resp.status_code == 200
Solution
# test_http_fixture.py
import pytest
import httpx


@pytest.fixture
async def http_client():
    async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client:
        yield client


async def test_client_works(http_client):
    resp = await http_client.get('/api/products/')
    assert resp.status_code == 200
03

Test a real API endpoint asynchronously

#

Move the `http_client` fixture to `conftest.py` with `scope='session'`. Also add to `pytest.ini`: ``` asyncio_default_fixture_loop_scope = session ``` Write two async tests: 1. `test_products_returns_list` — GET `/api/products/`, assert status 200, assert `'results'` key exists, assert `results` is a list. 2. `test_products_page_size` — GET `/api/products/?page_size=2`, assert the returned results list has at most 2 items.

# conftest.py
import pytest
import httpx


@pytest.fixture(scope='session')
async def http_client():
    async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client:
        yield client


# test_products_async.py


async def test_products_returns_list(http_client):
    resp = await http_client.get('/api/products/')
    ...


async def test_products_page_size(http_client):
    resp = ...
    ...
Solution
# conftest.py
import pytest
import httpx


@pytest.fixture(scope='session')
async def http_client():
    async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client:
        yield client


# test_products_async.py


async def test_products_returns_list(http_client):
    resp = await http_client.get('/api/products/')
    assert resp.status_code == 200
    data = resp.json()
    assert 'results' in data
    assert isinstance(data['results'], list)


async def test_products_page_size(http_client):
    resp = await http_client.get('/api/products/?page_size=2')
    assert resp.status_code == 200
    results = resp.json()['results']
    assert len(results) <= 2
04

Write an async function and test sequential awaited calls

#

Write an async function `get_first_product_slug()` that: 1. Creates its own `httpx.AsyncClient` (no fixture — use `async with` inside the function) 2. GETs `https://apilearn.tukas.dev/api/products/` 3. Returns the `slug` field of the first item in `results` Write `test_first_product_has_slug()` that awaits the function and asserts the result is a non-empty string.

# test_sequential.py
import httpx


async def get_first_product_slug() -> str:
    async with httpx.AsyncClient() as client:
        # Step 1: GET the products endpoint
        # Step 2: return the slug of the first product
        ...


async def test_first_product_has_slug():
    slug = ...
    assert isinstance(slug, str)
    assert len(slug) > 0
Solution
# test_sequential.py
import httpx


async def get_first_product_slug() -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.get('https://apilearn.tukas.dev/api/products/')
        resp.raise_for_status()
        products = resp.json()['results']
        return products[0]['slug']


async def test_first_product_has_slug():
    slug = await get_first_product_slug()
    assert isinstance(slug, str)
    assert len(slug) > 0
05

Test that an async function raises an exception

#

Write an async function `require_https(url: str) -> str` that: - Raises `ValueError` with a message containing `"HTTPS required"` if `url` does not start with `'https://'` - Returns the URL unchanged otherwise Write two tests: 1. `test_require_https_rejects_http` — assert `ValueError` is raised for `'http://example.com'` 2. `test_require_https_accepts_https` — assert the return value equals `'https://example.com'`

# test_async_raises.py
import pytest


async def require_https(url: str) -> str:
    ...


async def test_require_https_rejects_http():
    with pytest.raises(...):
        await require_https(...)


async def test_require_https_accepts_https():
    result = await require_https(...)
    assert result == ...
Solution
# test_async_raises.py
import pytest


async def require_https(url: str) -> str:
    if not url.startswith('https://'):
        raise ValueError(f'HTTPS required, got: {url!r}')
    return url


async def test_require_https_rejects_http():
    with pytest.raises(ValueError, match='HTTPS required'):
        await require_https('http://example.com')


async def test_require_https_accepts_https():
    result = await require_https('https://example.com')
    assert result == 'https://example.com'