解决FastAPI异步测试中“Event loop is closed”错误

# app.py
from typing import Optional

import motor.motor_asyncio
import uvicorn
from bson import ObjectId
from fastapi import APIRouter, Body, FastAPI, HTTPException, Request, status
from pydantic import BaseModel, ConfigDict, EmailStr, Field
from pydantic.functional_validators import BeforeValidator
from typing_extensions import Annotated

# -------- Model --------
# 定义PyObjectId类型，用于处理MongoDB的ObjectId
PyObjectId = Annotated[str, BeforeValidator(str)]

class ItemModel(BaseModel):
    id: Optional[PyObjectId] = Field(alias="_id", default=None)
    name: str = Field(...)
    email: EmailStr = Field(...)
    model_config = ConfigDict(
        populate_by_name=True,
        arbitrary_types_allowed=True,
        json_schema_extra={
            "example": {"name": "Jane Doe", "email": "jane.doe@example.com"}
        },
    )

# -------- Router --------
mcve_router = APIRouter()

@mcve_router.post(
    "",
    response_description="Add new item",
    response_model=ItemModel,
    status_code=status.HTTP_201_CREATED,
    response_model_by_alias=False,
)
async def create_item(request: Request, item: ItemModel = Body(...)):
    db_collection = request.app.db_collection
    new_item = await db_collection.insert_one(
        item.model_dump(by_alias=True, exclude=["id"])
    )
    created_item = await db_collection.find_one({"_id": new_item.inserted_id})
    return created_item

@mcve_router.get(
    "/{id}",
    response_description="Get a single item",
    response_model=ItemModel,
    response_model_by_alias=False,
)
async def show_item(request: Request, id: str):
    db_collection = request.app.db_collection
    if (item := await db_collection.find_one({"_id": ObjectId(id)})) is not None:
        return item

    raise HTTPException(status_code=404, detail=f"Item {id} not found")

# FastAPI 应用实例
app = FastAPI()
app.include_router(mcve_router, tags=["item"], prefix="/item")

# 数据库客户端和集合配置
app.db_client = motor.motor_asyncio.AsyncIOMotorClient(
    "mongodb://127.0.0.1:27017/?readPreference=primary&appname=MongoDB%20Compass&ssl=false"
)
app.db = app.db_client.mcve_db
app.db_collection = app.db.get_collection("bars")

# 应用启动入口 (用于开发和生产环境)
if __name__ == '__main__':
    uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)

# test_app.py
import asyncio
import unittest
from fastapi.testclient import TestClient
from app import app # 从 app.py 导入 FastAPI 应用实例

class TestAsync(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self):
        """
        在每个测试方法运行前异步设置测试环境。
        这里初始化 FastAPI TestClient。
        """
        self.client = TestClient(app)

    async def asyncTearDown(self):
        """
        在每个测试方法运行后异步清理测试环境。
        这里关闭 MongoDB 客户端连接。
        """
        self.client.app.db_client.close()
        # 清理数据库（可选，但推荐在实际项目中进行）
        # await self.client.app.db_collection.delete_many({})

    async def run_async_test(self, coro):
        """
        辅助方法，用于在 IsolatedAsyncioTestCase 的上下文中运行一个协程。
        asyncio.run() 会为每次调用创建一个新的事件循环，确保隔离性。
        """
        return asyncio.run(coro)

    async def test_show_item(self):
        """
        测试创建和获取 Item 的异步流程。
        整个测试逻辑被封装在一个异步函数中，并通过 run_async_test 执行。
        """
        async def test_logic():
            bar_data = {"name": "John Doe", "email": "john.doe@example.com"}

            # 异步发送 POST 请求创建 Item
            create_response = await self.client.post("/item", json=bar_data)
            self.assertEqual(create_response.status_code, 201)

            created_item_id = create_response.json().get("id")
            self.assertIsNotNone(created_item_id)

            # 异步发送 GET 请求获取已创建的 Item
            response = await self.client.get(f"/item/{created_item_id}")
            self.assertEqual(response.status_code, 200)
            self.assertEqual(response.json().get("name"), "John Doe")
            self.assertEqual(response.json().get("email"), "john.doe@example.com")
            self.assertEqual(response.json().get("id"), created_item_id)

        # 运行异步测试逻辑
        await self.run_async_test(test_logic())

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

fastapi
httpx
motor
pydantic[email]
python-bsonjs
uvicorn==0.24.0

pip install -r requirements.txt

python -m unittest test_app.py