FastAPI+pytest测试多函数时触发“事件循环已关闭”错误排查

错误日志 #

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
==================================================================== short test summary info =====================================================================
FAILED app/test/test_subscriptions.py::test_post_subscriptions - RuntimeError: Event loop is closed
============================================================ 1 failed, 1 passed, 16 warnings in 7.45s ============================================================

相关代码 #

测试用例的SetUp和TearDown代码 #

class TestingSession(Session):
    def commit(self):
        self.flush()
        self.expire_all()


@pytest.fixture(scope="function")
def test_db():
    # Set Up
    TestSessionLocal = sessionmaker(
        class_=TestingSession,
        autocommit=False,
        autoflush=False,
        bind=engine
    )
    db = TestSessionLocal()

    def get_db_for_testing():
        try:
            yield db
            db.commit()
        except SQLAlchemyError as e:
            assert e is not None
            db.rollback()

    app.dependency_overrides[get_db] = get_db_for_testing

    # Test
    yield db

    # Tear Down
    db.rollback()
    close_all_sessions()
    engine.dispose()

测试函数 #

def test_read_subscriptions(test_db):
    test_team = test_db.query(Team).filter(
        Team.email == "unit_test_get@example.com"
    ).first()
    response = client.get(f'/api/v1/teams/{test_team.id}/subscriptions')
    assert response.status_code == 200

# intentionally duplicated the same function
def test_read_subscriptions_2(test_db):
    test_team = test_db.query(Team).filter(
        Team.email == "unit_test_get@example.com"
    ).first()
    response = client.get(f'/api/v1/teams/{test_team.id}/subscriptions')
    assert response.status_code == 200

main.py中的异步日志中间件代码 #

注释该中间件后测试可正常执行：

@app.middleware("http")
async def log_middleware(request: Request, call_next):
    # create request ID
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id

    logger = logging.getLogger("my_logger")

    is_not_develop_path = request.url.path not in ["/docs", "/openapi.json"]

    if is_not_develop_path:
        # request log
        logger.info(f"Request ID={request_id}, Method={request.method}, URL={request.url}")

    # calcurate processing time
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time

    if is_not_develop_path:
        # response log
        if 400 <= response.status_code < 600:
            # Log error if response status is 4xx or 5xx
            logger.error(f"Response ID={request_id}, Status={response.status_code}, ProcessTime={process_time}s")
        else:
            # Otherwise log info
            logger.info(f"Response ID={request_id}, Status={response.status_code}, ProcessTime={process_time}s")

    if is_not_develop_path:
        # write log to MongoDB
        await userLogsCrud.create_log(request=request)
    return response

原因分析 #

问题核心在于异步日志中间件与测试事件循环的冲突：

1. 第一个测试结束后，test_db fixture的teardown阶段执行engine.dispose()，同时pytest默认的事件循环会被重置或关闭。
2. 异步日志中间件中的await userLogsCrud.create_log(request=request)属于异步操作，第二个测试执行时，中间件尝试调用已关闭的事件循环执行MongoDB写入，直接触发RuntimeError: Event loop is closed。

解决步骤 #

1. 统一管理测试的异步事件循环 #

在测试文件顶部添加事件循环fixture，确保每个测试都使用独立且完整的事件循环：

import asyncio
import pytest

@pytest.fixture(scope="function")
async def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

2. 调整test_db fixture的清理逻辑 #

避免在单个测试后销毁整个数据库引擎，改为仅关闭当前会话，把引擎销毁移到所有测试结束后执行：

@pytest.fixture(scope="module")
def test_db():
    # 原有setup代码不变
    TestSessionLocal = sessionmaker(
        class_=TestingSession,
        autocommit=False,
        autoflush=False,
        bind=engine
    )
    db = TestSessionLocal()

    def get_db_for_testing():
        try:
            yield db
            db.commit()
        except SQLAlchemyError as e:
            assert e is not None
            db.rollback()

    app.dependency_overrides[get_db] = get_db_for_testing

    yield db

    # Tear Down：仅清理当前会话，引擎在模块测试结束后销毁
    db.rollback()
    db.close()
    close_all_sessions()
    engine.dispose()

3. 测试环境禁用异步日志写入 #

如果测试不需要实际写入MongoDB日志，可以在测试文件中临时移除该中间件：

# 在测试文件开头添加
app.middleware_stack = app.build_middleware_stack()
# 或者直接覆盖中间件
def dummy_middleware(request: Request, call_next):
    return call_next(request)
app.middleware("http")(dummy_middleware)

内容的提问来源于stack exchange，提问作者Dai