<@ULVA73B9P> I am using the `prefect_test_harness`...
# ask-marvin
j
@Marvin I am using the
prefect_test_harness
for my unit tests like so:
Copy code
@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?
m
thought for 321.9 seconds
n
you'll probably have to mess with the db url based on the pytest workers https://github.com/PrefectHQ/prefect/blob/main/tests/conftest.py
feel free to hoist this question to a discussion so we can assemble a complete answer there and potentially port that to docs
m
Short answer: you can’t share a single
prefect_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):
Copy code
prefect server start --host 127.0.0.1 --port 4200 --no-ui
- Run tests with a shared API URL:
Copy code
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:
Copy code
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:
Copy code
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:
Copy code
@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).
j
n
🙇