Pytest测试无法读取数据库更新后数据致断言失败问题排查
问题根因
这是SQLAlchemy会话一级缓存(身份映射机制)导致的跨会话数据不一致问题:
- 每个通过
generate_session()生成的Session实例是完全独立的,各自维护自己的对象缓存 - 测试用例中先创建了会话A,加载了测试用户和关联任务到会话A的缓存中
- 调用
hide_task_controller时,内部hide_task方法自行创建了独立的会话B,通过会话B修改任务状态并提交到数据库,数据库层面数据确实更新成功 - 回到测试用例上下文后,会话A完全不知道其他会话对数据做了修改,当你查询同ID任务时,SQLAlchemy会直接返回会话A缓存中存储的旧状态对象,不会主动重新查询数据库,因此断言读到的是旧值,和数据库实际值不一致。
执行流程复现
- 测试用例初始化:创建会话A,查询测试用户、关联任务,任务状态为
OPEN,该对象存入会话A缓存 - 调用隐藏任务接口:
hide_task方法创建会话B,查询任务、修改状态为HIDDEN、提交事务,数据库记录更新完成,会话B生命周期结束 - 测试断言阶段:通过会话A查询同ID任务,命中一级缓存直接返回旧对象,状态仍为
OPEN,断言失败 - 手动直连数据库查询:绕过所有SQLAlchemy会话缓存,直接读取磁盘上的已提交数据,因此能看到正确的
HIDDEN状态
可落地方案
方案1:临时清空当前会话缓存(改造成本最低)
在调用完业务方法、查询断言前,主动让当前会话的缓存失效,强制后续查询走数据库拉取最新值,仅需修改测试用例代码:
def test_tasks_hiding(give_test_user_account: UserTable): session: Session = next(generate_session()) user = get_user_by_id(give_test_user_account.id, session) task_id = user.tasks[0].id hide_task_controller(task_id, user) # 失效当前会话所有缓存对象,后续查询强制访问数据库 session.expire_all() task = session.query(TaskTable).filter_by(id=task_id).first() assert user.tasks[0] assert task.state == HIDDEN delete_task(user.tasks[0].id, session) delete_user_from_database(user, session)
如果不想清空所有缓存,也可以仅失效单个关联对象,或者在查询时加populate_existing()强制刷新缓存:
# 方式1:仅失效当前用户关联的任务对象 session.expire(user.tasks[0]) # 方式2:查询时强制刷新已有缓存 task = session.query(TaskTable).filter_by(id=task_id).populate_existing().first()
方案2:统一全链路会话管理(长期规范方案)
当前业务代码存在自行创建会话的逻辑,很容易在其他场景也踩跨会话不一致的坑,建议重构会话传递逻辑:整个请求/测试调用链路共用同一个Session实例,业务方法不要自行调用generate_session()创建新会话,而是将Session作为参数传入。
重构后的hide_task方法示例:
def hide_task(self, task_id: int, session: Session): task = get_task_by_id(task_id, session) task.state = HIDDEN session.commit()
控制器、测试代码调用该方法时,统一传入上下文持有的同一个会话实例,从根源上避免多会话缓存不一致问题。
内容的提问来源于stack exchange,提问作者Cross
相关产品推荐
相关产品推荐

