Great question — yes, there are a few solid patterns for running real, end-to-end Prefect flows locally in automated tests. Below are the ones we recommend for Prefect 3.x, ranging from “fast and local” to “full deployment + worker” so you can catch bugs that only show up in real runs. High-level options - Fast local runs with an ephemeral API (good for most integration tests) - Full E2E with a hosted test API, a work pool, a worker, and a deployment - CLI-driven worker in tests (optional; useful if you want to exercise the CLI path) Pattern A: Fast local runs with ephemeral API (prefect_test_harness) - What you get: a temporary DB and ephemeral API server; you run the flow in-process but still interact with the API if needed. - Use when: you want speed and isolation but don’t need the worker/deployment lifecycle.

from prefect import flow, task
from prefect.testing.utilities import prefect_test_harness

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

@flow
def my_flow(x: int = 2):
    return double(x)

def test_flow_in_process_with_ephemeral_api():
    with prefect_test_harness():
        result = my_flow(21)
        assert result == 42

Pattern B: True E2E with hosted test API + work pool + worker + deployment - What you get: an actual API server, a work pool, a worker that submits/executes runs, and real state transitions you can assert on. - Use when: you want to simulate your production path locally. Key fixtures/utilities you can rely on: - prefect.testing.fixtures - use_hosted_api_server: spins up a local API server for the test session - prefect_client: async client connected to that server - prefect.testing.utilities - prefect_test_harness: if you prefer ephemeral per-test API/DB instead of a single hosted server - Programmatic worker pattern: use ProcessWorker or BaseWorker to submit runs once then exit (fast and deterministic) Example (async pytest style): ``` import asyncio import pytest from datetime import timedelta from prefect import flow, task from prefect.client.orchestration import PrefectClient from prefect.client.schemas.actions import WorkPoolCreate from prefect.states import Scheduled from prefect.types._datetime import now as now_utc from prefect.workers.process import ProcessWorker @task def double(x): return x * 2 @flow def my_flow(x: int = 2): return double(x) async def wait_for_terminal_state(client: PrefectClient, flow_run_id, timeout=60): deadline = asyncio.get_event_loop().time() + timeout while True: fr = await client.read_flow_run(flow_run_id) if fr.state and fr.state.is_final(): return fr.state if asyncio.get_event_loop().time() > deadline: raise TimeoutError("Flow run did not reach a terminal state in time.") await asyncio.sleep(0.5) @pytest.mark.usefixtures("use_hosted_api_server") @pytest.mark.asyncio async def test_e2e_deployment_with_process_worker(prefect_client: PrefectClient): # 1) Create a process work pool pool = await prefect_client.create_work_pool( WorkPoolCreate(name="e2e-pool", type="process") ) # 2) Create a deployment for this flow (no image build/push) # In Prefect 3.x, you typically deploy from source. # For local tests, point at the current file/module. deployment_id = await my_flow.from_source( source=".", # repo or local path entrypoint="tests/test_e2e.py:my_flow", # adjust to your filepathfunction ).deploy( name="e2e-deployment", work_pool_name=pool.name, push=False, build=False, print_next_steps=False, ) # 3) Create a scheduled run so a worker will pick it up immediately flow_run = await prefect_client.create_flow_run_from_deployment( deployment_id, parameters={"x": 21}, state=Scheduled(scheduled_time=now_utc("UTC") - timedelta(seconds=1)), ) # 4) Submit/execute the run via a ProcessWorker (single cycle)

# Using get_and_submit_flow_runs() is a fast, deterministic test pattern. async with ProcessWorker(work_pool_name=pool.name, name="e2e-test-worker") as worker: await worker.get_and_submit_flow_runs() # 5) Wait for terminal state and assert final_state = await wait_for_terminal_state(prefect_client, flow_run.id, timeout=60) assert final_state.is_completed(), f"Flow failed: {final_state}"

Notes
- from_source(...).deploy(...) is the 3.x way to deploy in tests (avoid deprecated 2.x APIs like Deployment.build_from_flow).
- Using ProcessWorker.get_and_submit_flow_runs() lets your test do a single deterministic polling/submission cycle rather than running a long-lived worker loop.
- The wait_for_terminal_state helper polls the API until the run ends (Completed/Failed/Cancelled/etc.).

Pattern C: CLI-driven worker in tests (optional)
- If you prefer exercising the CLI code path, you can start a worker in a subprocess for a single cycle, then exit. This is slower but can surface CLI-specific issues.

Example outline:

import subprocess import sys import time from pathlib import Path # ...create work pool + deployment + scheduled run as shown above... # Start a one-shot worker; prefer bounded runs in CI proc = subprocess.Popen( [sys.executable, "-m", "prefect", "worker", "start", "--pool", pool.name], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True ) # Optionally tail logs for a short duration or until the run completes time.sleep(5) proc.terminate() proc.wait() # Then poll the run state and assert terminal completion as above ``` Additional helpful fixtures/utilities you can use - prefect.testing.fixtures.use_hosted_api_server: starts a test API server and sets PREFECT_API_URL for you. - prefect.testing.utilities.get_most_recent_flow_run: handy for grabbing the last run when asserting. - prefect.testing.cli.invoke_and_assert: use this if you want to assert CLI output/exit codes. A few best practices - Keep worker execution bounded in tests. The programmatic worker “single cycle” approach is ideal; if you use the CLI, make sure you stop it promptly. - Skip image builds and pushes during tests (

build=False

,

push=False

) and prefer the process worker locally for speed. - Schedule runs in the past to ensure immediate pickup by the worker (Scheduled with scheduled_time in the past). - Poll for terminal states and set reasonably short timeouts to keep your test suite fast and reliable. Questions for you - Which worker type do you want to exercise locally (process, docker, kubernetes)? - Are you using pytest? If so, I can share drop-in fixtures for creating a work pool, starting a one-shot worker, deploying the flow, and waiting for completion tailored to your repo layout.

Awesome — here are drop-in pytest fixtures and tests for a true E2E run with a local process worker in Prefect 3.x. Project layout - tests/flows/e2e_flow.py — your flow - tests/conftest.py — shared fixtures (register Prefect’s testing fixtures + your E2E fixtures) - tests/test_e2e_process_worker.py — the E2E test tests/flows/e2e_flow.py

from prefect import flow, task

@task
def double(x: int) -> int:
    return x * 2

@flow
def my_flow(x: int = 2) -> int:
    return double(x)

tests/conftest.py

import uuid
import asyncio
import pytest
from datetime import datetime, timezone
from prefect.client.orchestration import PrefectClient
from prefect.client.schemas.actions import WorkPoolCreate
from prefect.flows import Flow

# Register Prefect's built-in test fixtures (e.g., use_hosted_api_server, prefect_client)
pytest_plugins = ["prefect.testing.fixtures"]

@pytest.fixture
async def process_work_pool(prefect_client: PrefectClient):
    """
    Creates an isolated process work pool for this test session.
    """
    name = f"e2e-process-pool-{uuid.uuid4().hex[:8]}"
    pool = await prefect_client.create_work_pool(
        WorkPoolCreate(name=name, type="process")
    )
    return pool

@pytest.fixture
async def deployed_my_flow(process_work_pool):
    """
    Deploys the test flow to the process work pool.
    Adjust `entrypoint` to match your repo path.
    """
    deployment_name = f"e2e-deployment-{uuid.uuid4().hex[:8]}"
    deployment_id = Flow.from_source(
        source=".",  # repo root or local path
        entrypoint="tests/flows/e2e_flow.py:my_flow",  # adjust to your path
    ).deploy(
        name=deployment_name,
        work_pool_name=process_work_pool.name,
        build=False,          # speed: no image build
        push=False,           # speed: no push
        print_next_steps=False,
    )
    return deployment_id

async def wait_for_terminal_state(prefect_client: PrefectClient, flow_run_id, timeout: float = 60.0):
    """
    Polls the flow run until a terminal state (Completed/Failed/Cancelled) is reached.
    """
    loop = asyncio.get_event_loop()
    deadline = loop.time() + timeout
    while True:
        fr = await prefect_client.read_flow_run(flow_run_id)
        if fr.state and fr.state.is_final():
            return fr.state
        if loop.time() > deadline:
            raise TimeoutError(f"Flow run {flow_run_id} did not reach a terminal state in {timeout}s.")
        await asyncio.sleep(0.5)

@pytest.fixture
def utcnow():
    return lambda: datetime.now(timezone.utc)

tests/test_e2e_process_worker.py

import pytest
from datetime import timedelta
from prefect.client.orchestration import PrefectClient
from prefect.states import Scheduled
from prefect.workers.process import ProcessWorker

# Use Prefect's hosted test API server for true E2E behavior
pytestmark = pytest.mark.usefixtures("use_hosted_api_server")

@pytest.mark.asyncio
async def test_e2e_process_worker_runs_flow(
    prefect_client: PrefectClient,
    process_work_pool,
    deployed_my_flow,
    utcnow,
):
    # 1) Create a scheduled run in the past so the worker picks it up immediately
    flow_run = await prefect_client.create_flow_run_from_deployment(
        deployed_my_flow,
        parameters={"x": 21},
        state=Scheduled(scheduled_time=utcnow() - timedelta(seconds=1)),
    )

    # 2) Run a one-shot worker cycle to submit/execute the run
    async with ProcessWorker(
        work_pool_name=process_work_pool.name,
        name="e2e-test-worker",
    ) as worker:
        # A single deterministic cycle; quick and CI-friendly
        await worker.get_and_submit_flow_runs()

    # 3) Wait for terminal state and assert success
    #    wait_for_terminal_state is provided by conftest.py
    from .conftest import wait_for_terminal_state
    final_state = await wait_for_terminal_state(prefect_client, flow_run.id, timeout=60)
    assert final_state.is_completed(), f"Flow did not complete: {final_state}"

Notes

- This assumes Prefect 3.x. - Adjust entrypoint in Flow.from_source to match your repo path to the flow function. - The worker pattern uses a single get_and_submit_flow_runs() call for deterministic tests instead of running a long-lived worker loop. - We schedule the run “in the past” so it’s immediately eligible and gets picked up right away. - If you want to assert on the flow’s return data, configure result storage/serialization for your flow; otherwise, asserting Completed is typically enough for E2E coverage. If you’d like, I can adapt this to use a shared work pool across multiple tests, or show a variant with prefect_test_harness for faster non-worker runs.