Pete Fein
02/22/2020, 3:42 PMJeff Brainerd
02/22/2020, 4:06 PMflow.run()
. Then we examine the resulting state of the flow run to verify tasks succeeded or skipped appropriately. We also implemented a dry run mode that avoids running the guts of long running tasks.Chris White
02/22/2020, 11:52 PMtask.run()
directly with the appropriate kwargs
- you can unit test your task trigger / state change logic by creating a TaskRunner
and calling run
with any collection of upstream state objects (https://docs.prefect.io/api/latest/engine/task_runner.html)
- you can unit test your full flow logic by providing a (possibly incomplete) dictionary of {task: state}
pairs and providing it to flow.run(task_states=mocked_states)
and inspecting the outputs as @Jeff Brainerd mentions above
- you can unit test your schedules by testing your flow’s schedule
class directly
If there is another type of testing that you had in mind let me know and I can tell you how we would address it; this is also a good reminder that we should write a tutorial on testing flows!Stanley Yeh
04/11/2020, 7:44 AM@task
def boring_func(a):
return a + 5
def test_boring_func():
assert 5 == boring_func(0) # would not work. "ValueError: Could not infer an active Flow context"
assert 5 == boring_func.run(0) # this would work
Wish this is mentioned somewhere in the docs.