artur
03/14/2026, 1:31 PMMarvin
03/14/2026, 1:31 PMMarvin
03/14/2026, 1:35 PMflow.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.
# 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.
# 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:
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:
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.artur
03/14/2026, 1:37 PMMarvin
03/14/2026, 1:37 PMMarvin
03/14/2026, 1:39 PMget_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?