在上一篇流水线提效中我们提到:“用例执行顺序优化"会单独出文章讲解,本文就来填这个坑。当用例数量上千、并发数开到 8 时,经常会遇到一个反直觉现象:并发越高,反而越慢,通过率也变低。根因不在并发本身,而在"用例执行顺序不合理"导致的环境资源争抢。本文从问题定位、方案演进到最终代码实现,一步步讲清如何用一段 20 行的 conftest 钩子把 pytest 用例顺序打散到"文件级”。
一、问题现象:并发越高反而越慢
某项目用例规模 ~2000 条,pytest-xdist 8 进程并发。现象:
| 现象 | 表现 |
|---|---|
| CPU 利用率波动大 | api 阶段 20%、compute 阶段 95%+ 反复横跳 |
| 创建云主机用例集中爆发 | 8 个 worker 同时打云平台 API,触发限流 |
| 用例整体耗时变长 | 单条平均耗时从 2s → 6s |
| 通过率下降 | 95% → 82%,偶发超时增多 |
直观观察 Jenkins 日志:
gw0: tests/api/test_login.py::test_login ← 低负载 gw1: tests/api/test_user.py::test_create_user ← 低负载 ... gw0: tests/compute/test_vm.py::test_create_vm ← 高负载 gw1: tests/compute/test_vm.py::test_clone_vm ← 高负载 gw2: tests/compute/test_disk.py::test_create_disk ← 高负载 ... 8 个 worker 全在打 compute,云平台 API 限流资源使用呈现"闲时很闲、忙时很忙"的脉冲式波动——并发根本没有平滑负载。
二、根因分析:两个准则 + 一个默认行为
自动化用例管理有两个不成文的准则:
准则 1:用例按功能模块分类存放
tests/ ├── api/ ← 接口测试,低负载 │ ├── test_login.py │ └── test_user.py ├── compute/ ← 云主机模块,高负载 │ ├── test_vm.py │ └── test_disk.py └── network/ └── test_subnet.py好处:管理清晰、维护方便
隐患:相同功能的用例聚在一起,负载特征也聚在一起
准则 2:pytest 默认按名称排序执行
pytest 收集完用例后,默认按nodeid字母顺序排序:
tests/api/test_login.py::test_a tests/api/test_login.py::test_b tests/api/test_user.py::test_a tests/compute/test_vm.py::test_a ← 高负载集中在这里 tests/compute/test_vm.py::test_b tests/compute/test_disk.py::test_a问题叠加:并发 + 默认顺序
- xdist 默认
--dist=load:按用例逐个分发给 worker - 但分发顺序就是默认顺序(按文件名)
- 结果:所有 worker 同时从
api转到compute,高负载用例在同一时间窗集中爆发
时间 → gw0: api_a → api_b → vm_a → vm_b → disk_a gw1: api_a → api_b → vm_a → vm_b → disk_a gw2: api_a → api_b → vm_a → vm_b → disk_a ↑ 所有 worker 同时进入高负载区这就是"闲时很闲、忙时很忙"的根因。
三、方案一:用例级随机打乱(不推荐)
第一反应:用现成插件打乱顺序即可:
| 插件 | 粒度 | 用法 |
|---|---|---|
pytest-randomly | 用例级 | pip install pytest-randomly |
pytest-random-order | 用例级 | pytest --random-order |
装上后跑一遍,结果更慢了。为什么?
反例分析
假设tests/compute/test_vm.py有 3 个用例:
# tests/compute/test_vm.py@pytest.fixture(scope="module")defvm():"""module 级 fixture:一次创建,全文件共享"""vm=create_vm()yieldvm delete_vm(vm)deftest_a(vm):...# 共用 vmdeftest_b(vm):...# 共用 vmdeftest_c(vm):...# 共用 vm单独跑文件:1 次 create_vm + 3 次测试 + 1 次 delete_vm = 5 次云平台调用。
用例级打乱后:3 个用例被分到不同位置执行,可能跨文件、跨 worker,fixture 复用被破坏:
worker A: test_a(vm) → 创建 vm1 → 用 → 销毁 worker B: test_b(vm) → 创建 vm2 → 用 → 销毁 worker C: test_c(vm) → 创建 vm3 → 用 → 销毁变成 3 次 create_vm + 3 次测试 + 3 次 delete_vm = 9 次云平台调用。
fixture 范围越大(module/session),用例级打乱的代价越大。
四、方案二:文件级随机打乱(推荐)
正确的优化方向:打乱到文件级即可,文件内保持原顺序。
原顺序: 文件级打乱后: api/test_login.py compute/test_vm.py ← 高负载先来一个 api/test_user.py api/test_login.py compute/test_vm.py network/test_subnet.py compute/test_disk.py compute/test_disk.py network/test_subnet.py api/test_user.py好处:
| 维度 | 收益 |
|---|---|
| 负载分布 | 高负载用例被打散到不同时间段,避免集中爆发 |
| fixture 复用 | 文件内顺序不变,module/session fixture 仍然只跑一次 |
| 环境压力 | 云平台 API 请求被平滑到整个执行周期 |
用一句话总结:
打散要恰到好处——足够打散负载峰值,但不要破坏 fixture 复用。
五、核心代码:20 行 conftest 搞定文件级打散
直接上代码,加在项目根目录conftest.py:
# conftest.pyimportrandomdefpytest_collection_modifyitems(session,config,items:list):"""用于随机化打乱测试用例文件"""items_list=[]alike=[]foriinitems:i.add_marker(i.name)ifnotalikeori.parent.parent.nodeid==alike[0].parent.parent.nodeid:alike.append(i)else:items_list.append(alike)alike=[i]ifalike:items_list.append(alike)random.seed(80)random.shuffle(items_list)random.seed()items[:]=[itemforgroupinitems_listforitemingroup]用法:放到conftest.py即可,无需任何命令行参数。每次跑同一个种子(80)打出同样的顺序,可复现。
六、代码逐段拆解
1. 钩子签名
defpytest_collection_modifyitems(session,config,items:list):pytest_collection_modifyitems:pytest 收集完用例后、开始执行前的钩子,可修改 itemsitems:所有用例对象列表,按默认(字母)顺序排列- 修改
items[:]即可改变执行顺序
2. 按文件分组
items_list=[]alike=[]foriinitems:i.add_marker(i.name)ifnotalikeori.parent.parent.nodeid==alike[0].parent.parent.nodeid:alike.append(i)else:items_list.append(alike)alike=[i]ifalike:items_list.append(alike)核心:i.parent.parent.nodeid是用例所在的目录(粗粒度),相同目录的用例聚到一组alike,碰到新目录就把当前组存起来、开启新组。
最终items_list是"目录块"的列表:
items_list = [ [api_test_a, api_test_b, api_test_c], ← api 目录 [compute_test_a, compute_test_b], ← compute 目录 [network_test_a], ← network 目录 ]注:用
i.parent.parent.nodeid(祖父目录)作为分组键。如果你的目录结构更深,可以调整为i.parent.nodeid(父目录,即文件级)或更上层。
3. 随机打乱组
random.seed(80)random.shuffle(items_list)random.seed()random.seed(80):固定种子,保证每次跑顺序一致(可复现 bug)random.shuffle(items_list):打乱目录块的顺序,块内顺序保持不变random.seed():重置种子,避免影响后续代码的随机数
4. 展平回 items
items[:]=[itemforgroupinitems_listforitemingroup]把"列表的列表"展平回一维列表,赋值给items[:]——必须用items[:]而非items =,前者是原地修改,后者只是局部变量赋值,对 pytest 不生效。
5. 顺带加 marker
i.add_marker(i.name)把每个用例的名字作为 marker 加上去,方便后续用pytest -m test_a单独跑某条用例。可选,删掉也不影响主逻辑。
七、效果验证
优化前
worker 0: api_a → api_b → vm_a → vm_b → disk_a worker 1: api_a → api_b → vm_a → vm_b → disk_a ... CPU 使用率: ▁▁▂▃▆█████▆▃▂▁▁ (脉冲式) 云平台 QPS: ▁▁▁▁▁███▇▆▃▁▁▁ (集中爆发) 通过率: 82% 总耗时: 24min优化后
worker 0: vm_a → api_a → subnet_a → disk_a → api_b worker 1: api_b → vm_b → disk_b → api_a → subnet_b ... CPU 使用率: ▃▄▅▄▅▆▅▄▅▆▅▄▅ (平滑) 云平台 QPS: ▃▄▃▄▃▄▃▄▃▄▃▄▃ (均匀分布) 通过率: 94% 总耗时: 14min关键指标改善:
| 指标 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| 总耗时 | 24min | 14min | -42% |
| 通过率 | 82% | 94% | +12% |
| CPU 利用率方差 | 0.32 | 0.08 | 平滑 4x |
| 云平台 API 限流次数 | 17 | 0 | 消除 |
八、常见坑点速查
| 坑 | 现象 | 解决 |
|---|---|---|
用items =而非items[:] | 顺序没变 | 必须原地修改 |
| 种子不固定 | 每次顺序不同,bug 不可复现 | random.seed(80)固定 |
| 分组粒度选错 | 打散效果差或破坏 fixture | 按目录结构选parent或parent.parent |
与pytest-randomly共存 | 两个随机冲突 | 二选一,本文方案与randomly互斥 |
xdist--dist=loadscope冲突 | 文件级打散被 scope 收回 | 配合--dist=load用 |
| 同文件用例依赖顺序 | 打乱后失败 | 本就不该有顺序依赖,去掉依赖或加 marker 控 |
| session fixture 仍是热点 | 多 worker 同时进入 fixture | fixture 内部要幂等或加锁 |
九、扩展思路
1. 按 marker 加权打散
让"高负载 marker"之间留间隔:
defpytest_collection_modifyitems(session,config,items):# 把高负载用例按固定间隔插入heavy=[iforiinitemsifany(m.name=="heavy"formini.iter_markers())]light=[iforiinitemsifinotinheavy]random.seed(42)random.shuffle(light)random.shuffle(heavy)# 每隔 N 个 light 插一个 heavyresult=[]heavy_idx=0foridx,iteminenumerate(light):result.append(item)ifidx%5==0andheavy_idx<len(heavy):result.append(heavy[heavy_idx])heavy_idx+=1items[:]=result2. 按历史耗时排序
记录每条用例的历史耗时,长的用例先跑——早暴露问题:
defpytest_collection_modifyitems(session,config,items):durations=session.config.cache.get("case_durations",{})items.sort(key=lambdai:-durations.get(i.nodeid,0))配合pytest --cache用。
3. 配合--dist=loadgroup精细控制
@pytest.mark.group("heavy-vm")deftest_create_vm():...pytest--dist=loadgroup让带heavy-vmmarker 的用例集中到一个 worker,避免并发打云平台。
4. 失败用例优先重跑
defpytest_collection_modifyitems(session,config,items):failed=session.config.cache.get("lastfailed",[])items.sort(key=lambdai:i.nodeidinfailed,reverse=True)上次失败的用例排前面,早跑早暴露。
5. 同目录但想拆开
如果某目录用例太多,单纯按目录打散不够,可以按"文件 + 序号奇偶"再分:
foriinitems:group_key=(i.parent.nodeid,hash(i.name)%2)十、总结
pytest 用例执行顺序优化的核心思想:打散负载峰值,但保留 fixture 复用。掌握本文需要抓住三条主线:
- 根因:默认字母排序 + 同功能聚集 = 负载脉冲
- 粒度:用例级打散破坏 fixture,文件级/目录级打散刚刚好
- 可复现:用固定随机种子保证每次跑顺序一致,bug 可复现
一句话记法:
用例顺序问题,不在"打不打",而在"打多碎"——文件级打散是性价比最高的平衡点。
实践建议:
- 第一步:用
pytest --durations=10找出最慢的 10 个用例,看是不是同一目录 - 第二步:加上本文 20 行 conftest,对比耗时与通过率
- 第三步:根据实际目录结构调整分组粒度(
parentvsparent.parent) - 第四步:把高负载用例显式打
heavymarker,配合loadgroup进一步精细化
把执行顺序打散做对后,并发才真正发挥威力——8 个 worker 才能稳定跑出 4-6 倍速,而不是脉冲式争抢资源。
建议动手实验:在项目根
conftest.py加上本文 20 行代码,跑两次对比pytest --durations=20输出——直观看到高负载用例被打散到不同时间段。