ensure on_failure hook runs upon parameter validation error (#15109)

PrefectHQ · Aug 27, 2024 · a2320de · a2320de
1 parent f555104
commit a2320de
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 0 deletions.
diff --git a/src/prefect/flow_engine.py b/src/prefect/flow_engine.py
@@ -205,6 +205,7 @@ def begin_run(self) -> State:
                     result_factory=run_coro_as_sync(ResultFactory.from_flow(self.flow)),
                 )
                 self.short_circuit = True
+                self.call_hooks()
 
         new_state = Running()
         state = self.set_state(new_state)

diff --git a/tests/test_flows.py b/tests/test_flows.py
@@ -3183,6 +3183,20 @@ def my_flow():
         assert state.type == StateType.FAILED
         assert my_mock.call_args_list == [call(), call()]
 
+    def test_on_failure_hooks_run_on_bad_parameters(self):
+        my_mock = MagicMock()
+
+        def failure_hook(flow, flow_run, state):
+            my_mock("failure_hook")
+
+        @flow(on_failure=[failure_hook])
+        def my_flow(x: int):
+            pass
+
+        state = my_flow(x="x", return_state=True)
+        assert state.type == StateType.FAILED
+        assert my_mock.call_args_list == [call("failure_hook")]
+
 
 class TestFlowHooksOnCancellation:
     def test_noniterable_hook_raises(self):