如何在pytest xdist并行测试中避免重复生成带参数的昂贵文件？

我有多个测试需使用生成成本极高的文件，要求该文件在每次测试运行时重新生成，但每轮运行最多生成一次。更复杂的是，这些测试与该文件均依赖一个输入参数，代码示例如下：

def expensive(param) -> Path:
    # Generate file and return its path.

@mark.parametrize('input', TEST_DATA)
class TestClass:

    def test_one(self, input) -> None:
        check_expensive1(expensive(input))

    def test_two(self, input) -> None:
        check_expensive2(expensive(input))

请问如何确保在多线程并行运行这些测试时，该文件不会被重复生成？背景是我正在将基于Makefile的测试基础设施迁移至pytest。我可以接受使用基于文件的锁来同步，但更倾向于采用已有成熟解决方案。目前functools.cache在单线程环境下表现良好，但scope="module"的fixture完全无法适用，因为input参数为函数作用域级别。

解决方案 #

方法一：函数级fixture + 线程安全缓存 #

利用pytest的函数级fixture结合线程安全的缓存机制，既满足每个input参数只生成一次文件，又支持多线程并行：

import threading
from pathlib import Path
import pytest

# 全局缓存存储已生成的文件路径，搭配锁保证线程安全
_expensive_cache = {}
_cache_lock = threading.Lock()

def expensive(param) -> Path:
    # 原有的文件生成逻辑
    file_path = Path(f"./temp_{param}.txt")
    if not file_path.exists():
        # 模拟高成本生成过程
        with open(file_path, "w") as f:
            f.write(f"Data for param: {param}")
    return file_path

@pytest.fixture(scope="function")
def expensive_file(request):
    input_param = request.param
    with _cache_lock:
        if input_param not in _expensive_cache:
            _expensive_cache[input_param] = expensive(input_param)
    return _expensive_cache[input_param]

@pytest.mark.parametrize('expensive_file', TEST_DATA, indirect=True)
class TestClass:
    def test_one(self, expensive_file) -> None:
        check_expensive1(expensive_file)

    def test_two(self, expensive_file) -> None:
        check_expensive2(expensive_file)

• 核心逻辑：通过全局字典_expensive_cache记录每个input对应的文件路径，threading.Lock确保多线程下只有一个线程执行文件生成操作；
• 利用indirect=True让pytest将参数传递给fixture，实现每个input参数对应一次文件生成；
• 函数级fixture保证每轮测试运行时都会重新生成缓存，测试结束后可按需清理缓存和文件。

如果使用Python 3.9+，也可以直接用functools.lru_cache（functools.cache是其简化版），它本身是线程安全的，只需在测试会话开始前清空缓存保证每轮重新生成：

from functools import cache
import pytest

@cache
def expensive(param) -> Path:
    # 原生成逻辑

@pytest.fixture(autouse=True, scope="session")
def clear_cache_before_session():
    expensive.cache_clear()

@pytest.mark.parametrize('input', TEST_DATA)
class TestClass:
    def test_one(self, input) -> None:
        check_expensive1(expensive(input))

    def test_two(self, input) -> None:
        check_expensive2(expensive(input))

• autouse=True的会话级fixture会在测试开始前清空缓存，保证每轮测试重新生成文件；
• 此方案更简洁，但要求param是可哈希类型，若参数为不可哈希对象，需改用字典+锁的方式。

方法二：基于文件锁的成熟实现 #

如果需要跨线程/跨进程的同步支持，可使用filelock库（pytest生态常用的成熟工具）：

from pathlib import Path
from filelock import FileLock
import pytest

def expensive(param) -> Path:
    file_path = Path(f"./temp_{param}.txt")
    lock_path = Path(f"./temp_{param}.lock")
    # 文件锁保证同一param下只有一个进程/线程生成文件
    with FileLock(lock_path):
        if not file_path.exists():
            # 高成本生成逻辑
            with open(file_path, "w") as f:
                f.write(f"Data for param: {param}")
    return file_path

@pytest.mark.parametrize('input', TEST_DATA)
class TestClass:
    def test_one(self, input) -> None:
        check_expensive1(expensive(input))

    def test_two(self, input) -> None:
        check_expensive2(expensive(input))

• filelock支持跨线程、跨进程同步，适配pytest-xdist多进程运行场景；
• 锁文件会自动管理，也可在测试后按需删除锁文件和生成的目标文件。

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