[FastAPI/Python] API를 비동기적으로 테스트하기(with pytest, HTTPX)

Chapter 9: API를 비동기적으로 테스트하기(with pytest, HTTPX)

실전 프로젝트를 시작하기 전 마지막 단원으로 테스트에 대해서 배워보자. 유닛 테스트 작성은 고품질의 소프트웨어 제작에 필수적인 요소이다. 파이썬에서는 기본적으로 unittest라는 모듈을 제공한다. 하지만 많은 파이썬 개발자들은 더욱 가벼운 문법을 제공하기도 하고 advanced use case를 위한 강력한 툴들을 제공하기 때문에 pytest를 더 선호한다.

pytest를 이용한 unit test

pytest logo

unittest를 이용하는 경우

import unittest

from chapter9.chapter9_introduction import add

class TestChapter9Introduction(unittest.TestCase):
    def test_add(self):
        self.assertEqual(add(2, 3), 5)

unittest는 TestCase를 상속받아서 클래스를 구현하고 그 내부에 테스트할 함수를 각각 구현해야 한다. 그리고 assertEqual와 같은 메서드를 반드시 이용해야한다.

pytest를 사용하는 경우 더욱 간편하게 테스트를 만들 수 있다.

from chapter9.chapter9_introduction import add

def test_add():
    assert add(2, 3) == 5

• 클래스를 구현할 필요가 없다. → 단순한 함수만 만들면 끝
• 함수를 만들 때는 함수명에 test_ 접두사만 붙이면 된다.
• python built-in aseert statement를 사용할 수 있다(여기서는 ==).

이 책에서는 더욱 생산성이 높다고 판단되는 pytest에 집중하여 내용이 진행된다.

Generating tests with parametrize

앞선 예시에서는 2와 3을 더하는 덧셈 한 번만 테스트를 만들었다.

그러나 여러 케이스를 테스트하려면 어떤 방법을 택해야 할까?

def test_add():
    assert add(2, 3) == 5
    assert add(0, 0) == 0
    assert add(100, 0) == 100
    assert add(1, 1) == 2

이렇게 여러 번 반복하도록 구현해도 동작은 할 것이다. 하지만 여기에는 여러 부담이 따른다.

• 비슷한 코드의 반복이 발생
  • 테스트 케이스가 복잡한 경우, 실제 함수의 구현이 조금만 바뀌어도 수정해주어야 할 부분이 많다
• 하나의 덧셈만 실패해도 나머지 테스트들이 마저 진행되지 못한다. 따라서 우리는 코드 수정과 테스트 코드 수행을 여러 번 반복해야 할 수도 있다.

이러한 단점을 보완하기 위해서 pytest는 parametrize 마커를 제공한다. marker란 테스트에 메타데이터를 편하게 넘겨줄 수 있도록 하는 데코레이터이다. 이제 예시로 사용방법을 살펴보자

import pytest
from chapter9.chapter9_introduction import add

@pytest.mark.parametrize("a,b,result", [(2, 3, 5), (0, 0, 0), (100, 0, 100), (1, 1, 2)])
def test_add(a, b, result):
    assert add(a, b) == result

테스트 함수(test_...) 위에 parametrize 마커로 decorate 한다. 그리고 데코레이터에 인자를 넘겨주는데, 첫 번째 인자로는 각 parameter의 이름을 콤마로 구분하여 하나의 string으로 만들어서 넘겨준다. 그리고 두 번째 인자로는 각 parameter가 받아오는 인자 값을 tuple로 만들고 리스트에 담아서 보낸다.

이렇게 하면, assert 문을 한 번만 쓰고도 여러 개의 테스트 케이스를 작성할 수 있다는 장점이 있다(코드 중복 및 하드 코딩 방지).

fixture를 생성해서 테스트 로직 재사용하기

커다란 어플리케이션을 테스트할 때, 처음에 앱을 초기화하는 등의 과정을 진행하는 보일러플레이트 코드가 필요하다.

from datetime import date
from enum import Enum
from typing import List

from pydantic import BaseModel

class Gender(str, Enum):
    MALE = "MALE"
    FEMALE = "FEMALE"
    NON_BINARY = "NON_BINARY"

class Address(BaseModel):
    street_address: str
    postal_code: str
    city: str
    country: str

class Person(BaseModel):
    first_name: str
    last_name: str
    gender: Gender
    birthdate: date
    interests: List[str]
    address: Address

일단 예제를 위한 Pydantic 모델부터 정의해본다. 이러한 class를 매번 테스트마다 인스턴스화하는 것이 반복적이고 짜증 나는 과정일 것이다.

이러한 귀찮은 과정을 fixture를 이용해서 해결할 수 있다.

import pytest

from chapter9.chapter9_introduction_fixtures import Address, Gender, Person

@pytest.fixture
def address():
    return Address(
        street_address="12 Squirell Street",
        postal_code="424242",
        city="Woodtown",
        country="US",
    )

@pytest.fixture
def person(address):
    return Person(
        first_name="John",
        last_name="Doe",
        gender=Gender.MALE,
        birthdate="1991-01-01",
        interests=["travel", "sports"],
        address=address,
    )

def test_address_country(address):
    assert address.country == "US"

def test_person_first_name(person):
    assert person.first_name == "John"

def test_person_address_city(person):
    assert person.address.city == "Woodtown"

fixture는 함수 위에 붙여지는 데코레이터이고, 그 아래에 테스트에 사용할 (가짜) 데이터를 생성해서 반환하는 함수를 작성한다.

아래에 있는 테스트 함수를 확인해보면, 인자로 fixture를 넘겨받고, dot 접근자로 Model 인스턴스의 데이터에 접근하여 테스트를 진행한다.

fixture는 또 다른 fixture를 참조하여 사용할 수 있다는 것이 테스트를 더욱 간편하게 만든다는 이점이 있다. dependency가 다른 dependency를 주입받을 수 있는 것과 유사하다.

FastAPI 테스팅을 위한 환경 준비 with HTTPX

FastAPI 공식 사이트에서는 Starlette에서 제공하는 TestClient를 사용하길 권장한다. 하지만, 이 책에서는 HTTP client를 제공하는 HTTPX를 이용할 것이다.

그 이유는 TestClient가 완전히 synchronous 하게 구현되었기 때문이다. 따라서 테스트 코드를 작성할 때, async나 await을 고려하지 않아도 된다는 게 장점으로 들릴 수도 있다. 그러나 실제 작성 시에 우리가 asynchronous 하게 동작하는 로직들이 있기 때문에, 머릿속이 더 복잡해지고, 디버깅하기 힘든 버그가 발생한다.

HTTPX는 Starlette를 만든 팀에서 만든 HTTP client이고, pure asynchronous HTTP 클라이언트를 제공한다. 테스팅을 위해서는 총 3개의 라이브러리가 추가로 필요하다.

• HTTPX: HTTP 요청을 수행하는 클라이언트
• asgi-lifespan: FastAPI 앱을 켜고, 끌 수 있게 해주는 라이브러리
• pytest-asyncio: 비동기적 테스트 코드 작성을 위해 필요한 pytest extension

테스트가 필요한 코드를 먼저 작성해보자.

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def hello_world():
    return {"hello": "world"}

@app.on_event("startup")
async def startup():
    print("Startup")

@app.on_event("shutdown")
async def shutdown():
    print("Shutdown")

먼저 이벤트루프와 관련된 fixture를 만들어서 테스트를 진행할 때, 같은 event loop 인스턴스 내에 있다는 것을 보장할 수 있게 만든다.

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()

이 과정에서는 현재의 evnet_loop을 가져오고, yield한 다음 loop를 닫는(clean-up) 과정을 통해서 함수 다시 호출되기 전까지 하나의 하나의 loop만을 사용하도록 만든다. 이는 테스트를 준비하는 아주 일반적인 패턴이라 할 수 있다.

이제, test_client fixture를 구현해보자, HTTPX 클라이언트 인스턴스를 만드는 함수이다. 앱의 이벤트는 반드시 asgi-lifespan으로 트리거해야 한다.

from asgi_lifespan import LifespanManager

@pytest_asyncio.fixture
async def test_client():
    async with LifespanManager(app):
        async with httpx.AsyncClient(app=app, base_url="http://app.io") as test_client:
            yield test_client

예시 코드에서 app은 테스트를 진행할 코드에서 생성된 FastAPI app 객체이다. 따라서 해당 패키지에서 import 해와야 한다.

그리고 with이라는 syntax가 사용되었는데, 파이썬에서는 이것을 context manager라고 부른다. setup 로직이 필요한 경우(object.__enter__(self)에 바인딩), teardown 로직이 필요한 경우(object.__exit__(self)에 바인딩) 사용한다.

여기서는 with를 중첩하여 사용하였고, 처음 바깥쪽에 있는 것은 shutdown, teardown event가 수행되도록 보장하는 것이고, 안 쪽의 with는 http 세션이 준비되도록 만든다.

여기서 또 generator를 사용한다. 이 부분은 중요한데, 만약 코드가 뒤에 더 이상 진행할 것이 없으면, 우리는 context manager가 exit 할 수 있게 해야 한다. 따라서 yield문 다음에 우리는 with 블록을 빠져나온다.

큰 프로젝트에서는 여러 테스트 파일이 생기고, 이 테스트 파일들을 정리할 필요가 있다. 보통 이런 테스트 코드는 root 경로에tests 폴더를 생성하여 저장한다. 그리고 여러 테스트 파일들에서 공통적으로 사용할 코드를 global fixture로 선언할 수 있다. 이 fixture는 tests 폴더 내의 conftest.py에 작성해야 한다. 그렇다면 자동으로 import 되어 여러 테스트 파일에서 사용할 수 있다(참조 링크)

REST API 엔드포인트 테스트하기

이제 테스팅을 위한 툴들을 모두 공부하였으니, 실제 코드를 작성해보자.

@pytest.mark.asyncio
async def test_hello_world(test_client: httpx.AsyncClient):
    response = await test_client.get("/")
    assert response.status_code == status.HTTP_200_OK
    json = response.json()
    assert json == {"hello": "world"}

먼저 테스트 함수에 접두사 test_를 붙이는 것을 잊지 말자. 그리고 함수 위에 asyncio 마커를 붙여준다. 모든 asynchronous 테스트 코드는 이 마커를 붙여야 정상적으로 수행된다.

그다음으로는 test_client fixture를 파라미터로 선언하자. 주의해야 할 점은 fixture가 제공하는 것의 type hint를 직접 지정해 주어야 한다는 것이다.

함수 내부에서는 클라이언트에 GET 요청을 보낸다. 그리고 response를 받아와서 우리가 기대하는 값과 동일한지 확인한다.

POST 엔드포인트 테스트 작성하기

class Person(BaseModel):
    first_name: str
    last_name: str
    age: int

@app.post("/persons", status_code=status.HTTP_201_CREATED)
async def create_person(person: Person):
    return person

위와 같은 간단한 POST 엔드포인트를 생각해보자. 일단 이 모델에서 모두 필수적으로 값을 요구하는 상태이기 때문에 필드 값이 하나라도 빠진 경우 오류가 생길 것이다.

그래서 이 오류가 정상적으로 발생하는지 확인해보는 테스트 코드를 작성한다. 그리고 정상적으로 POST 되는 경우도 테스트해본다.

@pytest.mark.asyncio
class TestCreatePerson:
    async def test_invalid(self, test_client: httpx.AsyncClient):
        payload = {"first_name": "John", "last_name": "Doe"}
        response = await test_client.post("/persons", json=payload)

        assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY

    async def test_valid(self, test_client: httpx.AsyncClient):
        payload = {"first_name": "John", "last_name": "Doe", "age": 30}
        response = await test_client.post("/persons", json=payload)

        assert response.status_code == status.HTTP_201_CREATED

        json = response.json()
        assert json == payload

test_invalid의 경우, age가 빠졌기 때문에 오류가 발생할 것이다. 그래서 status 코드가 422인지 확인한다. 정상적으로 만들어지는 경우는 생성된 person이 우리가 payload로 지정했던 값과 동일한 지 확인해서 테스트를 진행한다.

여기서는 테스트 코드를 하나의 클래스를 생성하고 그 내부 메서드로 2개의 테스트 함수를 작성했다. 따라서 asyncio 마커를 한 번만 붙일 수 있다. 클래스 내부에 구현된 메서드이므로 함수의 첫 번째 파라미터로 self를 설정하는 것을 잊지 말자. 그리고 두 개의 테스트가 연관된 테스트임을 한눈에 파악할 수 있다.

Testing with a database

앱이 DB와 연결되어 있는 경우가 많을 것이다. 이러한 상황에서는, 새로운 테스트 데이터 베이스와 함께 테스트를 진행해야 한다.

이를 위해서 우리는 두 가지를 준비해야 한다.

• dependency_overrides: 런타임에 어떤 dependency를 교체할 수 있는 FastAPI 기능
  • Spying, stub, mocking 등의 방법과 유사?
• 가짜 데이터를 테스트 DB에 만드는 fixture를 만들어야 한다.

일단, 6장에서 보았던 MongoDB와 관련된 코드를 바탕으로 테스트 코드를 작성해보자

motor_client = AsyncIOMotorClient("mongodb://localhost:27017")
database = motor_client["chapter6_mongo"]

def get_database() -> AsyncIOMotorDatabase:
    return database

여기서 가져오는 DB를 여러 코드에서 사용했었다. 테스트 코드에서는 이 데이터베이스를 테스트용 데이터베이스로 교체해주어야 할 것이다.

테스트를 위해서 새로운 AsyncIOMotorDatabase를 만들어야 한다.

motor_client = AsyncIOMotorClient("mongodb://localhost:27017")
database_test = motor_client["chapter9_db_test"]

def get_test_database():
    return database_test

이렇게 테스트 코드에 구현하고, test_client fixture에서는 get_database 의존성 대신 get_test_database를 사용해야 하도록 오버라이딩 해야한다.

@pytest.fixture
async def test_client():
    app.dependency_overrides[get_database] = get_test_database
    async with LifespanManager(app):
        async with httpx.AsyncClient(app=app, base_url="http://app.io") as test_client:
            yield test_client

FastAPI 앱의 dependency_overrides 프로퍼티를 활용해서 오버라이딩 할 수 있다. 이제 DB에 더미 데이터를 저장하는 fixture를 만든다.

@pytest.fixture(autouse=True, scope="module")
async def initial_posts():
    initial_posts = [
        PostDB(title="Post 1", content="Content 1"),
        PostDB(title="Post 2", content="Content 2"),
        PostDB(title="Post 3", content="Content 3"),
    ]

    await database_test["posts"].insert_many(
        [post.dict(by_alias=True) for post in initial_posts]
    )

    yield initial_posts
    await motor_client.drop_database("chapter9_db_test")

여기서 fixture 데코레이터에 인자로 autouse, scope 인자를 넘겨주었다. 첫 번째는 pytest가 fixture가 다른 test에서 fixture가 호출되지 않더라도, 자동적으로 pytest 스스로 호출하도록 설정하는 것이다. 이 설정 덕분에 항상 DB에 더미 데이터가 생성되어있다고 보장할 수 있다. scope="module" 는 각 테스트 함수마다 fixture가 실행되지 않고, 현재 테스트 모듈(단일 파일)에서 한 번만 수행되도록 만든다. 매번 테스트마다 post가 다시 생성되지 않도록 테스트가 빠르게 진행되도록 한다.

그리고 initial_posts를 yield 하였다. 이 패턴은 테스트가 다 수행된 다음 DB에서 table을 지워서 새로 시작하는 테스트에서는 새로운 데이터베이스를 이용할 수 있게 한다.

이제, GET 엔드포인트에 대한 테스트를 작성하자.

@pytest.mark.asyncio
class TestGetPost:
    async def test_not_existing(self, test_client: httpx.AsyncClient):
        response = await test_client.get("/posts/abc")

        assert response.status_code == status.HTTP_404_NOT_FOUND

    async def test_existing(
        self, test_client: httpx.AsyncClient, initial_posts: List[PostDB]
    ):
        response = await test_client.get(f"/posts/{initial_posts[0].id}")

        assert response.status_code == status.HTTP_200_OK

        json = response.json()
        assert json["_id"] == str(initial_posts[0].id)

테스트 방법은 앞서 살펴본 테스트 코드와 유사하다.

POST 엔드포인트 테스트 코드는 아래와 같다.

@pytest.mark.asyncio
class TestCreatePost:
    async def test_invalid_payload(self, test_client: httpx.AsyncClient):
        payload = {"title": "New post"}
        response = await test_client.post("/posts", json=payload)

        assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY

    async def test_valid_payload(self, test_client: httpx.AsyncClient):
        payload = {"title": "New post", "content": "New post content"}
        response = await test_client.post("/posts", json=payload)

        assert response.status_code == status.HTTP_201_CREATED

        json = response.json()
        post_id = ObjectId(json["_id"])
        post_db = await database_test["posts"].find_one({"_id": post_id})
        assert post_db is not None

여기서는 database_test 객체로 직접 DB에 접근해서 DB에 insert하는 과정이 정상적으로 이뤄졌는지 확인할 수 있다. 이는 asynchronous 테스트를 이용하는 것의 장점이다. 덕분에, 실제 코드에서 쓰던 것처럼 DB에 접근할 수 있고, 같은 라이브러리와 툴을 쓸 수 있다. dependency_ovrrides을 알아야 하는 이유다.

이 기능은 외부에서 정의된 서비스를 포함한 코드를 테스트할 때도 유용하다.

다음과 같은 external API가 있다고 생각하자.

from typing import Any, Dict

import httpx
from fastapi import FastAPI, Depends

app = FastAPI()

class ExternalAPI:
    def __init__(self) -> None:
        self.client = httpx.AsyncClient(
            base_url="https://dummy.restapiexample.com/api/v1/"
        )

    async def __call__(self) -> Dict[str, Any]:
        async with self.client as client:
            response = await client.get("employees")
            return response.json()

external_api = ExternalAPI()

@app.get("/employees")
async def external_employees(employees: Dict[str, Any] = Depends(external_api)):
    return employees

실제 외부 API의 경우 시간, 비용, rate limiting 등에 의해 제약을 받기 때문에 모든 behavior를 재연하기 힘들다.

dependency_overrides를 이용하면, 손쉽게 ExternalAPI를 정적인 data를 반환하는 dependency class로 바꿀 수 있다. 이는 테스트 구현을 쉽게 만들어 준다.

class MockExternalAPI:
    mock_data = {
        "data": [
            {
                "employee_age": 61,
                "employee_name": "Tiger Nixon",
                "employee_salary": 320800,
                "id": 1,
                "profile_image": "",
            }
        ],
        "status": "success",
        "message": "Success",
    }

    async def __call__(self) -> Dict[str, Any]:
        return MockExternalAPI.mock_data

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()

@pytest_asyncio.fixture
async def test_client():
    app.dependency_overrides[external_api] = MockExternalAPI()
    async with LifespanManager(app):
        async with httpx.AsyncClient(app=app, base_url="http://app.io") as test_client:
            yield test_client

@pytest_asyncio.fixture
async def test_get_employees(test_client: httpx.AsyncClient):
    response = await test_client.get("/employees")

    assert response.status_code == status.HTTP_200_OK

    json = response.json()
    assert json == MockExternalAPI.mock_data

앞선 방식과 같이 event_loop와 test_client fixture를 만들어주고 dependency_overrides를 이용해서 의존성을 교체해준다!

그 후 우리가 예상하는 상황에 맞는 테스트를 작성하면 된다.

웹소켓 엔드포인트 테스트 작성하기

웹소켓을 테스트할 때는 아쉽게도 HTTPX를 사용할 수 없다.

이 경우 어쩔 수 없이, FastAPI에서 권장하던 FastAPI의 TestClient를(Starlette 기반) 사용해야 한다.

from fastapi import FastAPI, WebSocket
from starlette.websockets import WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")
    except WebSocketDisconnect:
        await websocket.close()

다음과 같은 간단한 웹소켓 엔드포인트를 생각해보자. 일단 앞선 테스트 구현과 유사하게, fixture부터 차례로 구현해야 한다.

import asyncio

import pytest
from fastapi.testclient import TestClient

from chapter9.chapter9_websocket import app

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()

@pytest.fixture
def websocket_client():
    with TestClient(app) as websocket_client:
        yield websocket_client

여기서 웹소켓 클라이언트를 만들 때, 마찬가지로 with 구문으로 context manager를 이용한다. 여기서는 TestClient가 context manager로 동작하고, 인자로 FastAPI app을 기대한다. 그리고 마찬가지로 yield 를 이용해서 exit logic이 테스트가 끝난 후 진행될 수 있게 하자. 여기서는 TestClient가 자동적으로 lifespan event를 트리거하기 때문에 신경 쓰지 않아도 된다.

이제 실제 테스트 함수를 작성해보자.

@pytest.mark.asyncio
async def test_websocket_echo(websocket_client: TestClient):
    with websocket_client.websocket_connect("/ws") as websocket:
        websocket.send_text("Hello")

        message = websocket.receive_text()
        assert message == "Message text was: Hello"

asyncio 마커를 붙이는 것도 동일하다(비록 TestClient가 동기적으로 동작하지만..). 이것은 함수 내에서 비동기적인 서비스를 부르거나, event loop으로 인해서 발생하는 이슈를 줄일 수 있다.

그다음 websocket_client로부터 웹소켓 엔드포인트에 연결한다. 그리고 이 연결은 context manager로 동작하게 된다. 그리고 websocket 변수를 우리에게 제공한다. send_text 나 receive_text 같은 메서드는 완료될 때까지 blocking 된다는 점을 주의한다.

이 이유 때문에 웹소켓 엔드포인트를 테스트할 때는 보내는 순서와 받는 순서를 고려해서 코드를 작성해야 한다.

그 외에 다른 테스트 툴과 방법은 동일하게 적용할 수 있다. 특히, 데이터베이스를 활용해서 테스트할 때, dependency_overrides를 사용해야 한다.

요약

이제 테스트 코드 작성을 통해 고품질의 FastAPI 앱을 만들 수 있게 되었다.

• pytest
  • fixture: 재사용 가능한 초기화 로직
• HTTPX
  • 테스트 클라이언트 for async
• TestClient for WebSocket
• 의존성 오버라이딩
  • for DB instance

등을 배웠다. 다음 단원에서는 배포(deploy) 방법에 대해 알아본다.