Тестирование FastAPI с pytest

Охватываемые темы: Установка, TestClient (синхронный), Тестовая БД, Тесты с фикстурами.

Установка

pip install pytest httpx pytest-asyncio

TestClient (синхронный)

from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello, World!"}

def test_create_task():
    response = client.post("/tasks/", json={"title": "Test Task"})
    assert response.status_code == 201
    assert response.json()["title"] == "Test Task"

Тестовая БД

# conftest.py
import pytest
from sqlmodel import SQLModel, create_engine, Session
from fastapi.testclient import TestClient

TEST_DATABASE_URL = "sqlite:///./test.db"
test_engine = create_engine(TEST_DATABASE_URL, connect_args={"check_same_thread": False})

@pytest.fixture(autouse=True)
def setup_db():
    SQLModel.metadata.create_all(test_engine)
    yield
    SQLModel.metadata.drop_all(test_engine)

@pytest.fixture
def session():
    with Session(test_engine) as session:
        yield session

@pytest.fixture
def client(session):
    def override_get_session():
        yield session

    app.dependency_overrides[get_session] = override_get_session
    yield TestClient(app)
    app.dependency_overrides.clear()

Тесты с фикстурами

def test_get_task(client, session):
    # Создать данные напрямую в БД
    task = Task(title="Test Task", status="todo")
    session.add(task)
    session.commit()

    response = client.get(f"/tasks/{task.id}")
    assert response.status_code == 200
    assert response.json()["title"] == "Test Task"

def test_get_task_not_found(client):
    response = client.get("/tasks/9999")
    assert response.status_code == 404

Аутентифицированные запросы

def test_protected_endpoint(client):
    # Без токена
    response = client.get("/tasks/")
    assert response.status_code == 401

    # С токеном
    token = create_access_token({"sub": "1"})
    response = client.get("/tasks/", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 200

Async тесты

import pytest
import httpx

@pytest.mark.asyncio
async def test_async():
    async with httpx.AsyncClient(app=app, base_url="http://test") as ac:
        response = await ac.get("/tasks/")
    assert response.status_code == 200