使用Pytest多AsyncClients做集成测试时数据库混淆问题

问题根源 #

你的问题核心在于Tortoise ORM的数据库连接是全局单例状态，一旦被初始化就会在整个测试会话中持续生效：

• session级的initialize_tests先初始化了内存数据库，但real_client里的LifespanManager(app)会触发FastAPI的启动事件，如果你的app在启动时默认初始化了真实数据库，就会直接覆盖之前的内存库连接。
• 测试执行时，先运行的客户端会把全局数据库连接锁定为自己对应的库，后续所有测试都会复用这个连接，导致环境隔离失效。

解决方案：按客户端隔离数据库环境 #

1. 移除全局session级数据库初始化 #

删掉initialize_tests这个autouse fixture，改为在每个客户端fixture中单独初始化对应数据库，确保各自环境完全独立。

2. 修改real_client fixture #

为真实数据库单独初始化连接，测试结束后清理：

@pytest.fixture(scope="function")
async def real_client():
    # 替换为你的真实数据库URL
    REAL_DB_URL = "mysql://user:pass@localhost/real_db"
    # 初始化真实库（假设已有schema，无需重新生成）
    await init_db(REAL_DB_URL, schemas=False)
    
    try:
        async with LifespanManager(app):
            async with AsyncClient(
                transport=ASGITransport(app), base_url="http://localhost"
            ) as c:
                yield c
    finally:
        # 测试结束后清理连接
        await Tortoise._drop_databases()
        await Tortoise.close_connections()

3. 修改test_client fixture #

每次初始化全新的内存数据库，包含测试所需的基础数据：

@pytest.fixture(scope="function")
async def test_client():
    TEST_DB_URL = "sqlite://:memory:"
    # 初始化内存库并生成schema
    await init_db(TEST_DB_URL, create_db=True, schemas=True)
    
    # 初始化测试基础数据
    dashboard_perm = CreatePermission(name="Dashboard", description="Dashboard Page")
    camera_perm = CreatePermission(name="Camera", description="Camera Page")
    await Permission.create(**dashboard_perm.dict())
    await Permission.create(**camera_perm.dict())

    admin_role = CreateRole(name="Admin", description="Admin dashboard")
    admin_user = await User.create(
        first_name="Admin",
        last_name="",
        email=ADMIN_EMAIL,
        password=get_password_hash(password="some@password"),
    )
    role_obj = await Role.create(**admin_role.dict())
    await UserRole.create(role=role_obj, user=admin_user)

    # 给管理员分配所有权限
    all_perms = await Permission.all()
    for perm in all_perms:
        await UserPermission.create(permission=perm, user=admin_user)
    
    try:
        async with AsyncClient(transport=ASGITransport(app), base_url="http://test") as client:
            yield client
    finally:
        # 清理内存库连接
        await Tortoise._drop_databases()
        await Tortoise.close_connections()

4. 动态覆盖App的数据库配置 #

如果你的FastAPI App在启动事件中硬编码了数据库URL，需要改为从配置读取，方便测试时替换：

# app/core/settings.py 示例配置
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DB_URL: str = "mysql://user:pass@localhost/real_db"

settings = Settings()

# app/main.py 启动事件修改
@app.on_event("startup")
async def startup_event():
    await init_db(settings.DB_URL, schemas=True)

然后在fixture中临时替换配置：

# real_client fixture中添加配置替换
from app.core import settings

@pytest.fixture(scope="function")
async def real_client():
    original_db_url = settings.DB_URL
    settings.DB_URL = "mysql://user:pass@localhost/real_db"
    
    await init_db(settings.DB_URL, schemas=False)
    try:
        async with LifespanManager(app):
            async with AsyncClient(transport=ASGITransport(app), base_url="http://localhost") as c:
                yield c
    finally:
        await Tortoise._drop_databases()
        await Tortoise.close_connections()
        settings.DB_URL = original_db_url  # 恢复原配置

# test_client fixture中添加配置替换
@pytest.fixture(scope="function")
async def test_client():
    original_db_url = settings.DB_URL
    settings.DB_URL = "sqlite://:memory:"
    
    await init_db(settings.DB_URL, create_db=True, schemas=True)
    # ... 初始化测试数据 ...
    try:
        async with AsyncClient(transport=ASGITransport(app), base_url="http://test") as client:
            yield client
    finally:
        await Tortoise._drop_databases()
        await Tortoise.close_connections()
        settings.DB_URL = original_db_url

核心改进点 #

• 彻底隔离两个客户端的数据库环境，避免全局状态污染
• 每个fixture独立管理数据库的初始化与销毁
• 通过动态配置确保App在测试时使用指定的数据库

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