使用unittest IsolatedAsyncioTestCase测试FastAPI路由时遇Event loop is closed错误

问题根源 #

• IsolatedAsyncioTestCase会为每个测试方法创建独立事件循环，测试结束后销毁该循环。
• 若全局初始化FastAPI应用或MongoDB异步客户端，这些实例会绑定到第一个测试的事件循环，后续测试调用时原循环已关闭，触发RuntimeError: Event loop is closed。
• TestClient的同步调用逻辑在异步测试环境中，可能错误复用已关闭的旧循环。

修复步骤 #

1. 避免全局初始化：不在测试类外创建FastAPI实例和MongoDB客户端，改为在每个测试的asyncSetUp中重新初始化。
2. 绑定当前循环：确保每个测试的应用、数据库连接都绑定到当前测试的事件循环。
3. 清理测试数据：在asyncTearDown中删除测试产生的数据，保证测试独立性。

修改后的完整测试代码 #

import unittest
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from motor.motor_asyncio import AsyncIOMotorClient
from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    description: str | None = None

class Item(ItemCreate):
    id: str

class TestItemAPI(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        # 为当前测试创建独立的FastAPI实例
        self.app = FastAPI()

        # 注册路由，每个路由内独立管理MongoDB客户端生命周期
        @self.app.post("/items/", response_model=Item)
        async def create_item(item: ItemCreate):
            client = AsyncIOMotorClient("mongodb://localhost:27017")
            db = client.test_db
            item_dict = item.dict()
            result = await db.items.insert_one(item_dict)
            client.close()
            return Item(id=str(result.inserted_id), **item_dict)

        @self.app.get("/items/{item_id}", response_model=Item)
        async def get_item(item_id: str):
            client = AsyncIOMotorClient("mongodb://localhost:27017")
            db = client.test_db
            item = await db.items.find_one({"_id": item_id})
            client.close()
            if not item:
                raise HTTPException(status_code=404, detail="Item not found")
            return Item(id=str(item["_id"]), name=item["name"], description=item.get("description"))
        
        # 创建TestClient绑定当前测试的app
        self.client = TestClient(self.app)

    async def asyncTearDown(self):
        # 清理测试数据
        client = AsyncIOMotorClient("mongodb://localhost:27017")
        await client.test_db.items.delete_many({})
        client.close()

    async def test_create_and_get_item(self):
        # 创建测试item
        create_resp = self.client.post("/items/", json={"name": "Test Item", "description": "Test Desc"})
        self.assertEqual(create_resp.status_code, 200)
        created_item = create_resp.json()
        item_id = created_item["id"]

        # 获取item，此时不会触发循环关闭错误
        get_resp = self.client.get(f"/items/{item_id}")
        self.assertEqual(get_resp.status_code, 200)
        fetched_item = get_resp.json()
        self.assertEqual(fetched_item["name"], "Test Item")
        self.assertEqual(fetched_item["description"], "Test Desc")

if __name__ == "__main__":
    unittest.main()

关键说明 #

• 每个测试方法都会通过asyncSetUp创建全新的FastAPI应用和MongoDB连接，完全绑定到当前测试的事件循环，避免跨测试的循环复用。
• 路由内每次请求创建并关闭MongoDB客户端，确保连接不会残留到已关闭的循环中。
• asyncTearDown清理测试数据，防止不同测试间的数据污染。

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