本文介绍在 pytest-xdist 多进程并行执行场景下，如何确保参数化测试中昂贵文件仅生成一次（跨 worker 进程），并通过文件锁机制实现线程与进程安全的共享访问。
本文介绍在 pytest-xdist 多进程并行执行场景下，如何确保参数化测试中昂贵文件仅生成一次（跨 worker 进程），并通过文件锁机制实现线程与进程安全的共享访问。
在使用 pytest-xdist 并行运行大量参数化测试时，一个常见痛点是：多个测试需依赖同一份“按输入参数动态生成”的昂贵资源（如预处理数据文件），但该资源既不能复用旧结果（需每次测试运行时重新生成），又不能被重复生成（否则浪费时间、引发竞态或磁盘冲突）。标准的 @pytest.fixture(scope="session") 无法直接满足需求——因为其作用域与 @pytest.mark.parametrize 的函数级参数不兼容；而 functools.cache 仅限单进程内有效，在多 worker 场景下完全失效。
解决方案的核心思路是：将生成逻辑下沉至 session 级 fixture，并利用 xdist 提供的 worker_id 和共享临时目录 + 文件锁，实现跨进程的首次生成与后续读取。以下是推荐实现：
import json
import pytest
from pathlib import Path
from filelock import FileLock
# 假设这是你的实际生成逻辑：根据 param 生成特定文件
def expensive(param: str) -> Path:
# 示例：生成带 param 标识的临时文件
output_path = Path(f"/tmp/expensive_{param}.dat")
# 模拟耗时操作（如模型推理、大文件转换等）
with open(output_path, "w") as f:
f.write(f"Generated for {param} at {__import__('time').time()}")
return output_path
@pytest.fixture(scope="session")
def shared_expensive_file(tmp_path_factory, worker_id, request):
"""
Session-scoped fixture that ensures expensive(param) runs exactly once per unique param,
even across xdist workers.
"""
# 获取当前 param 值（来自 parametrize）
param = request.node.callspec.params.get("input")
if param is None:
raise RuntimeError("Fixture requires 'input' parameter from @parametrize")
# 所有 worker 共享的根临时目录（由 xdist 管理）
root_tmp_dir = tmp_path_factory.getbasetemp().parent
cache_dir = root_tmp_dir / "shared_cache"
cache_dir.mkdir(exist_ok=True)
# 每个 param 对应独立的锁与缓存文件，避免不同参数间干扰
lock_file = cache_dir / f"{param}.lock"
data_file = cache_dir / f"{param}.json"
if worker_id == "master":
# 单进程模式：直接生成并返回
path = expensive(param)
return {"path": str(path), "param": param}
# 多 worker 模式：加锁协调首次生成
with FileLock(str(lock_file)):
if data_file.exists():
# 已存在，直接读取
return json.loads(data_file.read_text())
else:
# 首次生成，写入缓存
path = expensive(param)
data = {"path": str(path), "param": param}
data_file.write_text(json.dumps(data))
return data然后在测试类中使用该 fixture：
import pytest
TEST_DATA = ["A", "B", "C"]
@pytest.mark.parametrize("input", TEST_DATA)
class TestClass:
def test_one(self, input, shared_expensive_file):
assert Path(shared_expensive_file["path"]).exists()
assert shared_expensive_file["param"] == input
# ... your actual check logic
def test_two(self, input, shared_expensive_file):
# 同一 input 下的多个 test 共享同一个生成结果
assert "expensive" in shared_expensive_file["path"]✅ 关键要点说明：
- filelock 是跨平台、跨进程安全的轻量级锁，必须显式安装：pip install filelock；
- 使用 tmp_path_factory.getbasetemp().parent 获取所有 worker 共享的顶层临时目录（xdist 保证）；
- 每个 input 参数对应独立锁和缓存文件，支持多参数并行生成互不干扰；
- request.node.callspec.params 是获取当前参数化值的标准方式，适用于 class 或 function 级 parametrize；
- 若生成逻辑本身含副作用（如修改全局状态），请确保其线程/进程安全性。
⚠️ 注意事项：
- 不要将 shared_expensive_file 的 scope 设为 "function" 或 "class"，否则失去跨 worker 共享意义；
- 缓存目录（如 shared_cache）建议置于 xdist 管理的临时路径下，避免权限或清理问题；
- 生产环境建议增加异常处理（如锁超时、JSON 解析失败重试）和日志，便于调试竞态边界。
通过该方案，你既能保留 Makefile 时代“一次生成、多次复用”的效率优势，又能无缝迁移到现代 pytest-xdist 并行测试架构，兼顾正确性、可维护性与性能。
