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.
This plugin makes it possible to run tests quickly using multiprocessing (parallelism) and multithreading (concurrency).
You can pass arguments to fixtures with the params keyword argument in the fixture decorator, and you can also pass arguments to tests with the @pytest. mark. parametrize decorator for individual tests.
A fixture can use multiple other fixtures. Just like a test method can take multiple fixtures as arguments, a fixture can take multiple other fixtures as arguments and use them to create the fixture value that it returns.
What Are Pytest Fixtures? Pytest fixtures are functions that can be used to manage our apps states and dependencies. Most importantly, they can provide data for testing and a wide range of value types when explicitly called by our testing software. You can use the mock data that fixtures create across multiple tests.
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
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With