Great question. In Prefect 3.x, the simplest way to simulate a task failure in tests is to monkeypatch the task’s underlying function (

task.fn

). You can either make it raise an exception (engine will create a Failed state), or return a

Failed

state directly. Here are both patterns. - Make the task fail by raising an exception (includes traceback/exception details in the state):

from prefect import flow, task

@task
def do_work():
    return 42

@flow
def my_flow():
    fut = do_work.submit()
    return fut

def test_flow_handles_failure(monkeypatch):
    def boom(*args, **kwargs):
        raise RuntimeError("boom")

    # Patch the task's underlying callable
    monkeypatch.setattr(do_work, "fn", boom)

    fut = my_flow()          # run the flow
    state = fut.wait()       # get the task state
    assert state.is_failed()
    assert "boom" in (state.message or "") or "RuntimeError" in (state.name or "")

- Make the task fail by returning a Prefect Failed state (no Python exception in the task body):

from prefect import flow, task
from prefect.states import Failed

@task
def do_work():
    return 42

@flow
def my_flow():
    fut = do_work.submit()
    return fut

def test_flow_handles_failed_state(monkeypatch):
    monkeypatch.setattr(
        do_work,
        "fn",
        lambda *a, **k: Failed(message="forced failure for test"),
    )

    fut = my_flow()
    state = fut.wait()
    assert state.is_failed()
    assert state.message == "forced failure for test"

Notes: - If your task is async, patch

do_work.fn

with an async function (either raising or returning

Failed

). - Patch where the task object is imported/used (standard pytest monkeypatch practice). - Returning a

prefect.states.Failed

lets you control the failure message cleanly; raising an exception is closer to real failures (includes exception/traceback). References: - States overview: https://docs.prefect.io/latest/concepts/states/ - Tasks overview: https://docs.prefect.io/latest/concepts/tasks/ If you share a small snippet of your task/flow, I can tailor the patch to your exact setup (including async/caching/retries).