单测通过批量运行失败：FastAPI异步SQLite测试异常排查

问题原因分析 #

1. 表创建顺序错误：若Team表存在指向Club表的外键，先创建Team表会导致外键约束无法生效，后续插入数据可能静默失败。
2. 全局测试数据被污染：club_data和team_data为全局变量，第一个测试运行时可能意外修改数据，导致第二个测试填充数据库时数据为空。
3. 会话事务处理不当：populate_db fixture中commit后调用session.begin()开启未提交事务，结合override_get_db的async with包裹，可能导致查询无法读取已提交数据。
4. 全局引擎连接池冲突：全局共享的engine在批量测试时，连接池状态未正确重置，第二个测试可能复用了第一个测试的已清理连接。

解决方案 #

1. 修正表创建顺序 #

确保先创建被依赖的Club表，再创建Team表：

async with engine.begin() as conn:
    # 先创建被依赖的Club表
    await conn.run_sync(lambda conn: Club.__table__.create(conn))
    # 再创建依赖Club的Team表
    await conn.run_sync(lambda conn: Team.__table__.create(conn))

2. 使用动态生成的测试数据 #

将全局静态数据改为函数生成，每次测试获取新实例：

# 在database1.py中定义数据生成函数
def get_club_data():
    return [
        Club(id=1, name="Club A"),
        # 其他测试数据...
    ]

def get_team_data():
    return [
        Team(id=1, name="Team 1", club_id=1, status="active"),
        # 其他测试数据...
    ]

# 在populate_db fixture中调用函数获取新数据
async with AsyncSessionFactory() as session:
    try:
        session.add_all(get_club_data())
        session.add_all(get_team_data())
        await session.commit()
        yield session
    finally:
        await session.close()

3. 优化会话事务处理 #

移除不必要的事务开启逻辑，简化会话传递：

# 修改populate_db fixture
@pytest_asyncio.fixture(scope="function")
async def populate_db():
    async with engine.begin() as conn:
        await conn.run_sync(lambda conn: Club.__table__.create(conn))
        await conn.run_sync(lambda conn: Team.__table__.create(conn))

    async with AsyncSessionFactory() as session:
        session.add_all(get_club_data())
        session.add_all(get_team_data())
        await session.commit()
        # 直接yield会话，无需额外开启事务
        yield session

    async with engine.begin() as conn:
        await conn.run_sync(lambda conn: Team.__table__.drop(conn))
        await conn.run_sync(lambda conn: Club.__table__.drop(conn))

# 修改create_client中的override_get_db
async def override_get_db():
    # 直接yield会话，避免重复包裹async with
    yield populate_db

4. 为每个测试创建独立引擎 #

将全局引擎改为function级fixture，确保测试完全隔离：

@pytest_asyncio.fixture(scope="function")
async def async_engine():
    # 每个测试创建独立的内存数据库引擎
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    yield engine
    # 测试结束后销毁引擎
    await engine.dispose()

@pytest_asyncio.fixture(scope="function")
async def populate_db(async_engine):
    async with async_engine.begin() as conn:
        await conn.run_sync(lambda conn: Club.__table__.create(conn))
        await conn.run_sync(lambda conn: Team.__table__.create(conn))

    # 为当前引擎创建专属会话工厂
    AsyncSessionFactory = sessionmaker(
        bind=async_engine, class_=AsyncSession, expire_on_commit=True
    )

    async with AsyncSessionFactory() as session:
        session.add_all(get_club_data())
        session.add_all(get_team_data())
        await session.commit()
        yield session

    async with async_engine.begin() as conn:
        await conn.run_sync(lambda conn: Team.__table__.drop(conn))
        await conn.run_sync(lambda conn: Club.__table__.drop(conn))

验证方案 #

执行批量测试命令确认修复效果：

poetry run pytest team_test.py

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