Фикстуры pytest для асинхронного кода: от event loop до базы

Асинхронные тесты ломаются не там, где их пишут, а там, где их запускают все вместе: «Event loop is closed», соединение с базой из чужого цикла, тест, который проходит один и падает в наборе. Почти всегда причина — фикстуры живут в одном цикле событий, а тесты — в другом.

Разберём настройку pytest-asyncio 1.x, фикстуру базы с откатом транзакции и приёмы, которые держат набор из тысячи тестов в пределах минуты.

Один цикл событий на сессию

По умолчанию pytest-asyncio создаёт новый цикл для каждого теста. Это надёжно, но дорогие ресурсы — пул соединений, HTTP-клиент — приходится создавать заново каждый раз. Если поднять их один раз на сессию, тесты должны выполняться в том же цикле.

[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"

import pytest
from pytest_asyncio import is_async_test


def pytest_collection_modifyitems(items):
    session_loop = pytest.mark.asyncio(loop_scope="session")
    for item in items:
        if is_async_test(item):
            item.add_marker(session_loop, append=False)

База: транзакция на тест

Схему создаём один раз на сессию, а каждый тест работает внутри транзакции, которую в конце откатываем. Код приложения при этом может сам вызывать commit() — с join_transaction_mode="create_savepoint" он зафиксирует только точку сохранения.

@pytest.fixture(scope="session")
async def engine():
    engine = create_async_engine(TEST_DATABASE_URL, pool_size=5)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield engine
    await engine.dispose()


@pytest.fixture
async def session(engine):
    async with engine.connect() as conn:
        outer = await conn.begin()
        async with AsyncSession(bind=conn, join_transaction_mode="create_savepoint") as s:
            yield s
        await outer.rollback()

Чтобы набор оставался быстрым

• Создавайте данные фабриками внутри теста, а не огромными фикстурами на модуль.
• Запускайте тесты параллельно с pytest-xdist — у каждого процесса своя тестовая база.
• Отмечайте медленные тесты маркером slow и гоняйте их в CI отдельно.

Итог: один цикл и один пул на сессию, транзакция с откатом на каждый тест. На проекте из доклада это сократило набор с четырёх минут до 50 секунд.