Ensure a fixture runs once when using pytest-xdist and other workers wait for it to finish

test_fixture_run_once_xdist.py

"""
$ pip install posix_ipc pytest pytest-xdist

$ pytest -s -v -n 4 test_fixture_run_once_xdist.py
"""

import inspect
import logging
import time


def fixture_auto_run_once(func):
    func_name = func.__name__
    if inspect.isgeneratorfunction(func):
        wrapped = func
    else:
        def wrapped():
            yield func()

    def wrapper(worker_id, testrun_uid):
        if worker_id == "master":
            yield from wrapped()
        else:
            name = f"{func_name}-{testrun_uid}"
            try:
                logging.info("Trying exclusive semaphore %s", name)
                with Semaphore(name, flags=O_CREX, initial_value=1) as sem: 
                    yield from wrapped()
            except ExistentialError:
                time.sleep(2)
                with Semaphore(name):
                    yield

    return pytest.fixture(wrapper, scope="session", autouse=True)
  
  
@fixture_auto_run_once
def example_fixture():
    logging.warn('mygen started!')
    yield 55
    logging.warn('mygen finished!')
    
    
def test_a():
    pass

def test_b():
    pass

def test_c():
    pass

def test_d():
    pass