Hi! I'm looking for good design pattern for unit testing tasks. I've seen the examples that compose and test flows but it makes more sense to me to start with a test for each of my tasks instead.

Hi @Matt! One of the truly great things about Prefect tasks is that they're just normal python functions, meaning you can unit test them as you would any other python application. As such, you can directly feed any inputs to your task as you would expect at runtime. Some resources I would recommend would be the testing documentation and Prefect's own suite of tests, which run against Prefect itself

Would love some help from the community! And/or some direction to good examples on how to test Prefect tasks/flows.

Chiming in here — all tasks have a

.run()

method that routes inputs directly to the task as a python function. E.g.

@task
def my_task(x, y):
    return x + y

assert my_task.run(1, 2) == 3

if you reference things like

prefect.context

within your task, you might have to set up some scaffolding:

with prefect.context(my_special_key=value):
    my_task.run(1, 2)

but that’s on a case-by-case basis

Thanks @Chris White. This approach definitely works as a way to call tasks with an input. It's somewhat complicated in that my tasks don't all return something (though maybe I need to think that through). It seems to me the downside is that on the assertion side I can't then do a simple is_successful() to see if the task worked or not. I'd need to figure out another way to see if the task worked. That right?

Yea so then you would need to return to your original pattern of using the task runner. In that case you might look at this test file to see what setup is and is not required for that: https://github.com/PrefectHQ/prefect/blob/master/tests/engine/test_task_runner.py In particular, if you test specifically the runner’s “get task run state” method it should have the minimal interface you’re looking for: https://github.com/PrefectHQ/prefect/blob/master/tests/engine/test_task_runner.py#L987

This is perfect! Thanks @Chris White! Will look at this more but was able to use this approach to do my test. The test_inputs on line 1068 was what I needed.

Good to know - in the meantime, I’ll archive this thread to GitHub in case anyone else has similar questions, hopefully it can help them out too

This is definitely useful. What's a good pattern to import the tasks from the file that contains the flow in your tests?

Standard python import rules apply depending on the structure of your project and where you run your tests; you can also get tasks from your Flow via the

flow.get_tasks

method