https://prefect.io logo
Title
l

Lukáš Pravda

04/26/2022, 4:19 PM
Hello community, I’d like to test a rather simple scenario whether a task defined in a flow was run, but I dont seem to understand how to mock it properly. In simplicity my setup is similar to: file.py
@task
def foo():
    ....

@task
def bar():
    ....

with Flow() as flow:
   try:
       foo()
   except:
       bar()
test.py
from file import flow
from unittest.mock import patch

@patch("file1.bar")
@patch("file.foo")
def test_flow(mock1,  mock2):
    mock.side_effect = Exception("throw an error")

   flow.run()

   assert mock2.assert_called_once()
but the mock is never called, have found this:
<https://github.com/PrefectHQ/prefect/issues/1801>
, but could not really mount that solution to my exact problem. What am I missing? Thank you
k

Kevin Kho

04/26/2022, 4:25 PM
I think the first issue is using
try
except
inside the Flow. Are you really doing that? Because I don’t think the
except
will be reached since the task is not actually running. The
try
here will just be about adding
foo()
to the DAG. If you want to do this find of thing, it would be in the form of triggers where you set the trigger of
bar()
to
always_run
or
on_failure
l

Lukáš Pravda

04/26/2022, 10:13 PM
Thanks! that makes sense.
so I’ve changed the flow logic to use on_failure and by modifying the code I can see it works, but I’m not sure how to write unittest for that. Is there any way of mocking the flow failure?