title: How to mock PythonOperator's python_callable and on_failure_callback?
body: <p>What and where do I mock, with an Airflow <code>PythonOperator</code>,  such that:</p>
  
  <ul>
  <li>the <code>python_callback</code> raises a exception, triggering the call of the <code>on_failure_callback</code>, and </li>
  <li>I can test whether that callback is called, and with what arguments?</li>
  </ul>
  
  <p>I have tried mocking both the {python_callable} and <code>PythonOperator.execute</code>, in a number of places, without success.</p>
  
  <p>The code files look something like this:</p>
  
  <p><strong>dags/my_code.py</strong></p>
  
  <pre><code>class CustomException(Exception): pass
  
  def a_callable():
      if OurSqlAlchemyTable.count() == 0:
          raise CustomException("{} is empty".format(OurSqlAlchemyTable.name))
      return True
  
  def a_failure_callable(context):
      SlackWebhookHook(
          http_conn_id=slack_conn_id,
          message= context['exception'].msg,
          channel='#alert-channel'
          ).execute()
  </code></pre>
  
  <p><strong>dags/a_dag.py</strong></p>
  
  <pre><code>from my_code import a_callable, a_failure_callable
  
  new_task = PythonOperator(
      task_id='new-task', dag=dag-named-sue, conn_id='a_conn_id', timeout=30,
      python_callable=a_callable,
      on_failure_callback=a_failure_callable)
  </code></pre>
  
  <p><strong>dags/test_a_dag.py</strong></p>
  
  <pre><code>class TestCallback(unittest.TestCase):
  
      def test_on_failure_callback(self):
          tested_task = DagBag().get_dag('dag-named-sue').get_task('new-task')
  
          with patch('airflow.operators.python_operator.PythonOperator.execute') as mock_execute:
              with patch('dags.a_dag.a_failure_callable') as mock_callback:
                  mock_execute.side_effect = CustomException
                  tested_task.execute(context={})
  
              # does failure of the python_callable trigger the failure callback?
          mock_callback.assert_called()
  
              # did the exception message make it to the failure callback?
          failure_context = mock_callback.call_args[0]
          self.assertEqual(failure_context['exception'].msg,
                          'OurSqlAlchemyTable is empty')O
  </code></pre>
  
  <p>The test does raise <code>CustomException</code> at the line <code>self.task.execute(context={})</code> -- but, in the test code itself.  What I want is for that error to be
  raised in the Airflow code such that the <code>PythonOperator</code> fails and calls <code>on_failure_callback</code>.</p>
  
  <p>I have tried any number of permutations, all either raising in test without
  triggering, calling the python_callable, or not finding an object to patch:</p>
  
  <pre><code>patch('dags.a_dag.a_callable') as mock_callable
        'a_dag.a_callable'
        'dags.my_code.a_callable'
        'my_code.a_callable'
        'airflow.models.Task.execute'
  </code></pre>
  
  <p>(<code>Python3</code>, <code>pytest</code>, and <code>mock</code>.)</p>
  
  <p><strong>What am I missing / doing wrong?</strong></p>
  
  <p>(Better still, I would like to verify the arguments passed to <code>SlackWebhookHook</code>.  Something like:</p>
  
  <pre><code>with patch('???.SlackWebhookHook.execute') as mock_webhook:
      ... as above ...
  
  kw_dict = mock_webhook.call_args[-1]
  assert kw_dict['http_conn_id'] == slack_conn_id
  assert kw_dict['message'] == 'OurSqlAlchemyTable is empty'
  assert kw_dict['channel'] == '#alert-channel'
  </code></pre>
  
  <p>(But I am first focussing on testing the failure callback.)</p>
  
  <p>Thank you in advance.</p>