pytest并行未遵守模块范围固定装置

Question

假设我在一个文件test_something.py:中编写了以下测试用例

@pytest.fixture(scope="module")
def get_some_binary_file():
# Some logic here that creates a path "/a/b/bin" and then downloads a binary into this path
os.mkdir("/a/b/bin") ### This line throws the error in pytest-parallel
some_binary = os.path.join("/a/b/bin", "binary_file")
download_bin("some_bin_url", some_binary)
return some_binary

test_input = [
{"some": "value"},
{"foo": "bar"}
]

@pytest.mark.parametrize("test_input", test_input, ids=["Test_1", "Test_2"])
def test_1(get_some_binary_file, test_input):
# Testing logic here

# Some other completely different tests below
def test_2():
# Some other testing logic here

当我使用下面的pytest命令运行上述命令时，它们可以正常工作。

pytest -s --disable-warnings test_something.py

但是，我希望以并行的方式运行这些测试用例。我知道test_1和test_2应该并行运行。所以我研究了pytest并行，并做了以下操作：

pytest --workers auto -s --disable-warnings test_something.py. 

但是，如上面的代码所示，当它创建/a/b/bin文件夹时，会抛出一个错误，指出目录已经存在。因此，这意味着在pytest并行中没有遵守模块范围。它正在尝试为test_1的每个参数化输入执行get_some_binary_file。我有办法做到这一点吗？

我还研究了带有--dist loadscope选项的pytest-xlist，并为其运行了以下命令：

pytest -n auto --dist loadscope -s --disable-warnings test_something.py

但这给了我一个如下的输出，其中test_1和test_2都在同一个工作线程上执行。

tests/test_something.py::test_1[Test_1]
[gw1] PASSED tests/test_something.py::test_1[Test_1] ## Expected
tests/test_something.py::test_1[Test_2]
[gw1] PASSED tests/test_something.py::test_1[Test_2] ## Expected
tests/test_something.py::test_2
[gw1] PASSED tests/test_something.py::test_2 ## Not expected to run in gw1

从上面的输出可以看出，test_2正在gw1中运行。为什么？它不应该在另一个工人身上运行吗？

Answer 1

Group the definitions with xdist_group to run per process. Run like this to assign it to per process,  pytest xdistloadscope.py -n 2 --dist=loadgroup
@pytest.mark.xdist_group("group1")
@pytest.fixture(scope="module")
def get_some_binary_file():
# Some logic here that creates a path "/a/b/bin" and then downloads a binary into this path
os.mkdir("/a/b/bin") ### This line throws the error in pytest-parallel
some_binary = os.path.join("/a/b/bin", "binary_file")
download_bin("some_bin_url", some_binary)
return some_binary

test_input = [
{"some": "value"},
{"foo": "bar"}
]

@pytest.mark.xdist_group("group1")
@pytest.mark.parametrize("test_input", test_input, ids=["Test_1", "Test_2"])
def test_1(get_some_binary_file, test_input):
# Testing logic here

# Some other completely different tests below
@pytest.mark.xdist_group("group2")
def test_2():
# Some other testing logic here