When I'm running pytest normally it works perfectly fine (before first test it's upgrading migration and after last test it's downgrading migration). Without using xdist I can just set fixture scope='session' and its work that way. But when I run pytest-xdist (tests in parallel) it's upgrading migration before each test and downgrading migration after each test (I want to setUp before first test - first worker, and tearDown after last test - when last worker finishes).
I think that setUp before first worker I can handle with File Lock (and also I don't need it that much because in migration upgrade I'm just creating some tables if not exist so if it was already upgraded it just won't do anything).
I'm interested especially in tearDown after last worker finshes.
import pytest
import os
import subprocess
@pytest.fixture(autouse=True, scope='session')
def handle_database():
"""Create necesarry tables in db (using migration) and drop them after tests"""
# Change directory to test/api_simulator
os.chdir('test/api_simulator')
# Perform migration upgrade
subprocess.run(['flask', 'db', 'upgrade'])
print('\n\nUpgrade migration\n\n')
# Change directory back to the original directory
os.chdir('../..')
yield
# Change directory to test/api_simulator
os.chdir('test/api_simulator')
# Perform migration downgrade
subprocess.run(['flask', 'db', 'downgrade'])
print('\n\nDowngrade migration\n\n')
# Change directory back to the original directory
os.chdir('../..')
I was searching about this issue in web but I found nothing good about tearDown with xdist.
The fixture
is running as part of a test, even if it's only once in session scope. This means it's running in the process xdist
opened.
To run this only once you move the code to pytest_sessionstart
and pytest_sessionfinish
in conftest.py
. Those methods will be executed once in
"regular" scope before/after xdist
kicks in. Note that the pytest_session
functions will run again as part of the tests cycle so you need to check it
def __handle_database(config, command):
if not hasattr(config, 'workerinput'):
os.chdir('test/api_simulator')
# Perform migration upgrade
subprocess.run(['flask', 'db', command])
print('\n\nUpgrade migration\n\n')
# Change directory back to the original directory
os.chdir('../..')
def pytest_sessionstart(session):
__handle_database(session.config, 'upgrade')
def pytest_sessionfinish(session, exitstatus):
__handle_database(session.config, 'downgrade')