Using the prefect_test_harness across multiple processes via pytest-xdist

[jonahduffin]

I am using the prefect_test_harness for my unit tests like so:

@pytest.fixture(autouse=True, scope="session")
def prefect_test_fixture():
    with prefect_test_harness():
        yield

However, I also run pytest using the pytest-xdist extension to create multiple processes and run unit tests in parallel. This is resulting in many temporary prefect servers attempting to be spun up at once, and it isn't working well.

I'd like to create a single prefect_test_fixture and share it across multiple threads/processes. Is there a Prefect-recommended way to do so?

Note that pytest-xdist does have some documentation addressing how to make session-scoped fixtures execute only once. But, this won't apply the context of the prefect_test_harness to all of my separate processes, and I need the temporary Prefect settings determined here to apply to all of my processes. My preference is not to pass these settings along via environment variables, because handling environment variables in multi-process unit tests is a headache.

[jonahduffin]

@zzstoatzz has suggested that we'll probably have to mess with the DB url based on the pytest workers:
https://github.com/PrefectHQ/prefect/blob/main/tests/conftest.py

[jonahduffin]

I've tried a lot of things, most recently something like this, which will create a temporary Prefect server if one does not exist yet, and will read in the settings of this server if it does:

@pytest.fixture(autouse=True, scope="session")
def setup_temp_prefect_server(tmp_path_factory):
    from prefect.settings import (
        PREFECT_API_URL,
        PREFECT_API_DATABASE_CONNECTION_URL,
        temporary_settings,
    )
    root_tmp_dir = tmp_path_factory.getbasetemp().parent
    fn = root_tmp_dir / "temp_prefect_server.json"
    lock = FileLock(str(fn) + ".lock")

    # Fast path: reuse existing settings written by another worker
    with lock:
        if fn.is_file():
            data = json.loads(fn.read_text())

    if fn.is_file():
        with temporary_settings(
            updates={
                PREFECT_API_URL: data["PREFECT_API_URL"],
                PREFECT_API_DATABASE_CONNECTION_URL: data["PREFECT_API_DATABASE_CONNECTION_URL"],
            }
        ):
            yield
        return

    # First worker creates the ephemeral server and publishes settings
    with prefect_test_harness():
        data = {
            PREFECT_API_URL.name: PREFECT_API_URL.value(),
            PREFECT_API_DATABASE_CONNECTION_URL.name: PREFECT_API_DATABASE_CONNECTION_URL.value(),
        }
        with lock:
            # It’s fine if another worker wrote this just before us; overwrite or skip
            fn.write_text(json.dumps(data))
        yield

This doesn't work great because once the first worker process that created the temp server finishes, the other processes that still need a temp server start failing.