You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.26 12:03:25