Great question. Here’s a practical, repeatable approach to testing complex Prefect flows with pytest (assuming Prefect 3.x): High-level approach (test pyramid) - Unit: test your task/flow functions directly with

.fn()

(fast, no engine/DB). Mock external systems here. - Integration: run the real Prefect engine + orchestration against an isolated ephemeral DB using

prefect_test_harness()

and assert on states, events, and persisted metadata. - End-to-end: register and run a deployment in the harness and validate orchestration behavior (scheduling, parameters, triggers, etc.). Optionally hit real infra in nightly runs. Unit tests (fastest) - Use

.fn()

to call the underlying function of tasks/flows. - If your code uses

get_run_logger()

, wrap tests in

disable_run_logger()

to avoid needing a run context. Example

from prefect import flow, task
from prefect.logging import disable_run_logger

@task
def add(x, y): 
    return x + y

@flow
def pipeline(a, b, c):
    return add(a, b) + c

def test_task_unit():
    assert add.fn(2, 3) == 5

def test_flow_unit():
    with disable_run_logger():
        assert pipeline.fn(1, 2, 3) == 6

Integration tests (engine + DB, still fast) - Use

prefect_test_harness()

to spin up an isolated SQLite-backed orchestration environment. This avoids mocking Prefect internals. - Assert on states with

return_state=True

, query runs with the Prefect client, and fast-forward time with

mock_anyio_sleep

. Example

import pytest
from prefect import flow, task
from prefect.testing.utilities import prefect_test_harness
from prefect.states import StateType
from prefect.client.orchestration import get_client

@task
def work(x):
    return x * 2

@flow(name="example-flow")
def do_work(x):
    return work(x)

def test_integration_state_and_db():
    with prefect_test_harness():
        state = do_work(return_state=True, x=21)
        assert state.type == StateType.COMPLETED
        assert state.result() == 42

        # optional: check persisted run metadata
        async def _check():
            async with get_client() as client:
                runs = await client.read_flow_runs(limit=1)
                assert runs and runs[0].name
        import anyio
        anyio.run(_check)

Mocking time (no real sleeps)

import pytest
import anyio
from prefect import task, flow

@task
async def delayed():
    await anyio.sleep(5)
    return "ok"

@flow
async def f():
    return await delayed()

@pytest.mark.asyncio
async def test_no_real_sleep(mock_anyio_sleep):
    assert await f() == "ok"

End-to-end deployment tests (orchestration) - Register a deployment in the harness using

flow.to_deployment(...).deploy()

. - Run it with

run_deployment(...)

and assert on results/state. Example

import pytest
from prefect import flow
from prefect.testing.utilities import prefect_test_harness
from prefect.deployments import run_deployment

@flow(name="my-flow")
def my_flow(x: int) -> int:
    return x + 1

@pytest.mark.asyncio
async def test_deployment_e2e():
    with prefect_test_harness():
        # Register deployment in the test DB
        my_flow.to_deployment(name="test-deploy", parameters={"x": 0}).deploy()

        # Run by slug "flow-name/deployment-name" or by UUID
        fr = await run_deployment("my-flow/test-deploy", parameters={"x": 41}, timeout=30, poll_interval=0.1)
        assert fr.state.is_completed()

What to mock vs not mock - Do mock: - External services (HTTP, DBs, message queues). Use

unittest.mock

,

respx

(for httpx), or your favorite library. - Time/sleeps with

mock_anyio_sleep

. - Randomness/clock if your logic depends on it. - Don’t mock: - Prefect’s flow/task engine, context, or API for integration tests—use

prefect_test_harness()

instead. - Flow/task decorators—use

.fn()

for unit tests and real engine for integration tests. - Optional infra mocking:

- For unit tests of deployment wiring, you can mock infra blocks (e.g., Docker/K8s objects) to validate configuration without requiring the provider to be installed or running. Best practices - Keep most tests as unit tests using `.fn()`; add focused integration tests via `prefect_test_harness()`; reserve a small number of E2E tests for deployment/infra paths. - Test states where relevant with

return_state=True

. - If you use logging in functions, wrap tests in

disable_run_logger()

unless you’re exercising the engine. - Keep external calls behind thin task wrappers to make mocking trivial. - Use

pytest.mark.asyncio

for async flows/tasks and prefer async-aware mocks (e.g.,

AsyncMock

) where needed. - Prefer realistic scheduling/trigger tests in the harness over trying to mock Prefect internals. - CI tip: run unit and integration tests on every push; run E2E or provider-backed tests nightly or behind a tag. Handy fixtures/utilities you can use -

prefect_test_harness()

for an isolated orchestration DB -

mock_anyio_sleep

to skip real waits - Event capture:

asserting_events_worker

,

events_pipeline

- Block comparisons:

assert_blocks_equal

- Ensure no warnings:

assert_does_not_warn

References and examples - Test workflows guide: Test workflows (Prefect 3.x) - Testing utilities API: prefect.testing.utilities - Testing fixtures API: prefect.testing.fixtures - Flows concept: Flows in Prefect 3.x - Deployments concept: Deployments in Prefect 3.x Would you like me to tailor examples to your stack? If you’re using specific providers (e.g., S3, Snowflake, Docker, Kubernetes) or Prefect Cloud vs. local, I can show concrete patterns for mocking those dependencies and/or running realistic E2E tests.