Address deprecations, make DeprecationWarning and RemovedInAirflow3Wa…

…rning raise error (#4951)

* Fix implicit schedule deprecation warning

* Ignore Airflow Elasticsearch provider warning

* Make DeprecationWarnings and RemovedInAirflow3Warnings raise errors

* Link to specific issue

catalog/pytest.ini

@@ -19,6 +19,13 @@ addopts =
     --allow-unix-socket
 
 
+
+# ==ERRORS==
+# Deprecation warnings and Airflow 3 warnings should be treated as errors so we
+# can address them as soon as possible
+# ==IGNORATIONS==
+# The following warnings are due to upstream dependencies and are not within our
+# control to fix. We will ignore them for now.
 # flask
 #   https://docs.sqlalchemy.org/en/20/errors.html#error-b8d9
 #   Warning in dependency, nothing we can do
@@ -30,10 +37,16 @@ addopts =
 #   https://github.com/marshmallow-code/marshmallow/issues/2227
 #   Upstream dependency is fixed but Airflow constraints are keeping us on an
 #   older version
+# elasticsearch
+#   https://github.com/apache/airflow/pull/41871
+#   Warning in upstream Airflow hook, fixed in the next version
 filterwarnings=
+    error::DeprecationWarning
+    error::airflow.exceptions.RemovedInAirflow3Warning
     ignore:.*is deprecated and will be (remoevd|removed) in Flask 2.3.:DeprecationWarning
     ignore:datetime.datetime.utcnow:DeprecationWarning:botocore
     ignore::DeprecationWarning:marshmallow.*
+    ignore:Importing from the 'elasticsearch.client' module is deprecated:DeprecationWarning:airflow.providers.elasticsearch.hooks.elasticsearch
 
 # Change the pytest cache location since Docker cannot write within the module file
 # structure due to permissions issues

catalog/tests/dags/common/sensors/test_single_run_external_dags_sensor.py

@@ -37,6 +37,7 @@ def create_dag(
 ):
     with DAG(
         f"{sample_dag_id_fixture}_{dag_id}",
+        schedule=None,
         default_args={
             "owner": "airflow",
             "start_date": DEFAULT_DATE,
@@ -91,6 +92,7 @@ def test_fails_if_external_dag_does_not_exist(clean_db, setup_pool):
     ):
         dag = DAG(
             "test_missing_dag_error",
+            schedule=None,
             default_args={
                 "owner": "airflow",
                 "start_date": DEFAULT_DATE,
@@ -134,6 +136,7 @@ def test_fails_if_external_dag_missing_sensor_task(clean_db, setup_pool):
     with pytest.raises(AirflowException, match=error_msg):
         dag = DAG(
             "test_missing_task_error",
+            schedule=None,
             default_args={
                 "owner": "airflow",
                 "start_date": DEFAULT_DATE,
@@ -182,6 +185,7 @@ def test_succeeds_if_no_running_dags(
     # Create the Test DAG and sensor with dependent dag Ids
     dag = DAG(
         "test_dag_success",
+        schedule=None,
         default_args={
             "owner": "airflow",
             "start_date": DEFAULT_DATE,
@@ -233,6 +237,7 @@ def test_retries_if_running_dags_with_completed_sensor_task(
     # Create the Test DAG and sensor and set up dependent dag Ids
     dag = DAG(
         "test_dag_failure",
+        schedule=None,
         default_args={
             "owner": "airflow",
             "start_date": DEFAULT_DATE,

catalog/tests/dags/common/sensors/test_utils.py

@@ -9,11 +9,13 @@
 
 
 TEST_DAG_ID = "data_refresh_dag_factory_test_dag"
-TEST_DAG = DAG(TEST_DAG_ID, default_args={"owner": "airflow"})
+TEST_DAG = DAG(TEST_DAG_ID, schedule=None, default_args={"owner": "airflow"})
 
 
 def _create_dagrun(start_date, sample_dag_id_fixture, conf={}):
-    return DAG(sample_dag_id_fixture, default_args={"owner": "airflow"}).create_dagrun(
+    return DAG(
+        sample_dag_id_fixture, schedule=None, default_args={"owner": "airflow"}
+    ).create_dagrun(
         start_date=start_date,
         execution_date=start_date,
         data_interval=(start_date, start_date),

catalog/tests/dags/database/staging_database_restore/test_staging_database_restore.py

@@ -104,7 +104,7 @@ def test_get_staging_db_details(details, mock_rds_hook):
 
 def test_make_rename_task_group():
     rule = TriggerRule.NONE_FAILED
-    with DAG(dag_id="test_make_rename_task_group", start_date=datetime(1970, 1, 1)):
+    with DAG(dag_id="test_make_rename_task_group", schedule=None):
         group = staging_database_restore.make_rename_task_group("dibble", "crim", rule)
     assert group.group_id == "rename_dibble_to_crim"
     rename, await_rename = list(group)

catalog/tests/dags/providers/test_provider_dag_factory.py

@@ -1,4 +1,4 @@
-from datetime import datetime, timedelta
+from datetime import timedelta
 from unittest import mock
 
 import pytest
@@ -131,7 +131,7 @@ def test_apply_configuration_overrides():
     # Create a mock DAG
     dag = DAG(
         dag_id="test_dag",
-        start_date=datetime(1970, 1, 1),
+        schedule=None,
         default_args={"execution_timeout": timedelta(hours=1)},
     )
     with dag: