FastAPI+Beanie测试Fixture配置问题：异步循环冲突报错

问题背景 #

我开发了一个基于Beanie的FastAPI应用，使用pytest编写测试用例。为了隔离真实数据库，改用AsyncMongoMockClient模拟Beanie的数据库连接后，运行测试时出现错误：
RuntimeError: Task <...> attached to a different loop
测试运行参数为--asyncio-mode=auto。

应用代码 #

def custom_generate_unique_id(route: APIRoute) -> str:
    return f"{route.tags[0]}-{route.name}"


@asynccontextmanager
async def lifespan(app: FastAPI):  # pylint: disable=redefined-outer-name
    client = AsyncIOMotorClient(settings.cosmos_db_url.get_secret_value())
    await init_beanie(
        database=client[settings.database_name], document_models=__beanie_models__
    )
    app.include_router(api_router, prefix="/api")

    yield

    client.close()


app = FastAPI(
    title="My API",
    openapi_url="/api/openapi.json",
    lifespan=lifespan,
    generate_unique_id_function=custom_generate_unique_id,
)

测试Fixture代码（原conftest.py） #

@pytest.fixture(scope='session')
def event_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop
    loop.close()


@pytest_asyncio.fixture(scope='session')
async def mocked_app(event_loop):
    # Use AsyncMongoMockClient instead of real Motor client
    client = AsyncMongoMockClient()
    db = client[settings.database_name]

    # Initialize Beanie with mock DB
    await init_beanie(database=db, document_models=__beanie_models__)

    # Override the startup dependencies if needed
    yield app
    client.close()


@pytest_asyncio.fixture(scope='session')
async def client(event_loop, mocked_app):
    with TestClient(mocked_app) as c:
        yield c


@pytest_asyncio.fixture(scope="function")
async def populate_db(event_loop, client):
    await ChatDocument.delete_all()
    await MessageDocument.delete_all()

    await ChatDocument.insert_many([ChatDocument.from_base(chat) for chat in chat_models])
    await MessageDocument.insert_many([MessageDocument.from_base(message) for message in message_models])

    yield

    await ChatDocument.delete_all()
    await MessageDocument.delete_all()

解决方案 #

1. 创建测试专用FastAPI实例，避免复用全局app #

全局app绑定了原有的lifespan逻辑（使用真实数据库客户端），测试时复用会导致异步循环冲突。需要创建全新的测试应用，手动初始化路由和Beanie：

@pytest_asyncio.fixture(scope='session')
async def mocked_app(event_loop):
    # 创建测试专用FastAPI实例，跳过原lifespan
    test_app = FastAPI(
        title="My API Test",
        openapi_url="/api/openapi.json",
        generate_unique_id_function=custom_generate_unique_id,
    )
    # 手动添加路由
    test_app.include_router(api_router, prefix="/api")
    
    # 初始化Mock数据库
    client = AsyncMongoMockClient()
    db = client[settings.database_name]
    await init_beanie(database=db, document_models=__beanie_models__)
    
    yield test_app
    
    # 清理资源
    await client.close()

2. 改用异步测试客户端替代同步客户端 #

将TestClient替换为httpx.AsyncClient（FastAPI官方推荐的异步测试工具），同时将client fixture的作用域改为function，避免跨测试用例的状态污染：

from httpx import AsyncClient

@pytest_asyncio.fixture(scope="function")
async def client(mocked_app):
    async with AsyncClient(app=mocked_app, base_url="http://testserver") as ac:
        yield ac

3. 简化populate_db fixture，移除冗余循环依赖 #

pytest-asyncio会自动管理异步循环，无需显式传入event_loop参数：

@pytest_asyncio.fixture(scope="function")
async def populate_db():
    await ChatDocument.delete_all()
    await MessageDocument.delete_all()

    await ChatDocument.insert_many([ChatDocument.from_base(chat) for chat in chat_models])
    await MessageDocument.insert_many([MessageDocument.from_base(message) for message in message_models])

    yield

    await ChatDocument.delete_all()
    await MessageDocument.delete_all()

4. 优化全局循环Fixture #

确保测试使用单一的全局异步循环，简化event_loop fixture：

@pytest.fixture(scope='session')
def event_loop():
    policy = asyncio.get_event_loop_policy()
    loop = policy.new_event_loop()
    yield loop
    loop.close()

错误原因解析 #

出现该错误的核心原因是：

• 全局app的lifespan会在启动时创建绑定到特定循环的异步任务
• 测试中用AsyncMongoMockClient初始化的Beanie绑定了session级测试循环
• 原TestClient是同步客户端，执行接口请求时会在另一个循环中运行，导致任务跨循环绑定冲突

内容的提问来源于stack exchange，提问作者André Dias