이벤트 테스트하기: lifespan과 startup - shutdown

이벤트 테스트하기: lifespan과 startup - shutdown (Testing Events)

앱이 뜰 때와 꺼질 때 실행되는 lifespan 로직을 테스트에서도 돌리고 싶다면, TestClientwith과 함께 사용하면 돼요. 이번 장에서는 그 방법을 배워볼게요.

출처: 공식문서

lifespan 테스트하기

TestClientwith 블록 안에서 쓰면, 그 블록 안에서 lifespan이 시작돼요. 블록이 끝나면 앱이 종료된 것처럼 lifespan도 끝나요.

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.testclient import TestClient

items = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}
    yield
    # clean up items
    items.clear()

app = FastAPI(lifespan=lifespan)

@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]

def test_read_items():
    # Before the lifespan starts, "items" is still empty
    assert items == {}

    with TestClient(app) as client:
        # Inside the "with TestClient" block, the lifespan starts and items added
        assert items == {"foo": {"name": "Fighters"}, "bar": {"name": "Tenders"}}

        response = client.get("/items/foo")
        assert response.status_code == 200
        assert response.json() == {"name": "Fighters"}

        # After the requests is done, the items are still there
        assert items == {"foo": {"name": "Fighters"}, "bar": {"name": "Tenders"}}

    # The end of the "with TestClient" block simulates terminating the app, so
    # the lifespan ends and items are cleaned up
    assert items == {}

테스트 안에서 TestClient(app) 블록에 진입하면 시작 부분의 yield 앞 코드가 실행되면서 items가 채워져요. client.get("/items/foo")로 요청을 보내고, 블록이 끝나면 yield 뒤의 정리 코드(items.clear())가 실행되면서 items가 다시 비워져요. 이 흐름을 테스트의 assert로 하나씩 확인하고 있어요.

테스트에서 lifespan을 실행하는 방법에 대한 더 자세한 내용은 공식 Starlette 문서에서 읽을 수 있어요.

startupshutdown 이벤트 테스트하기

이제 더 이상 권장되지 않는(deprecated) startupshutdown 이벤트를 테스트하는 방법도 알아둘게요. TestClient를 이렇게 쓰면 돼요:

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

items = {}

@app.on_event("startup")
async def startup_event():
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}

@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]

def test_read_items():
    with TestClient(app) as client:
        response = client.get("/items/foo")
        assert response.status_code == 200
        assert response.json() == {"name": "Fighters"}

with TestClient(app) as client: 블록 안에서는 startup 이벤트가 실행돼서 items가 채워진 상태예요. 그래서 client.get("/items/foo")가 정상 응답을 반환해요.

참고lifespan(권장) 방식을 쓴다면 앞선 코드처럼 @asynccontextmanager 기반의 lifespan을 사용하는 게 좋아요. @app.on_event("startup")/shutdown은 옛 방식이라 새 코드에서는 피하는 게 좋아요.

더 알아보기 (Learn more)