pytest中必须用隔离的临时PostgreSQL实例(如pytest-postgresql),通过function级fixture封装显式rollback,确保每个测试拥有独立、干净的数据库连接,避免污染与并发问题。

pytest中用pytest-postgresql启动临时数据库实例
直接复用生产数据库测事务回滚,风险高、污染大、不可并行。必须用隔离的临时实例——pytest-postgresql是目前最稳的选择,它在测试前拉起一个独立PostgreSQL进程,测试完自动杀掉。
安装后,在conftest.py里注册fixture:
from pytest_postgresql import factories
postgresql_proc = factories.postgresql_proc()
postgresql = factories.postgresql('postgresql_proc')
这样每个测试函数拿到的postgresql fixture 就是一个干净、空的数据库URL(如postgresql://localhost:5432/test_db_123),互不干扰。
- 别用
psycopg2.connect()硬连本地默认库,那不是隔离环境 - 若用Docker启动PG,需确保端口不冲突,且
postgresql_proc能等容器就绪再返回URL - Windows下注意
postgresql_proc可能因路径空格失败,建议把PostgreSQL二进制放在无空格路径
用transaction.rollback()手动触发回滚而非依赖autocommit
Pytest本身不干预数据库连接行为。很多新手以为“加了@pytest.mark.usefixtures("postgresql")就会自动回滚”,其实不会——Python DB-API默认是autocommit=False,但事务仍需显式commit()或rollback(),否则连接关闭时可能隐式提交(取决于驱动实现)。
正确做法是在测试结束前强制回滚:
def test_user_insert_rolls_back(postgresql):
conn = psycopg2.connect(**get_pg_url(postgresql))
cur = conn.cursor()
cur.execute("INSERT INTO users (name) VALUES ('alice')")
# 不调 commit()
conn.rollback() # 关键:显式回滚
cur.execute("SELECT COUNT(*) FROM users")
assert cur.fetchone()[0] == 0
conn.close()
- 别依赖
conn.close()自动回滚;不同驱动行为不一致,psycopg2 2.9+ 默认close时不rollback - 如果用SQLAlchemy,得用
session.rollback(),不是connection.rollback();二者层级不同 - 在
teardown阶段rollback比在setup阶段更安全——避免setup失败导致后续测试误用脏数据
用pytest.fixture(scope="function")封装带rollback的DB连接
重复写conn.rollback()易漏,也难统一。用fixture封装是最省心的做法:
@pytest.fixture(scope="function")
def db_conn(postgresql):
conn = psycopg2.connect(**get_pg_url(postgresql))
try:
yield conn
finally:
conn.rollback()
conn.close()
这样每个测试拿到的db_conn都是开箱即用、用完即清的连接。
- scope必须是
"function",不能是"module"或"session",否则多个测试共享连接,rollback会互相覆盖 - yield前不要执行任何SQL;所有操作应在测试函数内进行,保证fixture只管生命周期
- 如果测试中需要提前commit(比如测幂等性),这个fixture就不适用——得另建一个
db_conn_with_commit,明确区分语义
遇到ActiveConnectionError或TransactionRollbackError怎么查
常见报错不是因为rollback没写,而是事务状态已破坏:比如在已rollback的连接上又执行cur.execute(),psycopg2会抛ActiveConnectionError;或者并发测试中两个测试误用了同一连接,触发TransactionRollbackError。
排查优先级:
- 确认是否在
yield后还对db_conn调用了任何方法(包括cursor())——fixture退出后连接已关闭 - 检查SQLAlchemy是否启用了
expire_on_commit=False,导致session缓存了已回滚的数据 - 用
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)看实际发了哪些SQL,确认rollback是否真被执行
事务回滚本身很简单,难的是让整个测试链路里的连接、session、cursor都处在预期状态。一旦出现异常,先盯住连接生命周期,而不是急着改SQL逻辑。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!











