'Test the execute function in case where the run failed.'

@mock.patch('airflow.contrib.operators.databricks_operator.DatabricksHook')
def test_exec_failure(self, db_mock_class):

    run = {'new_cluster': NEW_CLUSTER, 'notebook_task': NOTEBOOK_TASK}
    op = DatabricksSubmitRunOperator(task_id=TASK_ID, json=run)
    db_mock = db_mock_class.return_value
    db_mock.submit_run.return_value = 1
    db_mock.get_run_state.return_value = RunState('TERMINATED', 'FAILED', '')
    with self.assertRaises(AirflowException):
        op.execute(None)
    expected = op._deep_string_coerce({'new_cluster': NEW_CLUSTER, 'notebook_task': NOTEBOOK_TASK, 'run_name': TASK_ID})
    db_mock_class.assert_called_once_with(DEFAULT_CONN_ID, retry_limit=op.databricks_retry_limit)
    db_mock.submit_run.assert_called_once_with(expected)
    db_mock.get_run_page_url.assert_called_once_with(RUN_ID)
    db_mock.get_run_state.assert_called_once_with(RUN_ID)
    self.assertEquals(RUN_ID, op.run_id)
