如何在FastAPI中玩转“时光倒流”的数据库事务回滚测试？

1. 多模块集成测试实践

在 FastAPI 项目中，随着功能模块增多，集成测试成为确保系统整体稳定性的关键。

1.1 为何需要事务回滚测试

1.2 核心实现方案

使用 pytest + SQLAlchemy + FastAPI TestClient 组合，通过以下组件实现事务控制：

2. 数据库事务回滚测试模式

通过事务嵌套实现 "沙盒环境"，测试中所有数据库操作会在用例结束时自动回滚，就像从未发生过。

2.1 实现流程图

graph TD
    A[测试开始] --> B[创建新事务]
    B --> C[执行测试操作]
    C --> D[接口请求]
    D --> E[数据库修改]
    E --> F[结果断言]
    F --> G[事务回滚]
    G --> H[清理资源]

2.2 代码实现

创建 tests/conftest.py 配置文件：

# 所需依赖：pytest==7.4.4, sqlalchemy==2.0.28, pytest-asyncio==0.23.6
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# 初始化异步数据库引擎
TEST_DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/test_db"
engine = create_async_engine(TEST_DATABASE_URL, echo=True)

# 配置异步session工厂
AsyncTestingSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False
)

# 核心事务回滚夹具
@pytest.fixture(scope="function")
async def db_session():
    """创建嵌套事务的沙盒环境"""
    async with AsyncTestingSessionLocal() as session:
        async with session.begin_nested():  # 创建SAVEPOINT
            yield session
        # 退出时自动回滚所有操作

3. 完整测试案例

模拟用户注册+资料修改的集成测试场景。

3.1 测试代码

# tests/test_user_integration.py
from fastapi.testclient import TestClient
from sqlalchemy import select
from app.main import app
from app.models import User

client = TestClient(app)

@pytest.mark.asyncio
async def test_user_flow(db_session: AsyncSession):
    # 1. 注册新用户
    res = client.post("/users/", json={"name": "test", "email": "[email protected]"})
    assert res.status_code == 201
    
    # 2. 验证数据库写入（此时在事务内可见）
    result = await db_session.execute(select(User).where(User.email == "[email protected]"))
    user = result.scalar_one()
    assert user.name == "test"
    
    # 3. 修改用户资料
    res = client.patch(f"/users/{user.id}", json={"name": "updated"})
    assert res.status_code == 200
    
    # 4. 再次验证
    await db_session.refresh(user)
    assert user.name == "updated"
    
# 测试结束时，所有数据库操作自动回滚

3.2 关键机制解析

1. begin_nested()：创建 SQL SAVEPOINT 保存点
2. yield session：测试代码在此作用域执行
3. 退出作用域：自动执行 ROLLBACK TO SAVEPOINT
4. 原子性：整个测试过程中即使有多个 DB 操作，也会被视作单一事务单元

4. 常见场景适配方案

4.1 第三方服务模拟

当集成支付等外部服务时，使用 responses 库拦截 HTTP 请求：

# 新增依赖：responses==0.24.1
import responses

@responses.activate
def test_payment_flow():
    responses.post("https://pay.demo.com", json={"success": True})
    res = client.post("/payments", json={"amount": 100})
    assert res.json()["status"] == "completed"

4.2 异步任务测试

对于 Celery 等异步任务，采用 pytest_celery 插件：

# 新增依赖：pytest-celery==0.0.1
from pytest_celery import CeleryTestWorker

def test_async_task(celery_worker: CeleryTestWorker):
    res = client.post("/tasks/email", json={"to": "[email protected]"})
    task_id = res.json()["id"]
    
    # 显式等待任务完成
    result = celery_worker.wait_for_result(task_id)
    assert result["status"] == "sent"

Quiz: 集成测试陷阱排查

❓ 当测试中遇到 IntegrityError 唯一约束冲突时，最可能的原因是什么？ A) 测试数据清理不彻底
B) 事务隔离级别配置错误
C) 未正确启用回滚机制
D) 数据库连接池耗尽

常见报错解决方案

报错 1：SAVEPOINT does not exist

现象：

sqlalchemy.exc.InvalidRequestError: This session's transaction has been rolled back 
due to a previous exception during flush...

原因：嵌套事务中执行了 DDL 操作（如 CREATE TABLE）

解决方案：

1. 避免在测试事务中执行迁移操作
2. 初始化时创建好测试表结构：

# 在 conftest.py 中添加
@pytest.fixture(scope="session", autouse=True)
def initialize_db():
    # 在此处运行数据库迁移脚本

报错 2：Connection is not acquired

现象：

RuntimeError: <Task ...> ... Connection is not acquired.

原因：异步连接未正确释放

解决方案：

# 改造夹具确保资源释放
@pytest.fixture
async def db_session():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with AsyncTestingSessionLocal() as session:
        try:
            async with session.begin_nested():
                yield session
        finally:
            await session.close()  # 确保显式关闭连接