<@ULVA73B9P> what’s the best way to unit test pref...
# ask-marvin
a
@Marvin what’s the best way to unit test prefect flows without invoking prefect? I’d like to be able to call the entire flow function as there is some logic in there, but then I’d also need to add .fn on every task to avoid needing prefect to run my test
m
thought for 190.1 seconds
Great question. Short answer: there isn’t a built-in switch to make tasks called inside a
flow.fn()
automatically bypass the engine. In Prefect 3.x, calling
flow.fn()
will execute your flow’s Python function body, but any
task()
calls inside will still try to go through the engine unless you call the tasks’
.fn
as well. Recommended approaches: 1) Best practice: separate pure logic from orchestration - Keep your business logic in plain functions, and make your tasks/flow thin wrappers. Then unit test the pure functions directly.
Copy code
# my_module.py
from prefect import flow, task, get_run_logger

def _compute(x):
    return x * 2

@task
def compute_task(x):
    logger = get_run_logger()
    y = _compute(x)
    <http://logger.info|logger.info>(f"computed {y}")
    return y

@flow
def my_flow(a):
    return compute_task(a)

# test_my_module.py
from my_module import _compute

def test_compute_unit():
    assert _compute(3) == 6
2) If you want to run the flow function “as-is” without Prefect, monkeypatch tasks to their .fn in tests - This lets you call
flow.fn(...)
and have inner tasks also execute their raw functions.
Copy code
# test_my_module.py
import my_module as m
from prefect.logging import disable_run_logger

def test_flow_logic_without_prefect(monkeypatch):
    # unwrap the tasks the flow uses
    monkeypatch.setattr(m, "compute_task", m.compute_task.fn)

    # disable run logger if your code calls get_run_logger()
    with disable_run_logger():
        result = m.my_flow.fn(3)  # calls the flow body; inner task is unwrapped
    assert result == 6
- If you have multiple tasks, unwrap each:
Copy code
for name in ("task1", "task2", "task3"):
    monkeypatch.setattr(m, name, getattr(m, name).fn)
Notes/caveats when using .fn: - Any call to
get_run_logger()
, task/flow context, retries, caching, etc. will not work (or will need to be disabled) because the engine is not running. Use
with disable_run_logger():
if you reference
get_run_logger()
anywhere. - There’s no global setting to auto-coerce all task calls to
.fn
. 3) For integration tests, use the ephemeral test harness - If you do want to run through the engine (states, logging, retries) but keep things lightweight/isolated:
Copy code
from prefect.testing.utilities import prefect_test_harness

def test_flow_with_engine():
    with prefect_test_harness():
        assert my_flow(3) == 6
Docs for reference: - Testing workflows: Testing workflows - Testing utilities API: prefect.testing.utilities If you’d like, I can share a small pytest fixture that automatically unwraps a module’s known tasks to
.fn
so you don’t have to repeat monkeypatch lines in every test.
a
@Marvin is there any community discussion on this topic? maybe in github issues or in this slack?
m
thought for 83.5 seconds
Short answer: I couldn’t find a dedicated GitHub Discussion on “unit testing flows without the engine” or “using .fn” in Prefect. I did find one relevant GitHub issue (from 2.x) about
get_run_logger()
failing in unit tests, which is the same root cause you’ll hit when calling task functions directly: - GitHub Issues - prefect.get_run_logger() raises MissingContextError when testing task functions (2.x, but the pattern applies: use
.fn
or run with a test harness; disable the run logger if you don’t want the engine) - GitHub Discussions - I searched Discussions and didn’t find a thread focused on this topic. - Slack (last 90 days in our index) - I didn’t find a focused thread on unit testing flows without the engine or using
.fn
. Our Slack index only covers the last ~90 days, so there could be older posts we can’t surface. Official docs (for completeness): - Testing workflows - prefect.testing.utilities (includes
prefect_test_harness
and
disable_run_logger
) If you’d like, I can: - Start a GitHub Discussion summarizing the patterns (split pure logic,
flow.fn()
, monkeypatch tasks to
.fn
,
disable_run_logger
, or use
prefect_test_harness
for integration), or - Post a Slack thread recap and invite community input. Want me to kick off a Discussion and link it here?