Haren Samarasinghe
03/26/2026, 12:51 PMMarvin
03/26/2026, 12:51 PMMarvin
03/26/2026, 12:56 PMfrom prefect import flow, task
from prefect.logging import get_run_logger
def test_task_logs(caplog):
@task
def t():
get_run_logger().info("hello from task")
@flow
def f():
t()
f()
assert "hello from task" in caplog.text
2) Create a minimal run context for unit tests (no orchestration)
- If you want to call the task’s function body directly (task.fn()), wrap it in a minimal FlowRunContext to avoid MissingContextError.
from prefect import task
from prefect.context import FlowRunContext
from prefect.logging import get_run_logger
def test_task_fn_with_minimal_context(caplog):
@task
def t():
get_run_logger().info("hello from unit test")
# Minimal context: enough for get_run_logger to work
with FlowRunContext.model_construct(flow=None, flow_run=None):
t.fn() # call the raw function body
assert "hello from unit test" in caplog.text
- You can also pass the context explicitly:
ctx = FlowRunContext.model_construct(flow=None, flow_run=None)
logger = get_run_logger(context=ctx)
<http://logger.info|logger.info>("hello via explicit context")
3) If you don’t care about logging in a test, disable it
- This avoids raising when there’s no run context.
from prefect import task
from prefect.logging import disable_run_logger, get_run_logger
def test_task_no_logging_needed():
@task
def t():
# Will return a disabled logger instead of raising
get_run_logger().info("won't be emitted")
with disable_run_logger():
t.fn()
Notes
- get_run_logger() raises MissingContextError if called outside a flow/task run context. Running the task within a flow or providing a minimal FlowRunContext is the neat fix.
- If you specifically need task-run metadata (IDs, names), the simplest path is to run the task via a flow (option 1). Manually building a full TaskRunContext is possible but generally more work for unit tests.
If you can share a tiny snippet of how you’re testing now, I can suggest the minimal change that removes your monkeypatch.