[udemy] FastAPI - The Complete Course 2025 (Beginner + Advanced) - 학습 정리 3

1. Unit & Integration Testing

   1) pytest를 이용한 fastapi test 방법

      - TestClient()를 사용하면 실재 서버를 띄우지 않고, 테스트 서버를 사용하여 테스트를 진행할 수 있다.

         - TestClient는 내부적으로 httpx를 사용한다. 라이브러리 설치가 요구된다.

pip install httpx

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

client = TestClient(app)


def test_return_health_check():
    response = client.get("/healthy")
    assert response.status_code == status.HTTP_200_OK
    assert response.json() == {'status': 'Healthy'}

 

   2) pyteset의 testing dependencies 설정

      - 테스트를 위해 현재 로그인된 사용자를 mock 한다.

      - utils.py

         - 테스트에 사용될 db 및 override_get_current_user 정의

from sqlalchemy import create_engine, text
from sqlalchemy.pool import StaticPool
from sqlalchemy.orm import sessionmaker
from ..database import Base
from ..main import app
from fastapi.testclient import TestClient
import pytest
from ..models import Todos, Users
from ..routers.auth import bcrypt_context

SQLALCHEMY_DATABASE_URL = "sqlite:///./testdb.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass = StaticPool,
)

TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base.metadata.create_all(bind=engine)

def override_get_db():
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()

def override_get_current_user():
    return {'username': 'codingwithrobytest', 'id': 1, 'user_role': 'admin'}

client = TestClient(app)

@pytest.fixture
def test_todo():
    todo = Todos(
        title="Learn to code!",
        description="Need to learn everyday!",
        priority=5,
        complete=False,
        owner_id=1,
    )

    db = TestingSessionLocal()
    db.add(todo)
    db.commit()
    yield todo
    with engine.connect() as connection:
        connection.execute(text("DELETE FROM todos;"))
        connection.commit()


@pytest.fixture
def test_user():
    user = Users(
        username="codingwithrobytest",
        email="codingwithrobytest@email.com",
        first_name="Eric",
        last_name="Roby",
        hashed_password=bcrypt_context.hash("testpassword"),
        role="admin",
        phone_number="(111)-111-1111"
    )
    db = TestingSessionLocal()
    db.add(user)
    db.commit()
    yield user
    with engine.connect() as connection:
        connection.execute(text("DELETE FROM users;"))
        connection.commit()

 

      - test_*** 다른 파일에 utils 참조

         - 테스트에 사용될 db 및 override_get_current_user 정의

from .utils import *
...

app.dependency_overrides[get_db] = override_get_db
app.dependency_overrides[get_current_user] = override_get_current_user

...

 

      - app.dependency_overrides의 역할

         - 의존성 주입 시스템에 의해 호출되는 함수 또는 객체를 재정의(override)할 수 있는 딕셔너리이다.
            - 키(key): 원래의 의존성 함수
            - 값(value): 대체할 의존성 함수

from fastapi import FastAPI, Depends

app = FastAPI()

# 원래 의존성 함수
def original_dependency():
    return "Original Dependency"

# 라우터에서 의존성 사용
@app.get("/items/")
def read_items(dep: str = Depends(original_dependency)):
    return {"dependency": dep}

# 대체 의존성 함수
def override_dependency():
    return "Overridden Dependency"

# app.dependency_overrides를 사용하여 대체
app.dependency_overrides[original_dependency] = override_dependency

 

      - 일반적인 경우 app 정의와 테스트를 할 경우의 app의 wrap 방법

         - 일반적인 경우 app 정의

app = FastAPI()

 

         - 테스트를 할 경우 app의 wrap 방법

client = TestClient(APP)

 

         - poolclass 정리

            - poolclass는 데이터베이스 연결을 관리하는 데 사용되는 커넥션 풀링(Connection Pooling) 전략을 정의한다.

poolclass	풀링 지원	적합한 환경	특징
QueuePool	예	일반적인 환경	기본 커넥션 풀. 동시성 지원.
SingletonThreadPool	제한적	SQLite, 테스트 환경	스레드별 고유 연결 유지.
StaticPool	아니오	단일 연결, 테스트 환경	동일한 연결을 재사용.
NullPool	아니오	연결 재활용이 필요 없는 경우	매 요청 시 새로운 연결 생성.
AssertionPool	아니오	디버깅 환경	연결 누수 탐지용.

 

   3) pytest.raises()의 사용 방법

      - pytest.raises()는 pytest에서 예외가 발생할 것으로 예상되는 코드를 실행하고, 그 예외를 포착하여 검사하는 데 사용되는 유틸리티이다.

 

      1. 기본 사용법

         - pytest.raises()는 with 구문 내에서 사용하여, 예외가 발생하는지 확인한다. 발생하지 않으면 테스트가 실패한다.

import pytest

def function_that_raises():
    raise ValueError("This is an error!")

def test_function_that_raises():
    with pytest.raises(ValueError) as excinfo:
        function_that_raises()  # 여기서 예외가 발생해야 함
    assert str(excinfo.value) == "This is an error!"

 

      2. 예외가 발생해야 할 때

import pytest

def divide(x, y):
    if y == 0:
        raise ZeroDivisionError("division by zero")
    return x / y

def test_divide_by_zero():
    with pytest.raises(ZeroDivisionError) as excinfo:
        divide(1, 0)
    assert str(excinfo.value) == "division by zero"

 

      3. 예외 메시지 검사

import pytest

def function_that_raises():
    raise ValueError("Invalid input!")

def test_value_error_message():
    with pytest.raises(ValueError) as excinfo:
        function_that_raises()
    assert excinfo.value.args[0] == "Invalid input!"  # 예외 메시지 검증

 

      4. pytest.raises()와 assert 사용

         - pytest.raises()는 예외가 발생하는지 확인하는 것에 그치지 않고, 발생한 예외의 속성도 검증할 수 있다.

         - 예외가 발생한 후 excinfo 객체에서 예외의 속성(status_code, message 등)을 검사할 수 있다.

import pytest
from fastapi import HTTPException

def raise_unauthorized():
    raise HTTPException(status_code=401, detail="Unauthorized access")

def test_http_exception():
    with pytest.raises(HTTPException) as excinfo:
        raise_unauthorized()
    
    assert excinfo.value.status_code == 401
    assert excinfo.value.detail == "Unauthorized access"