pytest、xdist与共享生成的文件依赖

Question

我有多个测试需要生成一个很耗时的文件。每次测试运行时，我希望这个文件能重新生成，但每次测试只生成一次。更复杂的是，这些测试和文件都依赖于一个输入参数。

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" 的夹具根本不行，因为参数 input 是在函数范围内的。

Answer 1

在pytest-xdist的文档中，有一个现成的解决方案，具体内容在“让会话范围的固定装置只执行一次”这一部分。

import json

import pytest
from filelock import FileLock


@pytest.fixture(scope="session")
def session_data(tmp_path_factory, worker_id):
    if worker_id == "master":
        # not executing in with multiple workers, just produce the data and let
        # pytest's fixture caching do its job
        return produce_expensive_data()

    # get the temp directory shared by all workers
    root_tmp_dir = tmp_path_factory.getbasetemp().parent

    fn = root_tmp_dir / "data.json"
    with FileLock(str(fn) + ".lock"):
        if fn.is_file():
            data = json.loads(fn.read_text())
        else:
            data = produce_expensive_data()
            fn.write_text(json.dumps(data))
    return data

需要注意的是，filelock并不是标准库的一部分，但可以从PyPI上获取。你可以在这里找到文档。