pytest fixtures and threads synchronizations

Question

I'm trying to use pytest-xdist in order to make my tests run parallel, The issue is that each thread is going to the fixture that shared to all tests and executing it according to threads number.

It cause me an issue because that fixture role is to create data for my tests and once it's already created I get and error since it's already created(via REST).

conftest.py:

lock = threading.Lock()

@pytest.fixture(scope=session)
def init_data(request):

    lock.acquire()
    try:
        data = []
        if checking_data_not_created():
            data.append(some_rest_command_creating_data_1())
            data.append(some_rest_command_creating_data_2())
    finally:
        lock.release()

    yield data

    lock.acquire()
    try:
        remove_all_data()
    finally:
        lock.release()

tests_class.py:

class classTests():

    def first_test(init_data):
        test body and validation....

     def second_test(init_data):
        test body and validation....

I am using the command: pytest -v -n2

Assuming that 1st thread should run first_test() and 2nd thread should run second_test() one of them will always fail because the first already created the data in fixtures section and the other thread will get exception and all tests he should run will fail.

As you can see, I tried to use lock in order to synchronize the threads but it also doesn't work.

any idea how can I overcome this issue?

Thanks.

exception that related to my server, not an environment, that's not the point. The point is that I want to run the fixture with the scope=session only once — Gababi, Mar 04 '19 at 19:48
pytest-xdist uses multiple processes, not threads! I don't think `threading.lock` will be sufficient to prevent concurrent access. — Samuel Dion-Girardeau, Mar 10 '19 at 00:50
I understood it in the hard way, Thanks for your response Samuel — Gababi, Mar 10 '19 at 13:58
I've done similar things via making a custom pytest plugin. That can use a hook like `pytest_configure` to run once per process, and before _any_ process has started running tests. Then check `os.environ.get("PYTEST_XDIST_WORKER")` is falsy to make sure its the configure of the "outer" pytest and not any of the individual works. Not sure its a really good solution though... — afaulconbridge, Sep 07 '22 at 16:01

score 4 · Answer 1 · answered May 23 '19 at 11:36

This approach will not work with pytest-xdist as this uses multiprocessing not multithreading however it can be used with pytest-parallel using the --tests-per-worker option it will run the tests using multiple threads.

The data will only be setup once and cleaned up once in a multithreaded pytest execution with the following fixture:

conftest.py:

lock = threading.Lock()
threaded_count = 0

@pytest.fixture(scope='session')
def init_data():
    global lock
    global threaded_count

    lock.acquire()
    threaded_count += 1
    try:
        if threaded_count == 1:
            # Setup Data Once
            data = setup_data()
    finally:
        lock.release()

    yield data

    lock.acquire()
    threaded_count -= 1
    try:
        if threaded_count == 0:
            # Cleanup Data Once
            data = cleaup_data()
    finally:
        lock.release()

Command:

pytest -v --tests-per-worker 2

*"This approach will not work with pytest-xdist as this uses multiprocessing not multithreading [...] it will run the tests using multiple threads."* If it's using `multiprocessing` instead of `multithreading`, then the tests should run in multiple processes, not "multiple threads". — code_dredd, Aug 27 '20 at 20:06

score 3 · Answer 2 · answered Jul 26 '19 at 14:52

Note that pytest-xdist does not honor 'session' scoped fixtures. A 'session' scoped fixture in pytest-xdist actually means a process-specific session-level fixture, so it gets created and torn down separately for each process and the state of the fixture is not shared between processes. There are some long-standing proposals to add locking and sharing of true session-scoped fixtures to pytest-xdist, but they all run into many of the classic reasons to try at all costs to avoid multi-threading or multi-processing with locks and synchronization, and as a result it is heavily deprioritized by pytest-xdist devs (and understandably so).

score 2 · Answer 3 · answered Mar 16 '21 at 15:43

Python xdist has a way to do this.

I believe in your example, applying their suggestion would look like this:

def create_data():
    data = []
    data.append(some_rest_command_creating_data_1())
    data.append(some_rest_command_creating_data_2())

    return data


@pytest.fixture(scope="session")
def session_data(request, tmp_path_factory, worker_id):
    if worker_id == "master":
        # Not running multiple processes, just create the data.
        data = create_data()
    else:
        # Running multiple processes, manage lockfile.
        root_tmp_dir = tmp_path_factory.getbasetemp().parent

        fn = root_tmp_dir / "data.json"
        with FileLock(str(fn) + ".lock"):
            if fn.is_file():
                # Data has been created, read it in.
                data = json.loads(fn.read_text())
            else:
                # Data not created, create and write it out.
                data = create_data()
                fn.write_text(json.dumps(data))

    yield data

    remove_all_data()

Unlike in your example, this example is providing the data through a lockfile saved at the shared temp test directory at tmp_path_factory.getbasetemp().parent, which is provided with pytest.

Documentation link is now https://pytest-xdist.readthedocs.io/en/latest/how-to.html#making-session-scoped-fixtures-execute-only-once — afaulconbridge, Sep 07 '22 at 15:58

pytest fixtures and threads synchronizations

3 Answers3