使用SqlAlchemy为每个测试函数重建数据库运行失败该如何解决
问题原因
首先你的每个测试单独创建数据库的思路是完全合理的,这种方案隔离性极强,从根源避免了不同测试用例之间的数据污染,适合小型测试场景。你当前的问题出在代码实现的两个细节上:
- 全局复用ORM实例:你定义的
USERS是全局的User类实例列表,第一次测试提交后,这些实例已经被SQLAlchemy标记为已持久化状态,自带生成的主键和状态标识。第二个测试将这些旧实例添加到新session时,SQLAlchemy会判定为要执行更新操作而非插入,自然新数据库中不会生成对应记录。 - 全局
session_factory存在潜在状态泄漏风险,虽然你当前的重置逻辑暂时没有触发问题,但依赖全局变量管理数据库连接在多线程/多测试场景下容易出问题。
修复方案
最简单的修改方式是不要直接全局存储ORM实例,改为存储构造参数,每次插入数据时新建实例:
# 原USERS定义修改为存储参数字典 USERS = [ {"username": "Güllich", "password": "action direct"}, {"username": "Graham", "password": "the island"}, ] # fixture中插入数据的逻辑修改为新建实例 @pytest.fixture def dbase(tmp_path): sqlite_file = tmp_path / "test_db.sqlite" connection_string = f"sqlite+pysqlite:///{sqlite_file.as_posix()}" print("creating database at: %s", connection_string) init_db(connection_string) session = session_factory() # 每次新建User实例,避免复用旧状态 for user_data in USERS: session.add(User(**user_data)) session.commit() session.close() yield close_db()
如果想进一步优化,可以去掉全局session_factory,改为fixture直接返回session实例,避免全局变量带来的意外状态泄漏:
# 去掉全局session_factory定义 def init_db(connection_url: str): engine = create_engine(connection_url, future=True) SqlAlchemyBase.metadata.create_all(bind=engine) session_factory = orm.sessionmaker(bind=engine) # 返回session工厂,交由fixture管理生命周期 return engine, session_factory @pytest.fixture def dbase(tmp_path): sqlite_file = tmp_path / "test_db.sqlite" connection_string = f"sqlite+pysqlite:///{sqlite_file.as_posix()}" engine, session_factory = init_db(connection_string) session = session_factory() for user_data in USERS: session.add(User(**user_data)) session.commit() session.close() yield session_factory # 测试结束后清理 session_factory.close_all() engine.dispose()
两种测试方案对比
你提到的两种数据库测试方案各有适用场景:
- 每个测试新建独立数据库:隔离性最高,完全不用担心测试之间的状态干扰,实现逻辑简单,适合测试用例数量不多、表结构不复杂的场景,缺点是测试量大时建库插数据的开销较高。
- 单库+事务回滚:全局只创建一次数据库和表结构,每个测试启动时开启事务,测试执行完成后直接回滚事务,不用重建库表,性能更高,适合测试用例多、表结构复杂的场景,唯一要注意的是测试代码本身不能手动提交事务,否则会破坏回滚逻辑。
内容的提问来源于stack exchange,提问作者sanders
相关产品推荐
相关产品推荐

