FastAPI 테스팅
FastAPI는 TestClient (동기)와 AsyncClient (비동기)로 API를 테스트합니다.
TestClient 기초
# pip install httpx
from fastapi import FastAPI
from fastapi.testclient import TestClient
import pytest
app = FastAPI()
fake_db: dict[int, dict] = {}
next_id = 1
@app.post("/users/", status_code=201)
def create_user(name: str, email: str):
global next_id
user = {"id": next_id, "name": name, "email": email}
fake_db[next_id] = user
next_id += 1
return user
@app.get("/users/{user_id}")
def get_user(user_id: int):
from fastapi import HTTPException
if user_id not in fake_db:
raise HTTPException(status_code=404, detail="User not found")
return fake_db[user_id]
# 테스트
client = TestClient(app)
def test_create_user():
response = client.post("/users/?name=Alice&email=alice@example.com")
assert response.status_code == 201
data = response.json()
assert data["name"] == "Alice"
assert "id" in data
def test_get_user():
# 먼저 생성
create_resp = client.post("/users/?name=Bob&email=bob@test.com")
user_id = create_resp.json()["id"]
# 조회
response = client.get(f"/users/{user_id}")
assert response.status_code == 200
assert response.json()["name"] == "Bob"
def test_user_not_found():
response = client.get("/users/9999")
assert response.status_code == 404
assert "not found" in response.json()["detail"].lower()
pytest fixture와 DB 격리
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, StaticPool
from sqlalchemy.orm import sessionmaker
from database import Base, get_db, User
from main import app
# 테스트용 메모리 DB
TEST_DATABASE_URL = "sqlite:///:memory:"
test_engine = create_engine(
TEST_DATABASE_URL,
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
TestingSessionLocal = sessionmaker(bind=test_engine)
@pytest.fixture(autouse=True)
def setup_db():
"""각 테스트마다 새 테이블 생성·삭제"""
Base.metadata.create_all(bind=test_engine)
yield
Base.metadata.drop_all(bind=test_engine)
@pytest.fixture
def db_session():
session = TestingSessionLocal()
try:
yield session
finally:
session.close()
@pytest.fixture
def client(db_session):
"""DB 의존성을 테스트 DB로 교체"""
def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
with TestClient(app) as test_client:
yield test_client
app.dependency_overrides.clear()
def test_create_user(client):
response = client.post(
"/users/",
json={"name": "Alice", "email": "alice@example.com", "password": "secret"},
)
assert response.status_code == 201
assert response.json()["email"] == "alice@example.com"
def test_duplicate_email(client):
data = {"name": "Alice", "email": "dup@test.com", "password": "pass"}
client.post("/users/", json=data)
response = client.post("/users/", json=data)
assert response.status_code == 400
인증 테스트
import pytest
from fastapi.testclient import TestClient
from main import app
@pytest.fixture
def client():
return TestClient(app)
@pytest.fixture
def auth_headers(client):
"""로그인 후 헤더 반환"""
response = client.post(
"/auth/login",
data={"username": "alice", "password": "password123"},
)
token = response.json()["access_token"]
return {"Authorization": f"Bearer {token}"}
def test_login_success(client):
response = client.post(
"/auth/login",
data={"username": "alice", "password": "password123"},
)
assert response.status_code == 200
assert "access_token" in response.json()
def test_login_wrong_password(client):
response = client.post(
"/auth/login",
data={"username": "alice", "password": "wrong"},
)
assert response.status_code == 401
def test_protected_endpoint_without_token(client):
response = client.get("/users/me")
assert response.status_code == 401
def test_protected_endpoint_with_token(client, auth_headers):
response = client.get("/users/me", headers=auth_headers)
assert response.status_code == 200
assert response.json()["username"] == "alice"
def test_admin_only_endpoint(client, auth_headers):
"""alice는 admin 역할"""
response = client.get("/admin/users", headers=auth_headers)
assert response.status_code == 200
pytest-asyncio — 비동기 테스트
# pip install pytest-asyncio httpx
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from main import app
@pytest.mark.asyncio
async def test_async_endpoint():
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
response = await client.get("/")
assert response.status_code == 200
# 비동기 fixture
@pytest_asyncio.fixture
async def async_client():
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as client:
yield client
@pytest.mark.asyncio
async def test_create_item(async_client):
response = await async_client.post(
"/items/",
json={"name": "Test Item", "price": 10.5},
)
assert response.status_code == 201
assert response.json()["name"] == "Test Item"
@pytest.mark.asyncio
async def test_concurrent_requests(async_client):
import asyncio
responses = await asyncio.gather(*[
async_client.get(f"/items/{i}")
for i in range(10)
])
assert all(r.status_code in (200, 404) for r in responses)
정리
| 도구 | 용도 |
|---|---|
TestClient | 동기 테스트 (대부분의 경우) |
AsyncClient | 비동기 엔드포인트 테스트 |
dependency_overrides | DB 등 의존성 교체 |
pytest.fixture | 클라이언트, DB 세션 공유 |
pytest-asyncio | async 테스트 함수 지원 |
FastAPI 테스트는 실제 HTTP 요청 흐름(미들웨어 포함)을 검증하는 통합 테스트 방식으로 작성하세요.