第 14 章 · 并行与分布式执行
本章目标：用 pytest-xdist 把测试套件铺到多个 CPU 上，理解
--dist四种分发策略的适用场景，避开 fixture 隔离与共享资源的并行陷阱，并让 xdist 与覆盖率统计和平共处。
14.1 五分钟上手：-n auto
当套件规模增长到"跑一轮要喝杯咖啡"时，就该上 pytest-xdist 了。它把测试分发给多个进程（worker）并行执行：
pip install pytest-xdist
pytest -n auto # 按物理 CPU 核数启动 worker
pytest -n 8 # 显式指定 worker 数
pytest -n logical # 用逻辑核数（需 Python>=3.13 或 psutil）
pytest -n 0 # 关闭并行，回到单进程-n auto 的 worker 数还可以通过环境变量 PYTEST_XDIST_AUTO_NUM_WORKERS 或在 conftest 里实现 pytest_xdist_auto_num_workers(config) hook 来定制；另有 --maxprocesses=N 给 auto 模式设置上限：
# content of conftest.py —— 按运行环境自定义 auto worker 数
import os
def pytest_xdist_auto_num_workers(config):
if "CI" in os.environ: # CI 机器核少但测试重，固定 4 个
return 4
return None # 返回 None 则回退到默认行为# 本地最多开 4 个 worker，避免开发机卡死
pytest -n auto --maxprocesses=414.2 分发策略：--dist 的四种选择
默认策略 load 是"哪个 worker 空了就塞给它下一个"，不保证顺序。但有些测试必须在同一进程内连续执行——比如依赖昂贵的模块级 fixture、或有顺序耦合——这时就要换策略：
pytest -n 4 --dist load # 默认：自由负载均衡
pytest -n 4 --dist loadscope # 函数按模块分组、方法按类分组，同组同 worker
pytest -n 4 --dist loadfile # 同一文件的所有测试固定在一个 worker
pytest -n 4 --dist loadgroup # 按 xdist_group 标记分组
pytest -n 4 --dist worksteal # 负载窃取：忙闲不均时自动转移任务loadgroup 配合 xdist_group 标记可以手工圈定"必须同生共死"的用例：
# content of test_cluster.py
import pytest
@pytest.mark.xdist_group(name="group1")
def test_create_resource():
create_shared_resource() # 与 test_use_resource 保证在同一 worker
class TestA:
@pytest.mark.xdist_group("group1") # 类上的标记同样生效
def test_use_resource(self):
assert shared_resource_ready()选型经验
- 默认
load适合绝大多数纯单元测试； - 有昂贵的 module/class 级 fixture →
loadscope； - 测试间时长差异巨大（混合快慢测）→
worksteal； - 存在跨测试共享状态的存量代码 →
loadfile/loadgroup过渡，最终目标是消除状态耦合。
14.3 并行陷阱清单
并行不是免费午餐。xdist 官方文档列出的限制之外，工程上最常见的翻车点：
共享外部资源：两个 worker 同时连同一个数据库 schema / 写同一个文件 / 抢同一个端口。
python# ❌ 所有 worker 都往固定路径写，必然互踩 LOG = open("/tmp/app.log", "w") # ✅ 用 tmp_path 隔离（每个 worker 进程独立目录） def test_write(tmp_path): target = tmp_path / "app.log" write_log(target) assert target.exists()session 级 fixture 每 worker 执行一次：
scope="session"的 fixture 在每个 worker 进程里各跑一遍，不是全局唯一。需要真正全局唯一的资源（如一次性初始化的数据库）要用文件锁或主进程预创建。python# content of conftest.py —— 用文件锁保证数据库只被初始化一次 import filelock import pytest @pytest.fixture(scope="session") def database(tmp_path_factory): # rootdir 下的一次性锁文件，所有 worker 抢同一把锁 lock = tmp_path_factory.getbasetemp().parent / "db.lock" with filelock.FileLock(str(lock)): db = create_test_database() # 锁内初始化，其余 worker 等待后复用 yield db顺序假设：依赖执行顺序或模块级可变全局状态的测试，在
load策略下会随机炸。修测试而不是调策略。输出与报告交错：失败摘要仍会汇总，但 print 输出归属哪个 worker 不直观；定位问题时可加
-v看 worker 前缀（如gw0、gw1）。
# 排查并行-only 失败的标准三板斧
pytest -n 4 # 并行跑，复现
pytest -n 0 # 串行跑，确认是否为并行特有
pytest -n 4 -k "test_x" # 缩小范围定位冲突对14.4 xdist + coverage：并行下也要准确
多进程各自测量再简单相加是错的（同一行可能被不同 worker 各执行一部分）。pytest-cov 从设计上就兼容 xdist：worker 结束时会自动把数据回传合并，你不需要任何额外操作：
pytest -n auto --cov=src --cov-report=term-missing唯一的硬性要求是 所有 worker 都必须装了 pytest-cov（插件要通过 entry-point 在每个 worker 注册）。若还要测量被测程序拉起的子进程，按第 13 章提示配置 coverage 的 subprocess patch。
别忘了 CI 里的 CPU 数
CI 机器常见只有 2–4 核，-n auto 收益有限甚至因进程开销变慢。可以在 CI 中显式 -n 4，并把并行收益最大的集成测试放到独立的 job 里跑。
14.5 本章小结
-n auto/N一行命令开启多进程并行；--maxprocesses控制 auto 上限；--dist四策略：load（默认）、loadscope/loadfile（分组保序）、loadgroup（xdist_group 手工分组）、worksteal（负载窃取）；- 三大陷阱：共享资源要隔离（tmp_path）、session fixture 是每 worker 一份、禁止顺序与全局状态假设；
- pytest-cov 自动合并多 worker 覆盖率，前提是 worker 环境都装有该插件。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. -n auto 启动的 worker 数量由什么决定？
2. 一组测试依赖昂贵的模块级 fixture，希望它们始终在同一个 worker 内执行，最合适的策略是？
3. 关于 scope="session" 的 fixture 在 xdist 下的行为，正确的是？
4. pytest-cov 与 pytest-xdist 一起使用时，覆盖率的正确说法是？
🛠️ 动手实践
- 构造一个包含 20 个
time.sleep(0.1)参数化用例的套件，分别以-n 0和-n auto计时对比，记录加速比。 - 制造一个并行冲突：两个测试都向同一固定路径写文件且断言内容，观察
-n auto下的偶发失败，然后用tmp_path修复。 - 用
xdist_group把"先建后删"的两个测试圈进同名 group，分别在--dist loadgroup和默认load下运行多次，验证前者不再出现删除先于创建的竞态。
下一章进入参数化的深水区：indirect、笛卡尔积堆叠与自定义 ID 体系：第 15 章。