Тестирование FastAPI: TestClient, AsyncClient, dependency_overrides
#Система внедрения зависимостей FastAPI делает его одним из самых тестируемых веб-фреймворков Python. Вместо патчинга функций в месте вызова через `unittest.mock` вы заменяете целые цепочки зависимостей на уровне приложения до запуска теста -- чисто, не затрагивая продакшн-код.
## Два тестовых клиента
**`TestClient`** (синхронный, из `fastapi.testclient`):
```python
from fastapi.testclient import TestClient
from myapp.main import app
client = TestClient(app)
def test_get_items():
response = client.get('/items/')
assert response.status_code == 200
```
`TestClient` построен на httpx. Он запускает ваше ASGI-приложение синхронно внутри процесса теста -- реальный сервер не запускается. Правильный выбор для большинства тестов: простой, быстрый и работает без `pytest-asyncio`. Тесты -- обычные `def`-функции.
**`httpx.AsyncClient`** (асинхронный):
```python
# pytest.ini: asyncio_mode = auto
import httpx
import pytest
from myapp.main import app
@pytest.fixture
async def async_client():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test') as client:
yield client
async def test_get_items(async_client):
response = await async_client.get('/items/')
assert response.status_code == 200
```
Используйте `AsyncClient`, когда тестируемый код по своей природе асинхронен и нужно проверить async-поведение -- тестирование конкурентных запросов, асинхронных генераторов или потоковых ответов. Требует `pytest-asyncio` с `asyncio_mode = auto` в `pytest.ini`.
## `app.dependency_overrides` -- основной паттерн тестирования
В FastAPI маршруты объявляют свои зависимости как параметры функции:
```python
from fastapi import Depends
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get('/items/')
def list_items(db: Session = Depends(get_db)):
return db.query(Item).all()
```
В тестах заменяйте `get_db` фиктивной функцией, возвращающей in-memory хранилище, -- без изменения продакшн-кода:
```python
def fake_get_db():
yield {'items': []}
app.dependency_overrides[get_db] = fake_get_db
```
Переопределение применяется ко всем запросам через приложение, пока оно установлено. Всегда очищайте после завершения -- иначе переопределение сохраняется между тестовыми функциями и загрязняет последующие тесты:
```python
app.dependency_overrides.clear()
```
Стандартный паттерн оборачивает настройку и очистку в pytest-фикстуру с `yield`:
```python
@pytest.fixture
def client_with_fake_db():
store = []
def override():
yield store
app.dependency_overrides[get_db] = override
yield TestClient(app), store
app.dependency_overrides.clear() # всегда выполняется, даже при падении теста
```
## Какой клиент когда использовать
| Сценарий | Клиент |
|---|---|
| Тестирование запроса/ответа (статус, JSON-тело) | `TestClient` |
| Тестирование async-поведения или конкурентных запросов | `AsyncClient` |
| Потоковые ответы, WebSocket | `AsyncClient` |
| Простота и скорость | `TestClient` |
## События жизненного цикла в тестах
FastAPI-приложения часто имеют события запуска/завершения (инициализация пула соединений, загрузка ML-моделей). `TestClient` запускает их при использовании как контекстного менеджера:
```python
with TestClient(app) as client:
# startup уже выполнен
response = client.get('/items/')
# shutdown выполнен
```
Для тестов, проверяющих состояние, зависящее от запуска, всегда используйте форму контекстного менеджера.
## `yield` vs `return` в переопределениях зависимостей
Когда исходная зависимость использует `yield` (распространено для ресурсов, требующих завершения -- сессий БД, открытых файлов, сетевых соединений), ваш заменяющий переопределитель тоже **должен использовать `yield`**. Обычный `return` в замене молча пропускает завершение:
```python
# Оригинал -- зависимость с yield и завершением
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close() # <- это завершение выполняется после запроса
# НЕВЕРНО -- переопределение использует return; db.cleanup() никогда не выполнится
def fake_db_wrong():
return FakeSession() # FastAPI получает значение; фаза завершения отсутствует
# ВЕРНО -- переопределение использует yield; завершение выполняется после возврата обработчика
def fake_db_correct():
db = FakeSession()
try:
yield db
finally:
db.cleanup() # <- гарантированное завершение
```
FastAPI разрешает переопределение как генератор, если оно содержит `yield`. Если это обычная функция (без `yield`), FastAPI вызывает её, берёт возвращаемое значение и пропускает логику завершения. Это не ошибка -- FastAPI не предупреждает вас -- поэтому баг незаметен.
Переопределение через `lambda` работает только для зависимостей без завершения:
```python
# OK -- get_store просто возвращает список, завершение не нужно
app.dependency_overrides[get_store] = lambda: []
# НЕВЕРНО -- get_db имеет завершение; lambda молча его пропускает
app.dependency_overrides[get_db] = lambda: FakeSession() # db.close() никогда не выполнится
```
## Переопределение сессии базы данных SQLAlchemy
> **Примечание о версии SQLAlchemy:** паттерн ниже использует `bind=` для `sessionmaker` и `Session` -- эти именованные аргументы были устаревшими в SQLAlchemy 1.4 и **удалены в SQLAlchemy 2.0**. На SQLAlchemy 2.x замените `sessionmaker(bind=engine)` на `sessionmaker(engine)` и используйте паттерн `join_transaction_mode` (см. документацию по тестированию SQLAlchemy 2.x). Подход с `dependency_overrides` и концепция изоляции тестов идентичны во всех версиях.
Переопределение зависимости сессии базы данных -- наиболее распространённый паттерн `dependency_overrides` в реальных FastAPI-проектах. Стандартная настройка тестов создаёт in-memory SQLite-базу данных для тестовой сессии, оборачивает каждый тест в откатываемую транзакцию и внедряет одну и ту же сессию как в тест, так и в приложение:
```python
# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import TestClient
from myapp.main import app
from myapp.database import get_db, Base
engine = create_engine('sqlite://', connect_args={'check_same_thread': False})
TestingSession = sessionmaker(bind=engine)
@pytest.fixture(scope='session', autouse=True)
def create_tables():
Base.metadata.create_all(engine)
yield
Base.metadata.drop_all(engine)
@pytest.fixture
def db_session():
connection = engine.connect()
transaction = connection.begin()
session = TestingSession(bind=connection)
yield session
session.close()
transaction.rollback() # <- каждый INSERT/UPDATE/DELETE из теста отменяется
connection.close()
@pytest.fixture
def client(db_session):
def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
yield TestClient(app)
app.dependency_overrides.clear()
```
Ключевые решения в этом паттерне:
- `scope='session'` для `create_tables` -- создание схемы выполняется один раз за весь прогон тестов, а не для каждого теста.
- `db_session` для каждого теста оборачивает его в транзакцию и откатывает при завершении. Каждый тест начинается с чистой схемой без пересоздания таблиц.
- Фикстура `client` внедряет ту же сессию, которую держит тест, поэтому можно читать записи приложения напрямую из `db_session` без дополнительного запроса через отдельное соединение.
## BackgroundTasks: поведение `TestClient` vs `AsyncClient`
`BackgroundTasks` FastAPI запускает функции после отправки ответа клиенту. В тестах поведение различается между двумя клиентами:
```python
from fastapi import BackgroundTasks
results = []
def record(value: str):
results.append(value)
@app.post('/jobs/')
def create_job(bt: BackgroundTasks):
bt.add_task(record, 'done')
return {'status': 'queued'}
```
**С `TestClient`** фоновые задачи выполняются **синхронно перед** тем, как `TestClient` возвращает ответ. К моменту выполнения утверждения задача уже завершена:
```python
def test_job_with_test_client():
results.clear()
response = client.post('/jobs/')
assert response.status_code == 200
assert 'done' in results # задача выполнилась синхронно до возврата TestClient
```
**С `AsyncClient` и `ASGITransport`** фоновые задачи тоже завершаются **перед** тем, как `await async_client.post(...)` возвращает результат. Starlette вызывает фоновые задачи как часть жизненного цикла ASGI-ответа, и `ASGITransport` управляет этим жизненным циклом синхронно внутри корутины теста:
```python
async def test_job_with_async_client(async_client):
results.clear()
response = await async_client.post('/jobs/')
assert response.status_code == 200
assert 'done' in results # задача уже выполнилась до возврата ответа
```
Оба тестовых клиента ведут себя одинаково: фоновые задачи завершаются до возврата объекта ответа. Это отличается от продакшна -- реальный сервер сначала отправляет HTTP-ответ, а затем выполняет задачи. В тестах с любым клиентом можно сразу проверять результаты задач после запроса.
## Тестирование кастомных обработчиков исключений
По умолчанию `TestClient` повторно вызывает любое серверное исключение в процессе теста. Если маршрут вызывает `ValueError`, тест видит `ValueError`, а не 500-ответ -- утверждение по HTTP-статусу невозможно.
Когда у вас есть кастомный обработчик исключений и вы хотите проверить, что он возвращает корректный статус и тело, отключите повторный вызов с `raise_server_exceptions=False`:
```python
from fastapi import Request
from fastapi.responses import JSONResponse
class AppError(Exception):
def __init__(self, code: int, message: str):
self.code = code
self.message = message
@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
return JSONResponse(status_code=exc.code, content={'error': exc.message})
@app.get('/boom')
def boom():
raise AppError(409, 'conflict')
```
```python
def test_custom_error_handler():
error_client = TestClient(app, raise_server_exceptions=False)
response = error_client.get('/boom')
assert response.status_code == 409
assert response.json() == {'error': 'conflict'}
```
Без `raise_server_exceptions=False` `TestClient` распространил бы `AppError` в тест и строки с утверждениями никогда бы не выполнились. Держите этот флаг в области тестов, специально проверяющих обработку ошибок -- для всех остальных тестов нужно, чтобы исключения распространялись и неожиданные серверные ошибки вызывали немедленное падение тестов.
## Загрузка файлов
Используйте параметр `files` для отправки multipart-загрузки файлов через `TestClient`. Значение -- словарь, сопоставляющий имя поля формы с кортежем `(filename, file-like-object, content-type)`:
```python
from fastapi import UploadFile, File
@app.post('/upload/')
async def upload_file(file: UploadFile = File(...)):
content = await file.read()
return {'filename': file.filename, 'size': len(content)}
```
```python
import io
def test_file_upload():
data = b'hello world'
response = client.post(
'/upload/',
files={'file': ('hello.txt', io.BytesIO(data), 'text/plain')},
)
assert response.status_code == 200
assert response.json() == {'filename': 'hello.txt', 'size': 11}
```
Формат кортежа: `(filename, file-like-object, content-type)`. Часть `content-type` необязательна -- если пропущена, httpx выводит её из расширения имени файла. Для бинарных или неизвестных файлов используйте `'application/octet-stream'`.
Для multipart-форм, смешивающих обычные поля с файлами, комбинируйте `data` и `files`:
```python
response = client.post(
'/upload/',
data={'description': 'profile picture'},
files={'avatar': ('avatar.png', io.BytesIO(png_bytes), 'image/png')},
)
```