[FastAPI] pytest 사용법 (14)

 

 

1. pytest 란

1.1. 기본 설명

pytest는 함수 이름이 test_로 시작하는 함수를 찾아서 실행

함수 정의 시, assert 키워드를 이용해 성공 여부를 검사. True인 경우 성공

 

pytest 예시코드

def test_addition():
    assert 1 + 2 == 3

def test_string_upper():
    assert "pytest".upper() == "PYTEST"

def test_fail():
    assert 1 + 1 == 3, "이 테스트는 실패해야 합니다!"

 

특정 파일 테스트 실행

• pytest 키워드 사용

pytest test_sample.py

 

특정 디렉토리 안의 모든 테스트 실행

pytest tests/test_routers/

 

특정 함수만 테스트 실행

• -k 옵션을 사용

pytest tests/test_items.py -k "test_read_item"

 

---

1.2. Mock 이란

설명

• 특정 함수를 가짜 함수로 변경해 줌

사용 방법

• 매개변수 사용방법 : ("변경될 함수", "변경할 함수")

monkeypatch.setattr("app.repositories.item_repository.get_item_by_id", mock_get_item_by_id)

 

언제 Mock을 사용해야 할까?

• 데이터베이스 없이 독립적인 단위 테스트(Unit Test)를 할 때
• 서비스(Service) 로직이 DB와 강하게 연결될 필요가 없을 때
• 비즈니스 로직만 테스트하고 싶을 때
• 테스트 실행 속도를 빠르게 하고 싶을 때
• 외부 API를 호출하는 경우

Mock을 사용하지 않는 경우

• 통합 테스트(Integration Test)를 할 때
• ORM과의 연동이 잘 되는지 확인해야 할 때

 

---

2. 사용 방법

2.1. 디렉토리 구조

일반적으로, FastAPI 애플리케이션의 디렉토리 구조는 다음과 같은 형태로 만든다.

app과 test로 파일을 분리한 것을 볼 수 있다.

테스트하는 내용은 router, service, repository 등을 진행한다.

my_fastapi_project/
│
├── app/
│   ├── main.py
│   ├── routers/
│   │   └── items.py
│   ├── services/
│   │   └── item_service.py
│   ├── repositories/
│   │   └── item_repository.py
│   ├── models/
│   │   └── item.py
│   ├── database.py
│   └── dependencies.py
│
└── tests/
    ├── __init__.py
    ├── test_routers/
    │   └── test_items.py
    ├── test_services/
    │   └── test_item_service.py
    ├── test_repositories/
    │   └── test_item_repository.py
    ├── conftest.py

 

---

2.1. 공통 설정파일 만들기

• 기본적으로 tests/conftest.py에서 작업을 한다.
• @pytest.fixture 데코레이터를 이용한다.
• @pytest.fixture로 등록된 함수는 같은 테스트 파일에서 직접 호출할 필요 없이, 테스트 함수에서 인자로 전달하면 pytest가 자동으로 실행
• 테스트용 JWT 인증 토큰을 만들어 준다.

import pytest
from app.auth import create_access_token

@pytest.fixture
def test_token():
    """테스트용 JWT 토큰 생성"""
    return create_access_token({"sub": "testuser"})

 

---

2.2. 라우터 테스트

app/auth.py

• 일반적으로 라우터에서는 인증과정이 필요하므로, 인증관련 예제코드를 만들어 두겠다.

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from fastapi import HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer

# 시크릿 키 & 알고리즘
SECRET_KEY = "mysecretkey"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    """JWT 토큰 생성 함수"""
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str = Depends(oauth2_scheme)):
    """JWT 토큰 검증 함수"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

 

app/routers/items.py

• 토큰 검증을 의존성 주입을 통해 실행한다.

from fastapi import APIRouter, Depends
from app.auth import verify_token

router = APIRouter()

@router.get("/items/")
def read_items(user: dict = Depends(verify_token)):
    return {"message": "You are authenticated!", "user": user}

 

tests/test_routers/test_items.py

• client = TestClient(app) 를 이용해 테스트용 클라이언트를 정의한다.
• client.get("url") 로 라우터를 실행시킨다.
• @pytest.fixture로 등록했던 test_token 함수를 불러와 토큰을 발급받는다.
• test_token는 변수형태로 사용 가능하며, test_token 메소드의 실행 결과 값이 담겨있음

import pytest
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_read_items(test_token):
    """JWT 인증이 필요한 라우터 테스트"""
    headers = {"Authorization": f"Bearer {test_token}"}
    
    response = client.get("/items/", headers=headers)
    
    assert response.status_code == 200
    assert response.json()["message"] == "You are authenticated!"

 

---

2.3. 서비스 계층 테스트 예시

app/services/item_service.py

• 서비스 예시코드

from sqlalchemy.orm import Session
from app.repositories.item_repository import get_item_by_id

def get_item_details(db: Session, item_id: int):
    item = get_item_by_id(db, item_id)
    if not item:
        return {"error": "Item not found"}
    return {"item_id": item.id, "name": item.name}

 

tests/test_services/test_item_service.py

• ✅  monkeypatch.setattr() : 특정 속성(함수, 메서드, 클래스 등)을 일시적으로 변경(Mock)하는 메서드
• mock_get_item_by_id()를 사용하여 repository의 동작을 모의(Mock)
• db는 필요 없으니 None을 넣어준다

import pytest
from app.services.item_service import get_item_details

# Mock Repository 함수
def mock_get_item_by_id(db, item_id: int):
    class MockItem:
        id = item_id
        name = f"Mock Item {item_id}"
    return MockItem()

@pytest.fixture
def mock_db(monkeypatch):
    monkeypatch.setattr("app.repositories.item_repository.get_item_by_id", mock_get_item_by_id)

def test_get_item_details(mock_db):
    result = get_item_details(None, 1)  # mock DB를 사용하므로 실제 DB 세션이 필요 없음
    assert result == {"item_id": 1, "name": "Mock Item 1"}

 

---

3. 통합 테스트

통합 테스트(Integration Test)는 애플리케이션의 여러 구성 요소(데이터베이스, API, 서비스 로직 등)를 실제로 연결해서 동작을 검증하는 테스트. 즉 전체 시스템이 정상적으로 작동하는지 확인

여기서는 실제 데이터베이스를 사용함.

 

app/routers/items.py

• router 예시
• get 하나, post 하나씩 예시코드가 있음

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.auth import verify_token
from app.database import get_db
from app.models.item import Item
from pydantic import BaseModel

router = APIRouter()

class ItemCreate(BaseModel):
    name: str

@router.get("/items/")
def read_items(user: dict = Depends(verify_token)):
    return {"message": "You are authenticated!", "user": user}

@router.post("/items/")
def create_item(item: ItemCreate, db: Session = Depends(get_db), user: dict = Depends(verify_token)):
    new_item = Item(name=item.name)
    db.add(new_item)
    db.commit()
    db.refresh(new_item)
    return {"id": new_item.id, "name": new_item.name}

 

tests/conftest.py

• 테스트 이후, rollback을 통해 변경된 데이터베이스를 다시 원래대로 돌려놓음

import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
from app.auth import create_access_token

@pytest.fixture
def test_token():
    """테스트용 JWT 토큰 생성"""
    return create_access_token({"sub": "testuser"})


DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"
engine = create_engine(DATABASE_URL)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture(scope="function")
def db():
    """🔥 기존 데이터를 유지하면서 테스트 실행"""
    connection = engine.connect()
    transaction = connection.begin()  # 🔥 1️⃣ 트랜잭션 시작

    db = TestingSessionLocal(bind=connection)
    yield db  # 🔥 2️⃣ 테스트 실행 (commit 가능)

    db.rollback()  # 🔥 3️⃣ 테스트 종료 후 rollback (변경 사항 무효화)
    connection.close()

 

tests/test_routers/test_items.py

import pytest
from fastapi.testclient import TestClient
from app.main import app
from app.models.item import Item

client = TestClient(app)

def test_read_existing_item(db, test_token):
    """🔥 기존 데이터가 있는 경우 정상 조회되는지 테스트"""
    existing_item = db.query(Item).filter(Item.id == 1).first()
    assert existing_item is not None, "테스트를 위해 기존 데이터가 필요합니다."

    headers = {"Authorization": f"Bearer {test_token}"}
    response = client.get("/items/", headers=headers)

    assert response.status_code == 200
    assert response.json()["message"] == "You are authenticated!"

def test_create_new_item(db, test_token):
    """🔥 `POST /items/` 엔드포인트 테스트 (새로운 아이템 추가)"""
    headers = {"Authorization": f"Bearer {test_token}"}
    payload = {"name": "Test Item"}

    response = client.post("/items/", json=payload, headers=headers)

    assert response.status_code == 200  # 201로 설정할 수도 있음
    data = response.json()
    assert "id" in data
    assert data["name"] == "Test Item"

    # ✅ DB에서 실제로 존재하는지 확인
    item_from_db = db.query(Item).filter(Item.id == data["id"]).first()
    assert item_from_db is not None
    assert item_from_db.name == "Test Item"

    # 🔥 하지만 테스트 종료 후 rollback()이 실행되므로 실제 DB에는 데이터가 남지 않음!