@Marvin I'm aware that is possible to run unit test on tasks using the .fn call. But if this call depends on the context, like for create create_artifacts the test will fail I need to mocj that call. Is there a way to provide a run time context for this task? How its the recommended way to achieve this without mock?

In Prefect, it's possible to provide a runtime context for a task when you need to test it, especially for those tasks that have context dependencies. Accessing and mocking the runtime context can be achieved using

prefect.context.get_run_context()

or by explicitly accessing the flow run or task run context using

FlowRunContext.get()

or

TaskRunContext.get()

. Here's what you can do to provide a runtime context for your tasks: 1. Use

get_run_context()

within your task to access the context if it is available. 2. Use

FlowRunContext.get()

or

TaskRunContext.get()

to access specific contexts; these will return

None

rather than raising an error if the context is not available. Please note that these methods are designed for interaction within a running flow or task, so you may still need to use a testing framework like pytest to mock or simulate your runtime context during testing. You can find more information on the Prefect documentation page about Runtime Context and look at the Prefect codebase on GitHub in the test_tasks.py for examples of how runtime context is handled in tests.

@Marvin But how can I create a new context?

To explicitly create and manipulate a new runtime context within testing, you may have to use context managers or the tools provided by the testing frameworks such as pytest. Prefect's documentation does not explicitly describe creating a new runtime context, suggesting that the typical use case is to rely on the context provided during task and flow runtime. Since there isn't direct documentation on creating a new context manually, here's a general approach you can take using standard Python context management techniques:

python
from prefect.context import TaskRunContext

# Make a function that simulates task runtime context
def create_test_task_run_context(**kwargs):
    defaults = { ... default context values ... }
    defaults.update(kwargs)
    return TaskRunContext(**defaults)

# Usage in a test
def test_task():
    # Create a mock context for the task run
    task_run_ctx = create_test_task_run_context(my_key='my_value')
    
    # Use the context manager to set the context temporarily
    with prefect.context(task_run_ctx):
        # Your testing code here
        result = my_task.run()

    # Assertions or other test validations here

Remember, the above is a theoretical example; the actual implementation might vary based on the latest Prefect version and the context you need. To get more detailed and specific guidance tailored to your situation, I'd recommend looking into Prefect's user community or GitHub issues for examples of how other developers have handled the creation of a testing context, or perhaps asking a question there. For now, if you need detailed steps or examples tailored to Prefect 2.x, outside of the provided guidance, it could be worth reaching out to the Prefect community for more specific assistance.

@William Jamir i've run into this as well and it blocks me hard from using tasks as much as i'd like, do report back if Marvin has provided something useful here!

@William Jamir are you already familiar with

prefect_test_harness

? it will give you a mock db to run against for your tests so you can just run your flows / tasks against it without using

.fn

Sorry @Nate I think I didn't get how the proposed solution will help with the prefect context thing, when running tests for a task Let for example considering this example:

def test_my_task():

    @task
    def my_task():
        context = prefect.context.get_run_context()
        print(context.task_run.flow_run_id)
        create_markdown_artifact(key="X", markdown="ABC")
        return "HI"

    assert my_task() == "HI"

This will fail first because my_task() returns a TaskRun with the scheduled state. By using

prefect_test_harness

we are providing an in memory db, right? But how it will help with the context? I mean, if I call it like this:

with prefect_test_harness():
        assert my_task.fn() == "HI"

I still dont have any context available

>       raise MissingContextError(
            "No run context available. You are not in a flow or task run context."
        )
E       prefect.exceptions.MissingContextError: No run context available. You are not in a flow or task run context.

im saying that if you use prefect_task_harness

you can just run your flows / tasks against it without using

.fn

and then you have your context