fix-non db tests

providers/tests/amazon/aws/operators/test_batch.py

@@ -441,7 +441,7 @@ def test_override_not_sent_if_not_set(self, client_mock, override):
         client_mock().submit_job.assert_called_once_with(**expected_args)
 
     def test_cant_set_old_and_new_override_param(self):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="Invalid arguments were passed"):
             _ = BatchOperator(
                 task_id="task",
                 job_name=JOB_NAME,

providers/tests/google/cloud/operators/test_cloud_build.py

@@ -134,7 +134,7 @@ def test_create_build(self, mock_hook):
     @mock.patch(CLOUD_BUILD_HOOK_PATH)
     def test_create_build_with_missing_build(self, mock_hook):
         mock_hook.return_value.create_build_without_waiting_for_result.return_value = Build()
-        with pytest.raises(AirflowException, match="missing keyword argument 'build'"):
+        with pytest.raises(TypeError, match="missing keyword argument 'build'"):
             CloudBuildCreateBuildOperator(task_id="id")
 
     @pytest.mark.parametrize(
@@ -479,7 +479,7 @@ def test_async_create_build_error_event_should_throw_exception():
 @mock.patch(CLOUD_BUILD_HOOK_PATH)
 def test_async_create_build_with_missing_build_should_throw_exception(mock_hook):
     mock_hook.return_value.create_build.return_value = Build()
-    with pytest.raises(AirflowException, match="missing keyword argument 'build'"):
+    with pytest.raises(TypeError, match="missing keyword argument 'build'"):
         CloudBuildCreateBuildOperator(task_id="id")
 
 

providers/tests/google/cloud/operators/test_compute.py

@@ -349,7 +349,7 @@ def test_insert_instance_from_template_should_throw_ex_when_missing_zone(self):
             )
 
     def test_insert_instance_from_template_should_throw_ex_when_missing_source_instance_template(self):
-        with pytest.raises(AirflowException, match=r"missing keyword argument 'source_instance_template'"):
+        with pytest.raises(TypeError, match=r"missing keyword argument 'source_instance_template'"):
             ComputeEngineInsertInstanceFromTemplateOperator(
                 project_id=GCP_PROJECT_ID,
                 body=GCP_INSTANCE_BODY_FROM_TEMPLATE,
@@ -360,7 +360,7 @@ def test_insert_instance_from_template_should_throw_ex_when_missing_source_insta
             )
 
     def test_insert_instance_from_template_should_throw_ex_when_missing_body(self):
-        with pytest.raises(AirflowException, match=r"missing keyword argument 'body'"):
+        with pytest.raises(TypeError, match=r"missing keyword argument 'body'"):
             ComputeEngineInsertInstanceFromTemplateOperator(
                 project_id=GCP_PROJECT_ID,
                 source_instance_template=SOURCE_INSTANCE_TEMPLATE,
@@ -910,7 +910,7 @@ def test_insert_template_should_not_throw_ex_when_project_id_none(self, mock_hoo
         )
 
     def test_insert_template_should_throw_ex_when_missing_body(self):
-        with pytest.raises(AirflowException, match=r"missing keyword argument 'body'"):
+        with pytest.raises(TypeError, match=r"missing keyword argument 'body'"):
             ComputeEngineInsertInstanceTemplateOperator(
                 task_id=TASK_ID,
                 project_id=GCP_PROJECT_ID,
@@ -1552,7 +1552,7 @@ def test_insert_igm_should_not_throw_ex_when_project_id_none(self, mock_hook):
         )
 
     def test_insert_igm_should_throw_ex_when_missing_body(self):
-        with pytest.raises(AirflowException, match=r"missing keyword argument 'body'"):
+        with pytest.raises(TypeError, match=r"missing keyword argument 'body'"):
             ComputeEngineInsertInstanceGroupManagerOperator(
                 zone=GCE_ZONE,
                 task_id=TASK_ID,

providers/tests/google/cloud/operators/test_dataproc.py

@@ -1577,7 +1577,7 @@ def test_on_kill_after_execution_timeout(self, mock_hook):
         )
 
     def test_missing_region_parameter(self):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="missing keyword argument 'region'"):
             DataprocSubmitJobOperator(
                 task_id=TASK_ID,
                 project_id=GCP_PROJECT,
@@ -1692,7 +1692,7 @@ def test_execute(self, mock_hook):
             )
 
     def test_missing_region_parameter(self):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="missing keyword argument 'region'"):
             DataprocUpdateClusterOperator(
                 task_id=TASK_ID,
                 cluster_name=CLUSTER_NAME,
@@ -2678,7 +2678,7 @@ def test_execute(self, mock_hook):
         )
 
     def test_missing_region_parameter(self):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="missing keyword argument 'region'"):
             DataprocCreateWorkflowTemplateOperator(
                 task_id=TASK_ID,
                 gcp_conn_id=GCP_CONN_ID,

providers/tests/google/cloud/operators/test_kubernetes_engine.py

@@ -227,7 +227,7 @@ def test_create_execute_error_project_id(self, mock_hook):
 
     @mock.patch(GKE_HOOK_PATH)
     def test_create_execute_error_location(self, mock_hook):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="missing keyword argument 'location'"):
             GKECreateClusterOperator(
                 project_id=TEST_GCP_PROJECT_ID, body=PROJECT_BODY, task_id=PROJECT_TASK_ID
             )
@@ -270,14 +270,14 @@ def test_delete_execute_error_project_id(self, mock_hook):
 
     @mock.patch(GKE_HOOK_PATH)
     def test_delete_execute_error_cluster_name(self, mock_hook):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="missing keyword argument 'name'"):
             GKEDeleteClusterOperator(
                 project_id=TEST_GCP_PROJECT_ID, location=PROJECT_LOCATION, task_id=PROJECT_TASK_ID
             )
 
     @mock.patch(GKE_HOOK_PATH)
     def test_delete_execute_error_location(self, mock_hook):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="missing keyword argument 'location'"):
             GKEDeleteClusterOperator(
                 project_id=TEST_GCP_PROJECT_ID, name=CLUSTER_NAME, task_id=PROJECT_TASK_ID
             )
@@ -1270,7 +1270,7 @@ def test_execute(self, mock_hook, fetch_cluster_info_mock, file_mock, exec_mock)
         fetch_cluster_info_mock.assert_called_once()
 
     def test_config_file_throws_error(self):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="missing keyword argument 'queue_name'"):
             GKEStartKueueJobOperator(
                 project_id=TEST_GCP_PROJECT_ID,
                 location=PROJECT_LOCATION,
@@ -1478,7 +1478,7 @@ def setup_method(self):
         )
 
     def test_config_file_throws_error(self):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="Invalid arguments were passed to .*\n.*'config_file'"):
             GKESuspendJobOperator(
                 project_id=TEST_GCP_PROJECT_ID,
                 location=PROJECT_LOCATION,
@@ -1586,7 +1586,7 @@ def setup_method(self):
         )
 
     def test_config_file_throws_error(self):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="Invalid arguments were passed to .*\n.*'config_file'"):
             GKEResumeJobOperator(
                 project_id=TEST_GCP_PROJECT_ID,
                 location=PROJECT_LOCATION,

providers/tests/google/cloud/operators/test_speech_to_text.py

@@ -23,7 +23,6 @@
 from google.api_core.gapic_v1.method import DEFAULT
 from google.cloud.speech_v1 import RecognitionAudio, RecognitionConfig, RecognizeResponse
 
-from airflow.exceptions import AirflowException
 from airflow.providers.google.cloud.operators.speech_to_text import CloudSpeechToTextRecognizeSpeechOperator
 
 PROJECT_ID = "project-id"
@@ -59,26 +58,20 @@ def test_recognize_speech_green_path(self, mock_hook):
     def test_missing_config(self, mock_hook):
         mock_hook.return_value.recognize_speech.return_value = True
 
-        with pytest.raises(AirflowException) as ctx:
+        with pytest.raises(TypeError, match="missing keyword argument 'config'"):
             CloudSpeechToTextRecognizeSpeechOperator(
                 project_id=PROJECT_ID, gcp_conn_id=GCP_CONN_ID, audio=AUDIO, task_id="id"
             ).execute(context={"task_instance": Mock()})
-
-        err = ctx.value
-        assert "config" in str(err)
         mock_hook.assert_not_called()
 
     @patch("airflow.providers.google.cloud.operators.speech_to_text.CloudSpeechToTextHook")
     def test_missing_audio(self, mock_hook):
         mock_hook.return_value.recognize_speech.return_value = True
 
-        with pytest.raises(AirflowException) as ctx:
+        with pytest.raises(TypeError, match="missing keyword argument 'audio'"):
             CloudSpeechToTextRecognizeSpeechOperator(
                 project_id=PROJECT_ID, gcp_conn_id=GCP_CONN_ID, config=CONFIG, task_id="id"
             ).execute(context={"task_instance": Mock()})
-
-        err = ctx.value
-        assert "audio" in str(err)
         mock_hook.assert_not_called()
 
     @patch("airflow.providers.google.cloud.operators.speech_to_text.FileDetailsLink.persist")

providers/tests/google/cloud/sensors/test_dataproc.py

@@ -132,7 +132,7 @@ def test_cancelled(self, mock_hook):
 
     @mock.patch(DATAPROC_PATH.format("DataprocHook"))
     def test_missing_region(self, mock_hook):
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="missing keyword argument 'region'"):
             DataprocJobSensor(
                 task_id=TASK_ID,
                 project_id=GCP_PROJECT,

providers/tests/google/cloud/transfers/test_gcs_to_bigquery.py

@@ -906,7 +906,7 @@ def test_execute_should_throw_ex_when_no_bucket_specified(self, hook):
         ]
         hook.return_value.generate_job_id.return_value = REAL_JOB_ID
         hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET, TABLE)
-        with pytest.raises(AirflowException, match=r"missing keyword argument 'bucket'"):
+        with pytest.raises(TypeError, match=r"missing keyword argument 'bucket'"):
             GCSToBigQueryOperator(
                 task_id=TASK_ID,
                 source_objects=TEST_SOURCE_OBJECTS,
@@ -926,7 +926,7 @@ def test_execute_should_throw_ex_when_no_source_objects_specified(self, hook):
         ]
         hook.return_value.generate_job_id.return_value = REAL_JOB_ID
         hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET, TABLE)
-        with pytest.raises(AirflowException, match=r"missing keyword argument 'source_objects'"):
+        with pytest.raises(TypeError, match=r"missing keyword argument 'source_objects'"):
             GCSToBigQueryOperator(
                 task_id=TASK_ID,
                 destination_project_dataset_table=TEST_EXPLICIT_DEST,
@@ -946,9 +946,7 @@ def test_execute_should_throw_ex_when_no_destination_project_dataset_table_speci
         ]
         hook.return_value.generate_job_id.return_value = REAL_JOB_ID
         hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET, TABLE)
-        with pytest.raises(
-            AirflowException, match=r"missing keyword argument 'destination_project_dataset_table'"
-        ):
+        with pytest.raises(TypeError, match=r"missing keyword argument 'destination_project_dataset_table'"):
             GCSToBigQueryOperator(
                 task_id=TASK_ID,
                 schema_fields=SCHEMA_FIELDS,

providers/tests/salesforce/operators/test_bulk.py

@@ -20,7 +20,6 @@
 
 import pytest
 
-from airflow.exceptions import AirflowException
 from airflow.providers.salesforce.operators.bulk import SalesforceBulkOperator
 
 
@@ -33,7 +32,7 @@ def test_execute_missing_operation(self):
         """
         Test execute missing operation
         """
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="missing keyword argument 'operation'"):
             SalesforceBulkOperator(
                 task_id="no_missing_operation_arg",
                 object_name="Account",
@@ -52,7 +51,7 @@ def test_execute_missing_object_name(self):
         """
         Test execute missing object_name
         """
-        with pytest.raises(AirflowException):
+        with pytest.raises(TypeError, match="missing keyword argument 'object_name'"):
             SalesforceBulkOperator(
                 task_id="no_object_name_arg",
                 operation="insert",

tests/serialization/test_dag_serialization.py

@@ -1258,7 +1258,12 @@ def test_no_new_fields_added_to_base_operator(self):
             "max_active_tis_per_dag": None,
             "max_active_tis_per_dagrun": None,
             "max_retry_delay": None,
+            "on_execute_callback": None,
             "on_failure_fail_dagrun": False,
+            "on_failure_callback": None,
+            "on_retry_callback": None,
+            "on_skipped_callback": None,
+            "on_success_callback": None,
             "outlets": [],
             "owner": "airflow",
             "params": {},