테스트 코드
테스트가 필요한 이유
- 코드 변경 시 기존 기능이 정상적으로 작동하는지 보장하기 위해
- 자동화된 테스트를 통해 버그를 빠르게 발견하고 수정 가능
- 특히 API 개발에서는 요청/응답을 검증하는 것이 중요
- 유닛테스트
- 모델 테스트
- 라우터 테스트
- 통합 테스트
pytest
Python의 대표적인 테스트 프레임워크
pip install pytest
httpx
FastAPI에서 requests 대신 사용하는 비동기 HTTP 클라이언트
pip install httpx
TestClient
FastAPI 애플리케이션에 직접 요청을 보내고 응답 검증
종류
- 단위 테스트(Unit Test) : 개발 함수, 모듈 단위로 동작을 검증
- 통합 테스트(Integration Test) : 여러 모듈이 함꼐 동작할 때의 결과를 검증
- 엔드투엔드 테스트(E2E Test) : 실제 사용자 시나리오를 기반으로 전체 시스템을 테스트
환경구성
- pytest 및 httpx 패키지 설치 :
pip install pytest httpx- FastAPI 설정: 테스트 하고싶은 엔드포인트 작성 : 파일명 main.py
- 테스트 코드 실행,
pytest test_main.py
- 기본적인 pytest 테스트 코드 작성
- test_main.py
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_ping():
response = client.get("/ping")
assert response.status_code == 200
assert response.json() == {"msg": "pong"}
# pytest test_main.py테스트 시나리오
존재하지 않는 엔드포인트 테스트
def test_not_found():
response = client.get("/invalid-endpoint")
assert response.staus_code == 404필수 파라미터가 없을때 예외 처리
@app.get("/greet/")
async def greet(name: str):
return {"msg" : f"Hello, {name}!"}
def test_greet():
response = client.get("/greet/", params = {"name" : "Alice"})
assert response.status_code == 200
assert response.json() == {"msg" : "Hello, Alice!"}
def test_greet_missing_param():
response = client.get("/greet/")
assert response.status_code == 422단위 테스트 (Unit Test)
-
단위 테스트란?
- 코드의 개별 기능(함수, 클래스, 모듈) 단위로 동작을 검증하는 테스트
- 데이터베이스, 외부 API 호출 등 의존성이 없는 상태에서 실행됨
-
단위 테스트의 필요성
- 코드 변경 후에도 기능이 정상적으로 유지되는지 보장
- 버그를 조기에 발견하여 수정 비용 절감
- 빠른 실행 속도로 자주 실행 가능
-
최대한 많은 케이스를 찾아서 테스트
| 검증 항목 | 설명 |
|---|---|
| 정상적인 API 동작 | 요청이 정상적으로 처리되는지 확인 |
| 잘못된 입력값 테스트 | 잘못된 데이터 입력 시 FastAPI가 예외를 올바르게 처리하는지 검증 |
| 엣지 케이스 (경계값) | 0, 큰 숫자, 빈 값 등 예상치 못한 입력 처리 |
| FastAPI의 의존성 주입 테스트 | API 엔드포인트에서 의존성 주입(Depends)을 활용한 테스트 |
from fastapi import FastAPI
app = FastAPI ()
def subtract(a: int, b: int):
# 빼기 연산 함수
return {" result": a - b}
@app.get("/math/subtract")
def subtract_route(a: int, b: int):
# FastAPI 엔드포인트 (subtract API)
return subtract(a, b)● 정상적인 뺄셈
● 음수 포함 뺄셈
● 0을 포함한 뺄셈
● 잘못된 입력값
● 경계값 (최소, 최대값)Depends를 활용한 단위 테스트
app = FastAPI()
def get_default_number():
# 의존성 함수
return 10
def subtract(a: int, b:int, default= int = Depends(get_default_number)):
return {"result" : a-b-default}
@app.get("/math/subtract")
async def subtract_endpoint(a: int, b:int, default= int = Depends(get_default_number)):
return subtract(a, b, default)- 의존성 주입 함수에 dependency_overrides로 테스트 데이터를 덮어씌움
통합 테스트 (Integration Test)
- 통합 테스트의 필요성
- 단위 테스트만으로는 모듈 간 연결이 올바르게 동작하는지 확인할 수 없음
- API와 데이터베이스 간의 데이터 흐름이 정상적으로 이루어지는지 검증
필요- 다양한 API 요청 조합을 테스트하여 전체 시스템이 예상대로 동작하는지
확인
| 구분 | 단위 테스트 (Unit Test) | 통합 테스트 (Integration Test) |
|---|---|---|
| 테스트 대상 | 개별 함수, 모듈 | 여러 모듈, DB, API 간 연동 |
| 실행 속도 | 빠름 | 상대적으로 느림 |
| 의존성 | 없음 (Mock 사용) | 실제 DB, API 필요 |
| 목적 | 코드 로직 검증 | 전체 흐름 테스트 |
# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel
from database import get_db, Product
app = FastAPI()
# 요청 데이터 모델 정의
class ProductCreate(BaseModel):
name: str
price: int
# 제품 추가 API
@app.post("/products/")
async def create_product(product: ProductCreate, db: Session = Depends(get_db)):
product = Product(name=product.name, price=product.price)
db.add(product)
db.commit()
db.refresh(product)
return product
# 제품 조회 API
@app.get("/products/{product_id}")
async def read_product(product_id: int, db: Session = Depends(get_db)):
product = db.query(Product).filter(Product.id == product_id).first()
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
# database.py
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class Product(Base):
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
price = Column(Integer)
Base.metadata.create_all(bind=engine)
# DB 세션 의존성
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
# test_main.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import TestClient
from database import Base, get_db, Product
from main import app
# 테스트용 데이터베이스 설정
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
client = TestClient(app)
# 한 번만 실행하여 전체 테스트에서 DB를 공유
@pytest.fixture(scope="module")
def setup_test_db():
Base.metadata.create_all(bind=engine) # 새로 생성
db = TestingSessionLocal()
# 초기 데이터 삽입
test_product = Product(id=1, name="Laptop", price=1500)
db.add(test_product)
db.commit()
yield db # 모든 테스트에서 공유됨
db.close()
Base.metadata.drop_all(bind=engine) # 모든 테스트 종료 후 DB 삭제
# 제품 추가 API 테스트
def test_create_product(setup_test_db):
response = client.post("/products/", json={"name": "Phone", "price": 800})
assert response.status_code == 200
assert response.json()["name"] == "Phone"
assert response.json()["price"] == 800
# 존재하는 제품 조회 테스트 (Laptop & Phone 조회 가능)
def test_read_product(setup_test_db):
response = client.get("/products/1") # ID=1 Laptop 조회
assert response.status_code == 200
assert response.json()["name"] == "Laptop"
assert response.json()["price"] == 1500
response = client.get("/products/2") # ID=2 Phone 조회
assert response.status_code == 200
assert response.json()["name"] == "Phone"
assert response.json()["price"] == 800
# 존재하지 않는 제품 조회 테스트 (404 반환)
def test_read_product_not_found(setup_test_db):
response = client.get("/products/999") # 존재하지 않는 ID 요청
assert response.status_code == 404
assert response.json()["detail"] == "Product not found"테스트 코드 최적화 개념
- 테스트 코드가 많아질수록 발생하는 문제
- 실행 속도가 느려짐
- 중복 코드가 많아짐
- 테스트 케이스가 늘어남
테스트 최적화를 위한 핵심 기법
| 기법 | 설명 |
|---|---|
| Mocking | DB나 외부 API 호출 없이 가짜 데이터를 반환하도록 처리 |
| Fixture | 반복된 설정을 줄여 코드 재사용성 향상 |
| Parametrize | 같은 테스트를 여러 입력값으로 실행 |
| 성능 측정 | 테스트 실행 속도를 체크하여 병목 구간 식별 |
Mocking
- Mocking: DB나 외부 API 호출 없이 가짜 데이터를 반환하도록 처리
- 실제 데이터베이스와의 연결 없이 API 로직만 검증 가능
- 테스트 시 불필요한 DB 접근을 막고, 실행 속도를 개선
- 컴퓨터 입장에서 실제 DB를 연결하지 않고 뒤에 값이 저걸 쓰려나 보다 라고 생각을함
- 그래서 부하가 안걸림, 기존의 사용에
.return_value만 붙여주면 됨
| Mocking 방식 | 사용 예제 | 특징 |
|---|---|---|
| MagicMock() | mock_db = MagicMock() | 객체를 가짜로 생성하여 동작을 정의 가능 |
| patch() | @patch("main.get_db") | 특정 함수를 가짜로 대체 |
| AsyncMock() | mock_async = AsyncMock() | 비동기 함수(Mock)를 지원 |
- patch 에서 앞의 main에 불러와진 모듈은 전부 사용가능 그래서 get_db가 import 되어있어서 사용 가능
- @patch("모듈.함수") 에서 "함수"부분이 mock_함수명 형식으로 매칭
- 여러개의 @patch를 사용할 경우, 가장 마지막 @patch가 첫번쨰 매개변수로 등록
- Mocking된 객체는 MagicMokc() 타입이므로, .return_value를 설정해서 동작을 정의할 수 있음
- 정의된 순서대로 아래부터 바로 적용할 함수에 순차적으로 적어야 함
@patch("auth.verify_password")
@patch("auth.create_token")
def test_login(mock_create_token, mock_verify_password, mock_db, username, password, expected_status):- 기존방식
@app.get("/users/(user_id}")
async def get_user(user_id: int, db: Session = Depends (get_db)):
user = db.query(User).filter(User.id = user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
def test_get_user():
response = client.get("/users/1")
assert response.status_code ==200- Mocking 활용
@app.get("/users/(user_id}")
async def get_user(user_id: int, db: Session = Depends (get_db)):
user = db.query(User).filter(User.id = user_id).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
# mocking
from unittest.comk import MacigMock
def test_get_user_with_mock():
mock_db = MagicMock()
mock_db.query.return_value.filter.return_value.first.return_value= User(id=1, name="Mock User", email = "mock@example.com")
response = client.get("/users/1", dependencies={"db": mock_db})
assert response.status_code == 200
assert response.json()["name"] == "Mock User"
# patch
@patch("main.get_db") # 데이터 베이스가 있는거 같다 그냥 생각을 하게 만듬 컴퓨터에(실제 DB가 없어도 되는거임)
def test_get_user_with_mock(mock_get_db):
mock_db = MagicMock()
mock_db.query.return_value.filter.return_value.first.return_value =User(id=1, name="Mock User", email ="mock@example.com")
# Patch를 적용하면서 FastAPI의 의존성 오버라이드 설정
mock_get_db.return_value = mock_db
app.dependency_overrides[get_db] = lambda: mock_db
response = client.get("/users/1")
assert response.status_code == 200
assert response.json()["name"] == "Mock User"
# 테스트 완료 후 의존성 초기화
app.dependency_overrides.clear()- AsyncMock
- API 최적화를 위해 사용 할 경우도 있음 대부분 MagicMock 으로 처리
from unittest.mock import AsyncMock
mock_db = AsyncMock()
mock_db.query.return_value.filter.return_value.first.return_value = User(id=1, name="Async Mock User", email="asycn@example.com")Fixture
- Fixture: 테스트 실행 전 공통 환경을 설정하는 기능
- 여러 테스트에서 중복된 코드 없이 동일한 설정 공유
- DB 초기화, API 클라이언트 생성, 기본 데이터 삽입 등에 사용
- 공통 환경을 설정하는 기능
- Fixture의 scope 설정
| Scope 옵션 | 설명 |
|---|---|
| function | 각 테스트 함수 실행 전후로 초기화 (기본값) |
| class | 클래스 단위로 실행되며, 클래스 내부의 모든 테스트에서 공유 |
| module | 파일(.py 파일) 단위로 실행되며, 한 번만 초기화 후 공유 |
| session | pytest 실행 시 한 번만 실행 (가장 긴 수명) |
-
function : 각각의 함수에 각각의 데이터베이스를 세팅
- 한 함수가 끝나면 그 데이터베이스가 초기화 됨
-
module : 중복된 데이터를 줄이면서 테스트 하기위해, 한 파일안에서 중복된 데이터 사용
-
예시
def test_create_user():
db = TestingSessionLocal()
Base.metadata.create_all(bind=engine)
...
Base.metadata.drop_all(bind_engine)
db.close()
# 변경
import pytest
@pytest.fixture(scope="module")
def test_db():
""" 한 번만 실행하여 전체 테스트에서 공유하는 데이터베이스 설정 """
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
db = TestingSessionLocal()
yield db # 모든 테스트에서 공유됨
db.close()
Base.metadata.drop_all(bind=engine) # 모든 테스트 종료 후 DB 삭제Parametrize
- Parametrize: 동일한 테스트를 다양한 입력값으로 실행할 때 사용
- @pytest.mark.parametrize를 활용하여 자동 반복 실행
- 코드 중복 없이 테스트를 간결하게 유지 가능
-
데코레이터 밑의 인덱스로 처리된 값들을 데코레이터 안의 파라미터에 맞춰 데이터들을 순차적으로 전부 테스트함
-
기존방식
@app.get("/math/add")
async def add_numbers(a: int, b: int):
return {"result" : a + b}
def test_add_numbers_1():
response = client.get("/math/add?a=3&b=5")
assert response.status_code == 200
assert response.json()["result"] == 8
def test_add_numbers_2():
response = client.get("/math/add?a=-2&b=7")
assert response.status_code == 200
assert response.json()["result"] == 5- Parametrize 적용
import pytest
@pytest.mark.parametrize("a, b, expected", [
(3, 5, 8),
(-2, 7, 5),
(10, -10, 0),
(0, 0, 0)
])
def test_add_numbers(a, b, expected):
response = client.get(f"/math/add?a={a}&b={b}")
assert response.status_code == 200
assert response.json()["result"] == expected
@app.post("/users/")
async def create_user(user: dict):
return {"id": 1, "name": user["name"], "email": user["email"]}
@pytest.mark.parametrize("user_data, expected_status", [
({"name": "Alice", "email": "alice@example.com"}, 200),
({"name": "Bob"}, 422), # 이메일 누락
({"email": "invalid.com"}, 422), # 이름 누락 및 이메일 형식 오류
])
def test_create_user(user_daa, expected_status):
response = client.post("/users/", josn=user_data)
assert response.status_code == expected_status성능 측정
- 테스트 실행 시간이 길어질 경우, 특정 API의 성능을 분석해야 함
- time.time()을 사용하여 API 응답 시간을 측정
- 특정 API가 성능 저하를 일으키는지 파악 가능\
import time
def test_api_performance():
start_time = time.time()
response = client.get("/math/add?a=10000&b=20000")
end_time = time.time()
assert response.status_code == 200
assert response.json()["result"] == 30000
print(f"API 실행 시간: {end_time -start_time:.5f}초"}