3

Suppose I have the below test cases written in a file, 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

When I run the above using below pytest command then they work without any issues.

pytest -s --disable-warnings test_something.py

However, I want to run these test cases in a parallel manner. I know that test_1 and test_2 should run parallelly. So I looked into pytest-parallel and did the below:

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

But as shown above in the code, when it goes to create the /a/b/bin folder, it throws an error saying that the directory already exists. So this means that the module-scope is not being honoured in pytest-parallel. It is trying to execute the get_some_binary_file for every parameterized input to test_1 Is there a way for me to do this?

I have also looked into pytest-xdist with the --dist loadscope option, and ran the below command for it:

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

But this gave me an output like below, where both test_1 and test_2 are being executed on the same worker.

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

As can be seen from above output, the test_2 is running in gw1. Why? Shouldn't it run in a different worker?

user1452759
  • 8,810
  • 15
  • 42
  • 58
  • As pytest cannot know if a module-scoped fixture has to run for all related processes (for example if it provides an object that lives in memory), or only once (like in your case), the way `pytest-xdist` does it is the only way to do it correctly in my opinion. `pytest-parallel` takes the risky path, that will fail for global resources that cannot be initialized twice - though the fixture could be rewritten to handle the case that it is called more than once. – MrBean Bremen Oct 12 '20 at 17:40
  • But pytest-xdist is running all the test cases on the same gw1 worker. I'm guessing this is because of --dist loadscope. I will work towards rewriting the fixture code with a lockfile. – user1452759 Oct 16 '20 at 14:12
  • Not sure about `pytest-xdist`, but making the fixture reentrant is probably the best option. – MrBean Bremen Oct 16 '20 at 17:47
  • 1
    Yeah I thought so. I have re-written how my fixture is written using the example from https://pypi.org/project/qieman-xdist/ about how session fixtures are handled. And that helped. – user1452759 Oct 20 '20 at 16:13

1 Answers1

0
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
Chetana
  • 1
  • Your answer could be improved with additional supporting information. Please [edit] to add further details, such as citations or documentation, so that others can confirm that your answer is correct. You can find more information on how to write good answers [in the help center](/help/how-to-answer). – Community May 25 '22 at 11:53
  • This does not provide an answer to the question. Once you have sufficient [reputation](https://stackoverflow.com/help/whats-reputation) you will be able to [comment on any post](https://stackoverflow.com/help/privileges/comment); instead, [provide answers that don't require clarification from the asker](https://meta.stackexchange.com/questions/214173/why-do-i-need-50-reputation-to-comment-what-can-i-do-instead). - [From Review](/review/late-answers/31854594) – MD. RAKIB HASAN May 31 '22 at 10:33