Pytest-xdist: tearDown after all workers finish

Question

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.

Answer 1

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')