Great question! Assuming Prefect 3.x, the best way to run pytest integration tests for a flow is to execute it through the Prefect engine against an isolated, ephemeral API/database using

prefect_test_harness

. This gives you true end-to-end behavior (orchestration, state, retries, logging) without touching your real environment. Recommended setup - Use a session-scoped pytest fixture to bring up the ephemeral API once for the test session (fast and stable). - Call your flow normally to assert results, or with

return_state=True

to assert on states, exceptions, and retries. Minimal example

# conftest.py
import pytest
from prefect.testing.utilities import prefect_test_harness

@pytest.fixture(autouse=True, scope="session")
def prefect_api():
    # Increase timeout in slow CI if needed: prefect_test_harness(server_startup_timeout=60)
    with prefect_test_harness():
        yield

# test_flow_integration.py
from prefect import flow

@flow
def my_flow(x: int) -> int:
    return x + 1

def test_flow_runs_end_to_end():
    assert my_flow(1) == 2

def test_flow_state_and_result():
    state = my_flow(1, return_state=True)
    assert state.is_completed()
    assert state.result() == 2

Asserting persisted orchestration data (optional)

import pytest
from prefect.client.orchestration import get_client

@pytest.mark.asyncio
async def test_flow_run_persisted():
    # Run the flow through the engine
    my_flow(123)

    # Query the ephemeral API for the flow run
    async with get_client() as client:
        runs = await client.read_flow_runs(limit=1)
        assert len(runs) >= 1

Capturing logs

# add the fixture to your test to wire Prefect loggers into caplog
def test_logs(caplog, add_prefect_loggers_to_caplog):
    caplog.set_level("INFO")
    my_flow(1)
    assert "Flow run" in caplog.text  # or match a specific message you log

Notes and tips - Use session scope for speed; switch to function scope only for tests that require a fresh DB per test. - For unit tests (not integration), prefer `task.fn()`/`flow.fn()` to run your Python functions without the engine. - If the ephemeral server is slow to start in CI/containers, bump

server_startup_timeout

in

prefect_test_harness

. - To test retries/timeouts/failures, call your flow with

return_state=True

and assert on

state.is_*()

and

state.result()

. Helpful docs - How to test workflows (Prefect 3.x) - prefect.testing.utilities (incl. prefect_test_harness) - prefect.testing.fixtures (pytest helpers) - Flow API (return_state) If you can share what you want to assert (e.g., states/retries, logs, DB artifacts, external I/O), I can tailor an example to your specific flow.