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` повертає *об'єкт корутини* -- призупинене обчислення, що виконується лише тоді, коли воно поміщене всередину циклу подій і явно очікується. Стандартний pytest викликає вашу тест-функцію як звичайний callable: ```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 # Ніколи не виконується. ``` Це найнебезпечніший сценарій збою при асинхронному тестуванні: **тихий успіх тесту, що нічого не робить.** Python видає `RuntimeWarning: coroutine 'test_user_count' was never awaited`, але pytest може придушити це попередження, а статус PASSED у виводі виглядає цілком нормально. ## Виправлення: pytest-asyncio `pytest-asyncio` -- це плагін, що навчає pytest запускати асинхронні тест-функції. Він огортає кожну `async def test_*` у цикл подій, виконує корутину до завершення і передає будь-які винятки як провали тестів. ```bash pip install pytest-asyncio ``` Додайте один рядок у `pytest.ini`: ```ini [pytest] asyncio_mode = auto ``` Це повне налаштування. Тепер кожна функція `async def test_*` автоматично виконується всередині циклу подій. ## asyncio_mode: вибір дисципліни | Режим | Принцип роботи | |---|---| | `auto` | Кожна `async def test_*` автоматично обробляється як асинхронний тест | | `strict` | Кожен асинхронний тест має бути явно позначений `@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) # імітація асинхронного 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 ``` Цикл життя фікстури ідентичний sync-фікстурам: - Код до `yield` -- налаштування (`AsyncClient` створюється і пул з'єднань готовий) - `yield client` -- клієнт впроваджується у тест - Після `yield` -- teardown (вихід з `async with` закриває пул з'єднань) ## Спільний клієнт для тестів: scope Для session-scoped async-фікстури (один клієнт для всього запуску тестів) необхідно також вирівняти 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` кожен тест виконується у власному циклі подій, і session-scoped async-фікстури плутаються щодо того, якому циклу вони належать -- ви побачите `RuntimeError: Task attached to a different loop`. ## Фікстура аутентифікованої сесії Session-фікстура може також виконувати вхід один раз і повторно використовувати токен для всіх тестів: ```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' object has no attribute 'status_code' ``` Кожен async-виклик всередині тесту має бути очікуваним. Якщо ви бачите `AttributeError` на атрибуті відповіді -- перевірте відсутній `await`. **4. Невідповідність scope у session async-фікстурах** `scope='session'` async-фікстура без `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 # Teardown: видалити заголовок авторизації, щоб тести, що використовують # звичайний `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-фікстурах. Teardown видаляє заголовок `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-тестів точно так само, як у sync-тестах: ```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-aware -- і в цьому немає потреби. `await require_https(...)` запускає корутину; якщо вона підіймає виняток, він поширюється з виразу `await` як звичайний синхронний виняток. `pytest.raises` перехоплює його як завжди. Спеціальна async-версія не потрібна. ## Запуск повного набору тестів ``` $ 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-scoped (за замовчуванням): новий клієнт на кожен тест @pytest.fixture async def http_client(): async with httpx.AsyncClient(base_url='https://apilearn.tukas.dev') as client: yield client # Session-scoped: один клієнт для всіх тестів @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' object has no attribute ...` | Відсутній `await` перед async-викликом | Додати `await` | | `RuntimeError: Task attached to a different loop` | Session-фікстура без налаштування 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` для автоматичного teardown) Напишіть `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`, перевірте, що повернутий список results містить не більше 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` -- перевірте, що для `'http://example.com'` підіймається `ValueError` 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'