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.

Answer 1

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