最稳妥的方式是用scope="function"的fixture自动管理事务生命周期:每个测试前开启事务,结束后强制回滚,确保隔离性;需禁用autocommit、避免session.commit()、复用同一session对象。

pytest中用transactional fixture实现测试级回滚
直接在测试里手动rollback()容易漏写或顺序错,最稳妥的方式是用fixture自动管理事务生命周期。核心思路是:每个测试开始前开启新事务,结束后强制回滚,不提交任何变更。
关键点在于绕过ORM默认的自动提交行为,并确保fixture作用域精准控制到单个测试函数:
- 用
scope="function"避免跨测试污染 - 数据库连接必须支持保存点(PostgreSQL/SQLite OK,MySQL需InnoDB且关闭autocommit)
- fixture里不能调用
session.commit(),否则回滚失效
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
<p>@pytest.fixture(scope="function")
def db_session():
engine = create_engine("sqlite:///:memory:", echo=False)
Session = sessionmaker(bind=engine)
session = Session()</p><h1>开启事务</h1><pre class="brush:python;toolbar:false;">trans = session.begin()
try:
yield session
finally:
session.close()
trans.rollback()
避免使用autocommit=True导致回滚失效
很多初学者在创建engine时加autocommit=True,这会让所有SQL执行后立刻提交,trans.rollback()变成空操作——事务根本没真正开启。
正确做法是保持默认autocommit=False,靠session.begin()显式控制事务边界:
-
create_engine("postgresql://...", autocommit=False)(默认就是False,显式写出更安全) - ORM层所有
add()/delete()都只是暂存,直到commit()才生效 - 如果测试中意外调用了
session.commit(),回滚会失败,pytest会报InvalidRequestError: This session is in 'committed' state
测试中插入数据后查不到?检查session隔离级别
用db_session插入数据后,立即用同一session查询能查到;但如果换一个新session(比如在测试里另建Session()),默认查不到——这是事务隔离的正常行为,不是bug。
常见误判场景:
- 在测试里调用被测代码,而被测代码内部新建了session,导致看不到当前事务的数据
- 期望用
sqlite3.connect()直连查表,但内存数据库每个连接是独立实例
解决方式统一:所有操作复用同一个db_session fixture传入的session对象,不要在测试内部重新实例化。
SQLite内存数据库不适合多线程/多进程并发测试
sqlite:///:memory:对单测试有效,但pytest若开启-n auto(多进程)或测试本身涉及线程,每个worker会得到独立的内存DB,回滚只作用于当前进程,无法模拟真实数据库的并发行为。
真要测并发场景,得换真实DB实例:
- 用
pytest-xdist时,改用临时文件DB:sqlite:///tmp/test_{}.db,每个worker用唯一文件名 - PostgreSQL推荐用
pg_tmp扩展或template0快速建库,避免测试间残留数据 - 别忘了在fixture teardown里清理临时DB文件或schema
事务回滚本身很简单,难的是让整个测试环境的行为和生产一致——连接复用、隔离级别、并发模型,这些细节漏掉一个,回滚就形同虚设。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











