使用pytest-xdist实现测试夹具锁，避免数据下载写入冲突

要避免多进程并行写入缓存文件的冲突，核心思路是让仅一个进程完成数据下载与缓存，其他进程等待缓存就绪后直接读取，下面是具体实现方案：

方法1：使用文件锁控制单进程写入 #

借助filelock库实现文件级别的互斥锁，确保同一时间只有一个进程执行下载和写入操作，其他进程等待锁释放后直接使用已缓存的文件。

步骤： #

1. 安装依赖库：

pip install filelock

2. 在conftest.py中编写session级夹具：

import requests
from filelock import FileLock
import pytest

@pytest.fixture(scope="session")
def cached_test_data(tmp_path_factory, worker_id):
    # 共享缓存目录：所有worker进程都能访问的临时目录
    root_temp_dir = tmp_path_factory.getbasetemp().parent
    cache_dir = root_temp_dir / "test_data_cache"
    data_file = cache_dir / "dataset.json"
    lock_file = root_temp_dir / "cache_lock.lock"

    # 非并行模式直接下载
    if worker_id == "master":
        cache_dir.mkdir(exist_ok=True)
        response = requests.get("https://your-test-data-url.com/data")
        data_file.write_text(response.text)
        return data_file

    # 并行模式下，通过文件锁控制单进程写入
    with FileLock(lock_file):
        if not data_file.exists():
            # 仅当文件未缓存时才执行下载
            cache_dir.mkdir(exist_ok=True)
            response = requests.get("https://your-test-data-url.com/data")
            data_file.write_text(response.text)
    
    return data_file

关键细节： #

• worker_id：pytest-xdist为每个进程分配的标识，主进程为master，工作进程为gw0、gw1等
• 共享临时目录：用tmp_path_factory.getbasetemp().parent获取所有worker共享的临时根目录，避免每个worker使用独立临时目录导致无法共享缓存
• 文件锁：FileLock会自动处理锁的获取与释放，确保同一时间只有一个进程进入写入逻辑

方法2：利用xdist主进程提前完成缓存 #

如果你的测试允许主进程先完成所有初始化操作，再启动工作进程执行测试，可以通过pytest_configure钩子在主进程中提前下载缓存：

import requests
import os
import pytest

def pytest_configure(config):
    # 仅在主进程执行初始化
    if hasattr(config, "workerinput"):
        return
    
    # 下载并缓存数据到共享目录
    cache_dir = os.path.join(config.rootdir, ".test_cache")
    os.makedirs(cache_dir, exist_ok=True)
    data_file = os.path.join(cache_dir, "dataset.json")
    
    if not os.path.exists(data_file):
        response = requests.get("https://your-test-data-url.com/data")
        with open(data_file, "w") as f:
            f.write(response.text)

@pytest.fixture(scope="session")
def cached_test_data(config):
    cache_dir = os.path.join(config.rootdir, ".test_cache")
    return os.path.join(cache_dir, "dataset.json")

这种方法的优势是完全避免锁的使用，主进程在工作进程启动前就完成缓存，所有工作进程直接读取即可。

内容的提问来源于stack exchange，提问作者Mikko Ohtamaa