pytest fixture 中用 yield 管理 pymysql 连接更稳,因其确保无论测试是否异常都会执行清理;应避免全局连接池、多次 close、共享连接,并在 yield 前 ping 保活和重置状态。
频繁在 pytest 测试中用 pymysql.connect() 打开又关闭数据库连接,不是“看起来不优雅”的问题,而是会直接拖慢测试执行、触发 MySQL 的 max_connections 限制、甚至让 MySQL server has gone away 错误集中爆发。
pytest fixture 中用 yield 管理连接,为什么比 setup/teardown 更稳?
因为 yield 模式天然支持“资源获取 → 用例执行 → 异常时也能保证清理”的闭环。setup_method 可能因断言失败跳过 teardown;而 yield 后的代码块,无论测试函数是否抛异常、是否提前 return,都会执行。
常见错误现象:
- 测试中途报
OperationalError: (2013, "Lost connection to MySQL server during query") - 多个测试并行跑时出现
Too many connections - 单个测试里多次
connection.close()导致Connection already closed
实操建议:
- 把
pymysql.connect()放在yield前,把connection.close()放在yield后 - 不要在 fixture 内部做
cursor.execute()—— 那是测试逻辑,不是连接生命周期管理 - 加
scope="function"(默认),避免多个测试共享一个连接引发状态污染
示例:
@pytest.fixture
def db_conn():
conn = pymysql.connect(
host="localhost",
user="test",
password="pass",
database="testdb",
charset="utf8mb4",
autocommit=True,
)
yield conn
conn.close() # 这行一定会执行
为什么不能只靠 try/finally,非要用 yield?
你当然可以用 try/finally 手动写 fixture,但 yield 是 pytest 原生约定,它和 pytest 的 fixture 缓存、作用域、参数化机制深度集成。手动 try/finally 容易漏掉 scope 控制或与 request.addfinalizer 冲突。
Python 3.14.3
微软官方的 Python 扩展,是 VS Code 安装量最高的扩展(209M+)。集成 IntelliSense(通过 Pylance)、调试(通过 Python Debugger)、代码检查、格式化、重构和单元测试等功能。支持 Jupyter Notebook、虚拟环境管理和多 Python 版本切换。
关键差异点:
-
yield返回的对象可被 pytest 自动注入到测试函数参数中;try/finally写在普通函数里做不到这点 - 当 fixture 带参数(如
@pytest.mark.parametrize)时,yield能确保每个参数组合都独立走一遍生命周期 - pytest 会把
yieldfixture 的清理逻辑排入专用清理队列,比手写addfinalizer更可靠
连接池要不要在测试 fixture 里配?
要,但别用 DBUtils.PooledDB 这类全局池——测试间需要隔离,且池内连接可能跨测试复用,导致事务残留或锁未释放。
更稳妥的做法:
- 测试环境用单连接 +
yield管理,简单可控 - 若真需模拟高并发场景,改用
sqlalchemy.create_engine(pool_pre_ping=True, pool_recycle=300),并确保scope="function"下每次创建新 engine(不是新连接,而是新 engine 实例) - 绝对避免在 conftest.py 顶层初始化一个全局
pool对象,否则 pytest-xdist 并行时会出竞态
容易被忽略的一点:MySQL 默认 wait_timeout=28800(8 小时),但测试运行时间短,真正危险的是 interactive_timeout 和连接空闲后被中间网络设备(如 NAT 网关)静默断连。所以即使用了 yield,也建议在 fixture 中显式调用 conn.ping(reconnect=True) 开头保活:
@pytest.fixture
def db_conn():
conn = pymysql.connect(...)
conn.ping(reconnect=True) # 主动探测并重连
yield conn
conn.close()
复杂点在于:ping 不是万能的,它只检测 socket 是否通,不保证事务上下文干净。如果你的测试依赖某个特定的 isolation level 或 autocommit 状态,得在 yield 前重置 —— 比如执行 conn.autocommit(True) 和 conn.select_db("testdb")。