ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

pytest多进程或多线程执行测试实例

pytest多进程或多线程执行测试实例 前言先厘清标题里的一个含混之处。pytest 本体是单进程、单线程顺序跑用例的它自己既不做多进程也不做多线程。所谓「pytest 并行」靠的是插件最主流的是pytest-xdist而它的并行单位是进程不是线程——每个 worker 是一个独立的 Python 子进程。想真正用「多线程」跑用例需要别的插件或者自己用标准库调度。还有一个容易被忽略的前提并行只在测试彼此独立时才有意义。如果用例之间共享全局状态、依赖执行顺序、或者抢同一个数据库、同一个端口那并行后会出现「单独跑全过、一起跑就随机失败」的现象。这类问题的根源是把测试写成了有状态的而不是并行本身有错。本文分三部分用pytest-xdist做多进程并行、用标准库ThreadPoolExecutor调度 pytest 子进程实现「线程级并发」、以及并行场景下必须注意的坑。示例以 Python 3.8 及以上为基准。一、准备一组可并行的用例并行演示需要一批互不依赖的用例。下面这个文件里的用例只用标准库不连数据库、不占端口适合并行。# 适用于 Python 3.8文件保存为 test_sample.pyimport timeimport pytestpytest.mark.parametrize(n, [1, 2, 3, 4, 5, 6])def test_slow_case(n):time.sleep(0.3) # 模拟一个慢用例assert n 0def test_fast_case():assert sum(range(10)) 45用pytest test_sample.py跑6 个慢用例会依次耗掉约 1.8 秒。下面的并行方案就是要把这段等待压下去。请自己用pytest --durations0看真实耗时本文不给具体倍数。二、多进程并行pytest-xdist2.1 安装与最短用法pip install pytest-xdistpytest test_sample.py -n auto-n是--numprocesses的简写。取值规则来自该插件文档取值含义-n auto按机器的物理 CPU 核数开 worker 进程-n logical按逻辑 CPU 核数开需要装psutil取不到时退回auto-n 4明确指定 4 个 worker-n 0关闭 xdist回到主进程顺序执行pytest-xdist的工作方式是主进程收集用例把用例分发给若干 worker 子进程执行再把结果汇总回主进程。所以它带来两个必然限制——worker 之间不共享内存每个 worker 会各自建立自己的 fixture 环境。2.2 分发策略--dist-n只决定开几个 worker--dist决定用例怎么分。策略行为适合场景--dist load默认谁闲给谁发下一个一般情况负载均衡最好--dist loadscope按模块分组同一模块的用例进同一个 worker有昂贵的模块级 fixture--dist loadfile按文件分组文件级 fixture 很重时--dist loadgroup按xdist_group标记分组需要把若干用例绑在一起--dist worksteal先均分再让跑完的 worker 去「偷」活用例耗时差异很大--dist no不分发等价于顺序执行调试一个实用的组合是把常用参数写进pytest.ini避免每次手敲[pytest]addopts -n auto --dist workstealtestpaths tests想临时压过配置命令行再给一次参数即可-n 1表示单 worker。2.3 何时该用进程、何时不该进程并行的代价是每个 worker 都要重新 import、重新搭 fixture。如果 fixture 本身很贵比如要建一次数据库 schema进程越多反而越慢。这种情况下用--dist loadscope或--dist loadfile把用例聚起来能减少重复搭建。另外要记住默认的 CPython 里如果测试本身是 CPU 密集的进程并行是真有收益的各进程各有一把 GIL而如果测试主要是在等 I/O进程并行的收益来自「多个 I/O 同时等」但用线程也能达到类似效果且开销更小。三、多线程执行用标准库调度pytest 没有内置的线程并行。一个不依赖任何插件的可行做法是把用例集切成几份用subprocess各起一个 pytest 进程再用ThreadPoolExecutor并发调度这些子进程。线程在这里的角色不是「让用例并行计算」而是「同时盯住多个子进程、等它们各自结束」——恰好是线程擅长的 I/O 等待。# 适用于 Python 3.8文件保存为 run_parallel.pyimport concurrent.futuresimport subprocessimport sysimport time# 把用例文件名或 node id 前缀切成几组GROUPS [[tests/test_a.py],[tests/test_b.py],[tests/test_c.py],[tests/test_d.py],]def run_pytest(paths):在一个子进程里跑 pytest返回 (退出码, 输出)。cmd [sys.executable, -m, pytest, -q, *paths]proc subprocess.run(cmd, capture_outputTrue, textTrue)return proc.returncode, proc.stdoutdef main():start time.perf_counter()failed 0with concurrent.futures.ThreadPoolExecutor(max_workers4) as pool:futures {pool.submit(run_pytest, g): g for g in GROUPS}for fut in concurrent.futures.as_completed(futures):group futures[fut]code, out fut.result()status 通过 if code 0 else f失败(exit{code})print(f[{group[0]}] {status})print(out)if code ! 0:failed 1print(f总耗时 {time.perf_counter() - start:.2f} 秒失败组数 {failed})return 1 if failed else 0if __name__ __main__:sys.exit(main())这段代码只用标准库可以直接运行前提是先把tests/目录和上面的pytest.ini准备好。它把「并行」交给了操作系统调度多个 pytest 进程而线程池负责管理这四路的等待与结果汇总。为什么不用线程直接调用 pytest 的 API因为 pytest 的运行过程大量依赖全局状态插件注册表、配置对象、模块缓存在同一进程里并发调用多个 pytest 会话并不可靠。用子进程隔离是最稳妥的方式这也是pytest-xdist同样选择子进程的原因。四、并行必须处理的三件事事项说明对策用例独立性并行下执行顺序不确定用例之间不共享可变全局状态资源竞争多个 worker 抢同一资源每 worker 用独立临时目录/数据库/端口输出与报告多进程输出会交错用-q或让每个进程写自己的文件第三点尤其容易被忽略。多进程并行时如果每个进程都往同一个日志文件写内容会互相穿插。要么让每个进程写到独立文件要么把汇总交给主进程。常见坑点1. 以为pytest -n auto是开线程❌ 说「xdist 用多线程跑用例」。 ✅ 它是多进程每个 worker 是一个独立子进程。2. 用例之间共享全局状态还开并行❌ 一个用例改模块级变量另一个用例读它单独跑过、一起跑挂。 ✅ 用例保持独立共享状态用 fixture 提供并在每个用例里重置。3. 多个 worker 抢同一个端口或数据库❌ 所有 worker 都连同一个测试库并改同一张表。 ✅ 每个 worker 用独立 schema/临时目录或加锁串行化那部分用例。4. 用-n却忘了装插件❌ 直接pytest -n 4报unrecognized arguments。 ✅ 先pip install pytest-xdist。5. 用例耗时差异极大还用默认load❌ 一个 worker 抽到全部慢用例其他早早空转。 ✅ 用--dist worksteal让空闲 worker 去「偷」任务。6. 在同一进程里并发调用 pytest 会话❌ 用线程同时pytest.main()多次插件状态互相污染。 ✅ 用subprocess起独立进程或用pytest-xdist。7. 用-n auto却发现更慢了❌ fixture 很重每次建库进程越多总时间越长。 ✅ 用--dist loadscope/loadfile聚合或干脆减少 worker 数。8. 并行的输出糊成一团❌ 多个进程往同一日志文件写内容交错。 ✅ 每进程写独立结果文件由主进程汇总或统一用-q只打印失败摘要。总结需求方案关键点多进程并行pytest-xdistpytest -n auto进程级worker 不共享内存控制分发--dist loadscope/loadfile/worksteal按 fixture 粒度选多线程并发调度subprocessThreadPoolExecutor只用标准库进程隔离最稳关闭并行-n 0临时调试用前提用例必须独立否则出现随机失败选型很简单测试互相独立、fixture 不重就用pytest-xdist进程并行想自己控制分组和汇总就用子进程加线程池。无论哪种先花时间确认用例之间真的没有隐藏依赖——并行的收益上限是由测试套件的「独立性」决定的而不是由 worker 数量决定的。
返回列表