lenec ru

← все посты

Андрей Крылов Подписаться

Тестирование FastAPI с pytest: фикстуры для БД и httpx-клиент

13K

Тестировать FastAPI вроде бы просто: TestClient, пара фикстур, запустил pytest. Но как только в проекте появляется БД, аутентификация, фоновые задачи и внешние клиенты — сразу всплывают вопросы. Где брать базу? Как изолировать тесты? Как не сломать всё одним падающим тестом? Как тестировать асинхронные эндпоинты, не путаясь с event loop?

Расскажу, как у нас собрана базовая тестовая инфраструктура для FastAPI-сервисов: какой клиент, как готовить БД, какие фикстуры обязательны.

Выбор клиента: TestClient или httpx

FastAPI поставляется с TestClient на базе Starlette. Он запускает приложение в синхронном режиме (через requests), что удобно для простых тестов. Но если у тебя async-эндпоинты с BackgroundTasks, lifespan-эвентами и прочей асинхронной механикой — рано или поздно появляются странности.

Лично мы давно перешли на httpx.AsyncClient с ASGITransport. Это даёт настоящий асинхронный клиент, который работает с приложением в том же event loop, что и сам тест.

import pytest
from httpx import AsyncClient, ASGITransport
from app.main import app

@pytest.fixture
async def client():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c

@pytest.mark.asyncio
async def test_health(client):
    resp = await client.get("/health")
    assert resp.status_code == 200

Никаких вызовов через сеть, никакого слушающего сокета — приложение опрашивается напрямую через ASGI-протокол. Скорость отличная, поведение точное.

pytest-asyncio: режим работы

Чтобы async-тесты вообще работали в pytest, нужен pytest-asyncio. Ставишь:

pip install pytest-asyncio

И в pyproject.toml или pytest.ini:

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

Режим auto избавляет от необходимости лепить @pytest.mark.asyncio на каждый тест. Все async def-тесты автоматически считаются асинхронными. Если оставишь strict — будешь забывать декоратор и получать тесты, которые молча пропускаются.

База: dedicated database для тестов

Не используй продовую БД. Не используй dev-базу, которой пользуется ещё кто-то. Не пользуйся SQLite вместо Postgres, если в проде Postgres — поведение разное (чувствительность к регистрам, типы, индексы, JSONB).

Базовая схема:

  • Поднимаем тестовый Postgres (через docker-compose или testcontainers).
  • Создаём отдельную БД на старте сессии тестов.
  • Прогоняем миграции один раз.
  • Каждый тест выполняется в транзакции, которая откатывается в конце.

Это даёт изоляцию между тестами и быстрый прогон, потому что миграции не повторяются.

Фикстура для движка и сессии

Условные фикстуры на уровне сессии (session) и на уровне теста (function):

import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker

TEST_DSN = "postgresql+asyncpg://test:test@localhost:5433/test"

@pytest.fixture(scope="session")
async def engine():
    eng = create_async_engine(TEST_DSN, echo=False)
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield eng
    async with eng.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
    await eng.dispose()

@pytest.fixture
async def db_session(engine):
    async with engine.connect() as conn:
        trans = await conn.begin()
        async_session = async_sessionmaker(bind=conn, expire_on_commit=False)
        session = async_session()
        try:
            yield session
        finally:
            await session.close()
            await trans.rollback()

Каждый тест получает чистую транзакцию, после теста — откат. Тесты независимы и не текут друг в друга.

Если используешь session.commit() внутри тестируемого кода (а ты используешь, потому что иначе сервис не работает), смотри в сторону вложенных транзакций через SAVEPOINT. SQLAlchemy умеет: ставишь обработчик after_transaction_end, который пересоздаёт savepoint.

Подменяем БД в приложении

Чтобы FastAPI в тесте использовал нашу тестовую сессию, переопределяем зависимость:

@pytest.fixture
async def client(db_session):
    async def override_get_session():
        yield db_session

    app.dependency_overrides[get_session] = override_get_session
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

Эта связка — основа: один db_session, одна транзакция, один клиент. Все эндпоинты внутри теста ходят в одну и ту же транзакцию, и она откатывается в конце. Можно проверять, что эндпоинт сделал именно то, что ты ожидаешь, читая через ту же сессию.

Фикстуры с данными

Базовый паттерн — фикстуры, создающие объекты в БД и возвращающие их.

@pytest.fixture
async def user(db_session):
    user = User(email="u@example.com", name="Test User")
    db_session.add(user)
    await db_session.flush()
    return user

@pytest.fixture
async def auth_client(client, user):
    # допустим, у нас есть способ выдать токен по user.id
    token = make_token(user.id)
    client.headers["Authorization"] = f"Bearer {token}"
    return client

Не делай гигантскую фикстуру на 200 строк, которая создаёт «всё что нужно». Делай маленькие, композируй их. Pytest сам разрулит порядок вызова и переиспользование.

Изоляция внешних клиентов

HTTP-клиенты, S3, Redis — всё, что ходит во внешние сервисы, в тестах надо подменять. Я предпочитаю явные фейки через dependency_overrides, а не monkeypatch на уровне импорта.

class FakePaymentClient:
    def __init__(self):
        self.calls = []
    async def charge(self, order_id, amount):
        self.calls.append((order_id, amount))
        return {"status": "ok"}

@pytest.fixture
async def payment_client():
    return FakePaymentClient()

@pytest.fixture
async def client(db_session, payment_client):
    async def override_session():
        yield db_session
    app.dependency_overrides[get_session] = override_session
    app.dependency_overrides[get_payment_client] = lambda: payment_client
    ...

Плюс: ты можешь в тесте проверить, что фейк-клиент был вызван правильным образом — assert payment_client.calls == [(order.id, 100)]. Никаких mock.assert_called_with, всё прозрачно.

Тесты с миграциями

Хороший вопрос: запускать миграции в тестах или строить схему через Base.metadata.create_all? Мой ответ: и то, и другое.

  • В юнит/интеграционных тестах — create_all. Быстро, без зависимостей от Alembic.
  • В отдельной CI-стадии — прогон миграций на чистой БД и затем апгрейд/даунгрейд. Это ловит ошибки самих миграций.

Не смешивай: если миграции прогоняются вместе с тестами, ты тратишь секунды на каждый запуск, а упавшая миграция роняет всё подряд.

Тестируем фоновые задачи

Если эндпоинт использует BackgroundTasks, в тесте они выполняются после возврата ответа. TestClient и AsyncClient ждут их завершения автоматически.

Если у тебя arq/Celery — в тестах не запускаешь воркер. Подменяешь pool/delay:

@pytest.fixture
async def fake_queue(monkeypatch):
    enqueued = []
    async def fake_enqueue(name, *args, **kwargs):
        enqueued.append((name, args, kwargs))
    monkeypatch.setattr(arq_pool, "enqueue_job", fake_enqueue)
    return enqueued

В тесте проверяешь, что задача поставлена в очередь с правильными аргументами. Сама логика задачи тестируется отдельно как обычная функция.

Что важно помнить

  • Используй httpx.AsyncClient с ASGITransport. Это будущее тестов FastAPI.
  • Реальный Postgres вместо SQLite. Поведение отличается в важных мелочах.
  • Транзакция на тест с откатом в конце — золотой стандарт изоляции.
  • Подменяй внешние сервисы через dependency_overrides, не через monkeypatch.
  • Маленькие фикстуры, композируй их, не делай божественную setup_everything.
  • Миграции отдельной стадией, тесты — на create_all ради скорости.

Когда тестовая инфраструктура устроена так, написание новых тестов перестаёт быть болью. Запустил — упало — починил — побежал дальше. Тесты должны помогать, а не отбирать энергию.

Комментарии 0

  • Будьте первым, кто оставит комментарий.

Войдите, чтобы оставить комментарий.