Fixing “Event loop is closed” in FastAPI: Why I Use AsyncClient for Testing

FastAPI를 도입해서 테스트 코드를 작성하다 보면 한 번쯤 마주치게 되는 에러가 있다.

RuntimeError: Event loop is closed

이 에러의 정체와 해결법을 제대로 파헤쳐보자.

문제 발생 원인

표면적 원인

• 테스트 함수는 async def로 정의되어 있고, 내부에서 비동기 리소스인 aiopg와 redis를 함께 사용

• 하지만 테스트 클라이언트로는 동기 기반의 TestClient를 사용

심층적 메커니즘

TestClient는 내부적으로 자체 event loop를 생성해서 FastAPI 앱을 동기적으로 실행한다. 테스트가 끝나면 이 루프를 강제로 close() 한다.

# TestClient 내부 동작 (단순화)
def request():
    loop = asyncio.new_event_loop()
    try:
        response = loop.run_until_complete(app(request))
    finally:
        loop.close()  # 여기서 문제 발생

진짜 문제는 가비지 컬렉션 타이밍이다.

1. TestClient가 루프를 닫음

2. aiopg/redis 객체들이 아직 메모리에 살아있음

3. Python GC가 나중에 이 객체들을 정리할 때 __del__ 메서드 호출

4. __del__에서 이미 닫힌 루프에 접근하려 시도

5. RuntimeError: Event loop is closed 발생

TestClient는 비동기 생태계를 완전히 컨트롤하지 못한다. 특히 asyncpg, aioredis, aiopg 등은 루프 종료 시점에 연결 정리 작업이 완료되지 않으면 에러를 유발한다.

해결 방법

1. AsyncClient 사용하기

테스트 함수도 async def, 클라이언트도 httpx.AsyncClient로 통일해야 한다.

import pytest
from httpx import AsyncClient
from myapp import app

@pytest.mark.asyncio
async def test_something():
    async with AsyncClient(app=app, base_url="http://test") as client:
        res = await client.get("/ping")
        assert res.status_code == 200

2. Lifespan 이벤트 처리 (필수)

AsyncClient를 사용하는 것만으로는 FastAPI의 startup, shutdown 이벤트가 자동으로 실행되지 않는다.

비동기 의존성이 @app.on_event("startup"), @app.on_event("shutdown")에 등록되어 있다면, 반드시 asgi_lifespan.LifespanManager로 앱 라이프사이클을 수동으로 실행해야 한다.

from asgi_lifespan import LifespanManager

@pytest.mark.asyncio
async def test_with_lifespan():
    async with LifespanManager(app):
        async with AsyncClient(app=app, base_url="http://test") as client:
            res = await client.get("/api/users")
            assert res.status_code == 200

3. Lifespan 구조 사용하기

⛔️ FastAPI 0.95 이후로 @app.on_event("startup"), @app.on_event("shutdown") 방식은 deprecated 되었다. 대신 lifespan 파라미터 기반의 asynccontextmanager 사용이 권장된다.

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    redis = await connect_to_redis()
    db = await connect_to_db()
    app.state.redis = redis
    app.state.db = db

    yield

    # Shutdown - 순서와 메서드 중요!
    await redis.close()
    await redis.wait_closed()  # aioredis 완전 종료 대기
    db.close()
    await db.wait_closed()    # aiopg 완전 종료 대기

app = FastAPI(lifespan=lifespan)

테스트 최적화 패턴

테스트 픽스처 설계

매 테스트마다 새 클라이언트를 생성하면 성능 문제가 발생한다. 픽스처로 최적화하자.

@pytest_asyncio.fixture(scope="session")
async def app_with_lifespan():
    async with LifespanManager(app) as manager:
        yield manager.app

@pytest_asyncio.fixture(scope="session")
async def async_client(app_with_lifespan):
    async with AsyncClient(app=app_with_lifespan, base_url="http://test") as client:
        yield client

@pytest.mark.asyncio
async def test_endpoint(async_client):
    response = await async_client.get("/api/users")
    assert response.status_code == 200

테스트 격리 전략

옵션 1: 트랜잭션 롤백 (권장)

@pytest_asyncio.fixture
async def db_transaction():
    async with db_pool.acquire() as conn:
        trans = conn.transaction()
        await trans.start()
        try:
            yield conn
        finally:
            await trans.rollback()

옵션 2: 의존성 모킹

from unittest.mock import AsyncMock

@pytest.fixture
def mock_dependencies():
    with patch("myapp.dependencies.get_db", return_value=AsyncMock()):
        with patch("myapp.dependencies.get_redis", return_value=AsyncMock()):
            yield

성능 vs 격리 트레이드오프

결론 및 회고

RuntimeError: Event loop is closed 에러는 단순히 클라이언트만 바꾸면 해결되는 문제가 아니다. 비동기 리소스의 라이프사이클을 제대로 이해하고, 테스트 환경에서 이를 적절히 관리하는 것이 핵심이다.

FastAPI의 최신 lifespan 패턴을 활용하고, 적절한 테스트 격리 전략을 수립하면 안정적이고 빠른 테스트 환경을 구축할 수 있다.