如何在Pytest异步Fixture销毁后无事件循环错误删除MongoDB测试库

为FastAPI应用配置基于Pytest的异步测试Fixture，采用MongoDB（motor驱动）与ODMantic（AIOEngine），在Fixture销毁阶段尝试删除测试数据库时，触发以下错误：

Error dropping database: This event loop is already closed

原测试代码如下：

import asyncio
import pytest
import pytest_asyncio
from motor.motor_asyncio import AsyncIOMotorClient
from odmantic import AIOEngine
import os
from typing import AsyncGenerator
from app.main import app
from app.db.database import get_database
from fastapi.testclient import TestClient

# Test database configuration
TEST_DATABASE_URL = os.getenv("TEST_DATABASE_URL", "mongodb://localhost:27017")
TEST_DATABASE_NAME = "test_db"

@pytest_asyncio.fixture(scope="session")
async def event_loop():
    """Create an instance of the default event loop for the test session."""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop
    loop.close()

@pytest_asyncio.fixture(scope="function")
async def test_engine() -> AsyncGenerator[AIOEngine, None]:
    """Create a test database engine and clean up after each test."""
    client = AsyncIOMotorClient(TEST_DATABASE_URL)
    engine = AIOEngine(client=client, database=TEST_DATABASE_NAME)

    yield engine

    try:
        print(f"🗑️ Dropping database: {TEST_DATABASE_NAME}")
        await client.drop_database(TEST_DATABASE_NAME)  # <-- Fails here if event loop is closed
    except Exception as e:
        print(f"Error dropping database: {e}")
    finally:
        client.close()

@pytest_asyncio.fixture
async def client(test_engine):
    """Create a test client with dependency override."""
    app.dependency_overrides.clear()
    app.dependency_overrides[get_database] = lambda: test_engine

    with TestClient(app) as test_client:
        yield test_client

    app.dependency_overrides.clear()

@pytest.fixture
def auth_headers(client):
    """Fixture to get authenticated headers for testing protected routes."""
    # Register test user
    client.post("/api/v1/users", json={"email": "test@gmail.com", "password": "test123", "is_superuser": True})
    # Login and return token
    login_response = client.post("/api/v1/login", json={"email": "test@gmail.com", "password": "test123"})
    if login_response.status_code == 200:
        return {"Authorization": f"Bearer {login_response.json()['access_token']}"}
    return None

解决方案 #

方法1：修正事件循环Fixture的实现 #

原event_loop Fixture手动设置事件循环可能导致循环被提前绑定和关闭，改用事件循环策略创建循环，避免手动干预带来的冲突：

@pytest_asyncio.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

方法2：在清理时检测循环状态，临时创建循环执行删除操作 #

当检测到当前事件循环已关闭时，创建临时循环完成数据库删除操作：

@pytest_asyncio.fixture(scope="function")
async def test_engine() -> AsyncGenerator[AIOEngine, None]:
    """Create a test database engine and clean up after each test."""
    client = AsyncIOMotorClient(TEST_DATABASE_URL)
    engine = AIOEngine(client=client, database=TEST_DATABASE_NAME)

    yield engine

    try:
        print(f"🗑️ Dropping database: {TEST_DATABASE_NAME}")
        # 检查当前循环是否可用
        try:
            loop = asyncio.get_running_loop()
            if not loop.is_closed():
                await client.drop_database(TEST_DATABASE_NAME)
            else:
                raise RuntimeError("Loop is closed")
        except (RuntimeError, asyncio.InvalidStateError):
            # 创建临时循环执行删除
            temp_loop = asyncio.new_event_loop()
            temp_loop.run_until_complete(client.drop_database(TEST_DATABASE_NAME))
            temp_loop.close()
    except Exception as e:
        print(f"Error dropping database: {e}")
    finally:
        client.close()

方法3：改用异步测试客户端避免同步操作干扰循环 #

使用httpx.AsyncClient替代同步的TestClient，避免同步操作影响异步事件循环的生命周期：

1. 安装httpx：pip install httpx
2. 修改client和auth_headers Fixture：

import httpx

@pytest_asyncio.fixture
async def client(test_engine):
    """Create an async test client with dependency override."""
    app.dependency_overrides.clear()
    app.dependency_overrides[get_database] = lambda: test_engine

    async with httpx.AsyncClient(app=app, base_url="http://testserver") as test_client:
        yield test_client

    app.dependency_overrides.clear()

@pytest_asyncio.fixture
async def auth_headers(client):
    """Fixture to get authenticated headers for testing protected routes."""
    # Register test user
    await client.post("/api/v1/users", json={"email": "test@gmail.com", "password": "test123", "is_superuser": True})
    # Login and return token
    login_response = await client.post("/api/v1/login", json={"email": "test@gmail.com", "password": "test123"})
    if login_response.status_code == 200:
        return {"Authorization": f"Bearer {login_response.json()['access_token']}"}
    return None

总结 #

• 优先尝试方法1，从事件循环管理层面解决冲突
• 方法2适合快速修复现有代码，无需大规模改动
• 方法3适合全异步测试场景，能从根源避免同步操作对异步循环的干扰

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