Python · Тестирование с pytest · Продвинутый

Async-тестирование с pytest-asyncio

5 задач

Тестируйте async Python-код и async HTTP-клиенты с pytest-asyncio.

Почему pytest не может запускать async-тесты -- и как pytest-asyncio это исправляет

#
## Проблема: функции async def возвращают корутины, а не результаты Когда Python встречает `async def test_something():` и pytest вызывает её, тело функции не выполняется. Вызов функции `async def` возвращает *объект корутины* -- приостановленное вычисление, которое выполняется только при помещении в цикл событий и явном ожидании через `await`. Стандартный pytest вызывает вашу тест-функцию как обычный вызываемый объект: ```python # Что pytest фактически делает (упрощённо): result = test_function() ``` Если `test_function` является `async def`, `result` -- это объект корутины. Исключение не выбрасывается, поэтому pytest сообщает **PASSED** -- и тело теста никогда не выполнялось, ни одно утверждение не проверялось, ничего не верифицировалось. ```python # Этот тест всегда проходит и всегда лжёт. async def test_user_count(): count = await get_user_count() # Никогда не выполняется. assert count == 42 # Никогда не выполняется. ``` Это самый опасный режим отказа при async-тестировании: **молчаливый успех теста, который ничего не делает.** Python выдаёт `RuntimeWarning: coroutine 'test_user_count' was never awaited`, но pytest может подавить его, и статус PASSED в выводе выглядит совершенно нормально. ## Решение: pytest-asyncio `pytest-asyncio` -- плагин, обучающий pytest запускать async тест-функции. Он оборачивает каждую `async def test_*` в цикл событий, выполняет корутину до завершения и передаёт любые исключения как падения тестов. ```bash pip install pytest-asyncio ``` Добавьте одну строку в `pytest.ini`: ```ini [pytest] asyncio_mode = auto ``` Это полная настройка. Каждая функция `async def test_*` теперь автоматически запускается внутри цикла событий. ## asyncio_mode: выбор режима | Режим | Как работает | |---|---| | `auto` | Каждый `async def test_*` автоматически трактуется как async-тест | | `strict` | Каждый async-тест должен быть явно помечен `@pytest.mark.asyncio` | | `loose` | Устаревший -- используйте `auto` | **Используйте `asyncio_mode = auto` для новых проектов.** В режиме `strict` забытый `@pytest.mark.asyncio` возвращает вас к исходному багу: корутина никогда не ожидается, тест молча проходит, ничего не проверяется. `strict` полезен при добавлении pytest-asyncio в большую существующую кодовую базу, где нужен постепенный переход. ## Ваш первый async-тест После настройки `asyncio_mode = auto`: ```python # test_compute.py import asyncio async def slow_double(x): await asyncio.sleep(0.01) # имитация 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 ``` Никаких декораторов. Цикл событий запустился, `await` сработал, утверждение выполнилось. ## Async-фикстуры `@pytest.fixture` работает на `async def` функциях точно так же, как на обычных. При `asyncio_mode = auto` pytest-asyncio автоматически ожидает async-фикстуры перед их внедрением в тесты. Стандартный паттерн -- блок `async with` внутри yield-фикстуры: ```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` гарантирует вызов client.aclose() после каждого теста, # даже если тест выбросил исключение. async def test_products(http_client): resp = await http_client.get('/api/products/') assert resp.status_code == 200 ``` Жизненный цикл фикстуры идентичен синхронным фикстурам: - Код до `yield` -- инициализация (`AsyncClient` создан и пул соединений готов) - `yield client` -- клиент внедряется в тест - После `yield` -- завершение (выход из `async with` закрывает пул соединений) ## Совместное использование клиента между тестами: scope Для async-фикстуры с session scope (один клиент на весь прогон) нужно также согласовать scope цикла событий. Добавьте в `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 ``` Без `asyncio_default_fixture_loop_scope = session` каждый тест работает в собственном цикле событий, и async-фикстуры с session scope путаются в том, какому циклу принадлежат -- вы увидите `RuntimeError: Task attached to a different loop`. ## Фикстура аутентифицированной сессии Фикстура сессии может также выполнить вход один раз и переиспользовать токен во всех тестах: ```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 ``` Один вход. Один пул соединений. Каждый тест, запрашивающий `auth_client`, получает одну и ту же аутентифицированную сессию. ## Распространённые ловушки **1. Отсутствие `asyncio_mode = auto` в конфиге** Без этого функции `async def test_*` возвращают объекты корутин. Тесты выглядят прошедшими, потому что исключение не выбрасывается -- но тело теста никогда не выполнялось. Всегда настраивайте `pytest.ini`. **2. `asyncio.run()` внутри async-теста** ```python # Неправильно -- вы уже внутри работающего цикла событий async def test_bad(): result = asyncio.run(some_coroutine()) # RuntimeError: This event loop is already running ``` Внутри теста pytest-asyncio вы уже находитесь в работающем цикле событий. Используйте `await` напрямую, а не `asyncio.run()`. **3. Пропущенный `await` перед async-вызовом** ```python async def test_products(http_client): resp = http_client.get('/api/products/') # Пропущен await! assert resp.status_code == 200 # AttributeError: у объекта 'coroutine' нет атрибута 'status_code' ``` Каждый async-вызов внутри теста должен ожидаться через `await`. Если вы видите `AttributeError` на атрибуте ответа, проверьте пропущенный `await`. **4. Несоответствие scope у async-фикстур сессии** Async-фикстура с `scope='session'` без `asyncio_default_fixture_loop_scope = session` в конфиге вызывает `RuntimeError` о конфликтах цикла событий или ошибку `ScopeMismatch`. И `scope` фикстуры, и настройка конфига должны быть установлены вместе.

Async-тестирование на практике: от простых корутин до реального HTTP

#
## Настройка проекта ```bash pip install pytest-asyncio httpx ``` ```ini # pytest.ini [pytest] asyncio_mode = auto asyncio_default_fixture_loop_scope = session ``` ## Пример 1: Тестирование простой async-функции ```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)` уступает цикл событий на один цикл без реального ожидания -- минимальная async-операция, полезная для заглушек и фикстур, которые должны быть `async def` без реальных задержек. ## Пример 2: Async-фикстура, создающая и закрывающая HTTP-клиент ```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 # Пул соединений закрывается здесь, после последнего теста в сессии. ``` ```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) ``` Фикстура `client` создаётся один раз для сессии. Оба теста разделяют один экземпляр `AsyncClient` и пул соединений -- без накладных расходов на переподключение между тестами. ## Пример 3: Аутентифицированный async-клиент ```python # conftest.py (продолжение) @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 # Завершение: убрать auth-заголовок, чтобы тесты с обычным `client` не были случайно аутентифицированы. 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` зависит от `client` -- цепочка фикстур работает идентично в async-фикстурах. Завершение удаляет заголовок `Authorization`, чтобы тесты, использующие неаутентифицированный `client`, случайно не оказались аутентифицированными. ## Пример 4: Последовательные async-операции ```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 ``` Последовательные вызовы `await` выглядят как обычный синхронный Python. Цикл событий управляет параллелизмом между тестами, но внутри одного теста выполнение идёт строка за строкой. ## Пример 5: Тестирование async-исключений `pytest.raises` работает внутри async-тестов точно так же, как в синхронных: ```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` не является async-совместимым -- и не должен им быть. Когда вы выполняете `await require_https(...)`, Python запускает корутину и, если она выбрасывает исключение, оно распространяется из выражения `await` как обычное синхронное исключение. `pytest.raises` перехватывает его как обычно. ## Запуск полного тест-сьюта ``` $ 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

#
## Установка и конфигурация ```bash pip install pytest-asyncio httpx ``` ```ini # pytest.ini [pytest] asyncio_mode = auto asyncio_default_fixture_loop_scope = session ``` ## Значения asyncio_mode | Значение | Поведение | |---|---| | `auto` | Все `async def test_*` запускаются как async-тесты автоматически | | `strict` | Каждый тест требует `@pytest.mark.asyncio` | | `loose` | Устаревший | ## Async-тест (декоратор не нужен при `auto`) ```python async def test_something(): result = await some_coroutine() assert result == expected ``` ## Паттерны async-фикстур ```python # Function scope (по умолчанию): новый клиент на каждый тест @pytest.fixture async def http_client(): async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client: yield client # Session scope: один клиент на все тесты @pytest.fixture(scope='session') async def http_client(): async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client: yield client ``` Требуется `asyncio_default_fixture_loop_scope = session` в `pytest.ini` для session scope. ## Фикстура аутентифицированной сессии ```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'] ``` ## Тестирование исключений ```python async def test_raises(): with pytest.raises(SomeError, match='expected message'): await function_that_raises() ``` ## Краткий справочник httpx.AsyncClient | Вызов | Описание | |---|---| | `await client.get(url, params={...})` | GET-запрос | | `await client.post(url, json={...})` | POST с телом JSON | | `resp.status_code` | HTTP-статус (целое число) | | `resp.json()` | Распарсить тело JSON | | `resp.headers['X-Header']` | Прочитать заголовок ответа | | `resp.raise_for_status()` | Выбросить исключение при 4xx/5xx | ## Частые ошибки | Ошибка | Причина | Решение | |---|---|---| | Тест всегда проходит, тело не запускается | Отсутствует `asyncio_mode = auto` | Добавить в `pytest.ini` | | `RuntimeError: This event loop is already running` | Вызван `asyncio.run()` внутри async-теста | Использовать `await` напрямую | | `AttributeError: у объекта 'coroutine' нет атрибута ...` | Пропущен `await` перед async-вызовом | Добавить `await` | | `RuntimeError: Task attached to a different loop` | Фикстура сессии без настройки scope цикла | Установить `asyncio_default_fixture_loop_scope = session` |
01

Настроить pytest-asyncio и написать первый async-тест

#

Настройте `pytest.ini` с `asyncio_mode = auto`. Напишите async-функцию `slow_double(x)`, которая: - Использует `await asyncio.sleep(0)` для уступки циклу событий - Возвращает `x * 2` Напишите `test_slow_double()`, ожидающий `slow_double(7)` и проверяющий, что результат равен `14`. **Важно:** убедитесь, что в `pytest.ini` есть `asyncio_mode = auto` перед запуском. Без этого тест молча пройдёт без выполнения тела.

# pytest.ini (создать в корне проекта):
# [pytest]
# asyncio_mode = auto

# test_first_async.py
import asyncio


async def slow_double(x):
    # await asyncio.sleep(0) для уступки циклу событий, затем вернуть x * 2
    pass


async def test_slow_double():
    result = ...
    assert result == 14
Решение
# 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

Написать async httpx-фикстуру

#

Установите `httpx` и напишите **function-scoped** async-фикстуру `http_client`, которая: - Создаёт `httpx.AsyncClient` с `base_url='https://apilearn.tukas.dev'` - Передаёт клиент в тест через yield - Закрывает клиент после завершения теста (используйте `async with` для автоматического завершения) Напишите `test_client_works(http_client)`, выполняющий GET `/api/products/` и проверяющий, что код статуса равен 200.

# test_http_fixture.py
import pytest
import httpx


@pytest.fixture
async def http_client():
    # Используйте async with httpx.AsyncClient(...) as client: yield client
    ...


async def test_client_works(http_client):
    resp = ...
    assert resp.status_code == 200
Решение
# 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

Тестировать реальный API-эндпоинт асинхронно

#

Перенесите фикстуру `http_client` в `conftest.py` с `scope='session'`. Также добавьте в `pytest.ini`: ``` asyncio_default_fixture_loop_scope = session ``` Напишите два async-теста: 1. `test_products_returns_list` -- GET `/api/products/`, проверить статус 200, наличие ключа `'results'`, что `results` является списком. 2. `test_products_page_size` -- GET `/api/products/?page_size=2`, проверить, что возвращённый список результатов содержит не более 2 элементов.

# 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 = ...
    ...
Решение
# 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

Написать async-функцию и тестировать последовательные вызовы с await

#

Напишите async-функцию `get_first_product_slug()`, которая: 1. Создаёт собственный `httpx.AsyncClient` (без фикстуры -- используйте `async with` внутри функции) 2. Выполняет GET `https://apilearn.tukas.dev/api/products/` 3. Возвращает поле `slug` первого элемента в `results` Напишите `test_first_product_has_slug()`, ожидающий функцию и проверяющий, что результат является непустой строкой.

# test_sequential.py
import httpx


async def get_first_product_slug() -> str:
    async with httpx.AsyncClient() as client:
        # Шаг 1: GET эндпоинта products
        # Шаг 2: вернуть slug первого продукта
        ...


async def test_first_product_has_slug():
    slug = ...
    assert isinstance(slug, str)
    assert len(slug) > 0
Решение
# 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

Тестировать выброс исключения async-функцией

#

Напишите async-функцию `require_https(url: str) -> str`, которая: - Выбрасывает `ValueError` с сообщением, содержащим `"HTTPS required"`, если `url` не начинается с `'https://'` - Возвращает URL без изменений в противном случае Напишите два теста: 1. `test_require_https_rejects_http` -- проверить, что `ValueError` выбрасывается для `'http://example.com'` 2. `test_require_https_accepts_https` -- проверить, что возвращаемое значение равно `'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 == ...
Решение
# 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'