FastAPI异步Pytest测试current_user fixture数据库残留问题解决

我用于FastAPI应用测试的异步Pytest fixture如下：

# reset the database before each test
@pytest_asyncio.fixture(autouse=True)
async def reset_db() -> None:
    """重置数据库。"""
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)


# 覆盖数据库连接以使用测试数据库
async def get_database_override() -> AsyncGenerator[AsyncSession, Any]:
    """返回测试用数据库连接。"""
    async with async_test_session() as session:
        yield session


@pytest_asyncio.fixture()
async def test_db() -> AsyncGenerator[AsyncSession, Any]:
    """提供测试用数据库连接的fixture。"""
    async with async_test_session() as session:
        yield session


@pytest_asyncio.fixture()
async def client() -> AsyncGenerator[AsyncClient, Any]:
    """提供应用测试客户端的fixture。"""
    app.dependency_overrides[get_db] = get_database_override
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://testserver/api/v1",
        headers={"Content-Type": "application/json"},
        timeout=10,
    ) as client:
        yield client
    app.dependency_overrides = {}


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


@pytest_asyncio.fixture(scope="function")
def setup_cache():
    fake_redis = fakeredis.FakeStrictRedis()
    FastAPICache.init(RedisBackend(fake_redis), prefix="fastapi-cache")
    yield
    FastAPICache.clear()


@pytest_asyncio.fixture(scope="function")
async def current_user(test_db: AsyncSession):
    test_user = User(id=1, email="test@example.com", password="hashedpassword")
    test_db.add(test_user)
    await test_db.commit()
    await test_db.refresh(test_user)

    def override_get_current_user():
        return test_user

    app.dependency_overrides[get_current_user] = override_get_current_user

    yield test_user

    await test_db.delete(test_user)
    await test_db.commit()

尝试编写current_user fixture后，测试发现数据库的users表中仍残留测试用户数据：

SELECT * FROM users

1 | test@example.com | hashedpassword 

请问如何配置fixture，确保所有测试结束后数据库保持干净？

解决方案 #

问题根源 #

手动删除用户的方式不可靠，主要原因包括：

1. 测试过程中test_db的事务状态与删除操作的事务状态不匹配，导致删除未生效
2. 原reset_db的每次测试前删库重建逻辑，与session生命周期冲突，反而掩盖了事务层面的问题

更高效可靠的方案是利用数据库事务回滚，自动撤销测试中的所有数据操作，无需手动清理。

具体修改步骤 #

1. 修改test_db fixture，启用事务自动回滚
   将测试用session改为在事务中运行，测试结束后自动回滚所有操作：

@pytest_asyncio.fixture()
async def test_db() -> AsyncGenerator[AsyncSession, Any]:
    """提供测试用数据库连接，测试后自动回滚事务。"""
    async with async_test_session() as session:
        # 开启事务
        async with session.begin():
            yield session
        # 测试结束后回滚所有操作
        await session.rollback()

2. 简化current_user fixture，移除手动删除逻辑
   事务回滚会自动撤销用户的插入操作，无需手动删除：

@pytest_asyncio.fixture(scope="function")
async def current_user(test_db: AsyncSession):
    test_user = User(id=1, email="test@example.com", password="hashedpassword")
    test_db.add(test_user)
    await test_db.commit()
    await test_db.refresh(test_user)

    def override_get_current_user():
        return test_user

    app.dependency_overrides[get_current_user] = override_get_current_user

    yield test_user

    # 仅清理依赖覆盖，无需手动删除用户
    app.dependency_overrides.pop(get_current_user, None)

3. 优化数据库初始化逻辑
   原reset_db每次测试前删库重建效率极低，改为仅在测试会话启动时初始化一次：

@pytest_asyncio.fixture(scope="session", autouse=True)
async def init_db() -> None:
    """测试会话开始时初始化数据库一次。"""
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    await async_engine.dispose()

效果说明 #

每个测试使用的test_db session会开启独立事务，测试中所有的插入、修改操作都在这个事务内。测试结束后回滚事务，数据库会自动回到事务开启前的干净状态，完全避免数据残留问题，同时比每次删库重建的测试效率提升明显。

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