Jonah Duffin
09/12/2025, 4:47 PMprefect_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. How can I do so?Marvin
09/12/2025, 4:47 PMNate
09/12/2025, 4:51 PMNate
09/12/2025, 4:52 PMMarvin
09/12/2025, 4:53 PMprefect_test_harness across pytest-xdist workers. With xdist, a “session-scoped” fixture is per worker process, not global, and prefect_test_harness is not multi-process safe. Each worker will try to spin up its own ephemeral server.
Here are the practical options:
1) Run one shared Prefect server for all workers (recommended if you can relax isolation)
- Start a single server outside pytest (locally or in CI) and point all tests at it via PREFECT_API_URL.
- Example:
- Start once (in another terminal or a CI pre-step):
prefect server start --host 127.0.0.1 --port 4200 --no-ui
- Run tests with a shared API URL:
PREFECT_API_URL=<http://127.0.0.1:4200/api> pytest -n auto
- In your tests, replace the harness with a simple “use hosted server” fixture so everything connects to that URL:
import os
import pytest
from prefect.testing.fixtures import use_hosted_api_server
@pytest.fixture(scope="session", autouse=True)
def use_shared_prefect_server():
api_url = os.environ.get("PREFECT_API_URL", "<http://127.0.0.1:4200/api>")
with use_hosted_api_server(api_url):
yield
- Caveat: this reduces isolation. If your tests assume a blank DB, reset it before the run:
prefect server database reset -y
- Docs/API refs:
- Server concepts
- Testing fixtures API (see use_hosted_api_server)
- Verified CLI:
- prefect server start --help
- prefect server database reset --help
2) Keep per-worker isolation but make it reliable
- Keep your existing fixture but increase startup time and/or reduce worker count so multiple ephemeral servers don’t thrash your machine:
@pytest.fixture(autouse=True, scope="session")
def prefect_test_fixture():
from prefect.testing.utilities import prefect_test_harness
with prefect_test_harness(server_startup_timeout=90):
yield
- Then run with fewer workers (e.g., pytest -n 2) or split tests so only Prefect-integration tests run serially.
3) Split the test run
- Mark Prefect-integration tests and run them without xdist (-n 1) while everything else runs in parallel.
Why not “one harness for all workers”?
- xdist runs separate Python processes. A session-scoped fixture runs once per worker, and the process that created the harness will also tear it down when that worker finishes, potentially while others are still running. Cross-process sharing and teardown coordination isn’t supported by prefect_test_harness.
If you can share a bit about your CI environment and how much isolation you need, I can suggest a concrete setup (e.g., pre-starting the server in CI, waiting for health, and resetting the DB before tests).Jonah Duffin
09/12/2025, 5:01 PMNate
09/12/2025, 5:02 PMBring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.
Powered by