Resolve deprecations in tests/utils/

tests/deprecations_ignore.yml

@@ -176,44 +176,6 @@
 - tests/models/test_timestamp.py::test_timestamp_behaviour_with_timezone
 - tests/models/test_xcom.py::TestXCom::test_set_serialize_call_old_signature
 - tests/ti_deps/deps/test_prev_dagrun_dep.py::TestPrevDagrunDep::test_first_task_run_of_new_task
-- tests/utils/log/test_log_reader.py::TestLogView::test_read_log_stream_should_read_each_try_in_turn
-- tests/utils/log/test_log_reader.py::TestLogView::test_read_log_stream_should_support_multiple_chunks
-- tests/utils/log/test_log_reader.py::TestLogView::test_supports_external_link
-- tests/utils/log/test_log_reader.py::TestLogView::test_task_log_filename_unique
-- tests/utils/log/test_log_reader.py::TestLogView::test_test_read_log_chunks_should_read_all_files
-- tests/utils/log/test_log_reader.py::TestLogView::test_test_read_log_chunks_should_read_one_try
-- tests/utils/log/test_log_reader.py::TestLogView::test_test_test_read_log_stream_should_read_all_logs
-- tests/utils/log/test_log_reader.py::TestLogView::test_test_test_read_log_stream_should_read_one_try
-- tests/utils/test_cli_util.py::TestCliUtil::test_get_dags
-- tests/utils/test_dates.py::TestDates::test_days_ago
-- tests/utils/test_dates.py::TestUtilsDatesDateRange::test_both_end_date_and_num_given
-- tests/utils/test_dates.py::TestUtilsDatesDateRange::test_delta_cron_presets
-- tests/utils/test_dates.py::TestUtilsDatesDateRange::test_end_date_before_start_date
-- tests/utils/test_dates.py::TestUtilsDatesDateRange::test_invalid_delta
-- tests/utils/test_dates.py::TestUtilsDatesDateRange::test_negative_num_given
-- tests/utils/test_dates.py::TestUtilsDatesDateRange::test_no_delta
-- tests/utils/test_dates.py::TestUtilsDatesDateRange::test_positive_num_given
-- tests/utils/test_db_cleanup.py::TestDBCleanup::test_no_models_missing
-- tests/utils/test_email.py::TestEmailSmtp::test_send_mime
-- tests/utils/test_email.py::TestEmailSmtp::test_send_mime_complete_failure
-- tests/utils/test_email.py::TestEmailSmtp::test_send_mime_custom_timeout_retrylimit
-- tests/utils/test_email.py::TestEmailSmtp::test_send_mime_dryrun
-- tests/utils/test_email.py::TestEmailSmtp::test_send_mime_noauth
-- tests/utils/test_email.py::TestEmailSmtp::test_send_mime_partial_failure
-- tests/utils/test_email.py::TestEmailSmtp::test_send_mime_ssl_complete_failure
-- tests/utils/test_email.py::TestEmailSmtp::test_send_mime_ssl_default_context_if_not_set
-- tests/utils/test_email.py::TestEmailSmtp::test_send_mime_ssl_default_context_with_value_set_to_default
-- tests/utils/test_email.py::TestEmailSmtp::test_send_mime_ssl_none_context
-- tests/utils/test_log_handlers.py::TestFileTaskLogHandler::test_file_task_handler
-- tests/utils/test_log_handlers.py::TestFileTaskLogHandler::test_file_task_handler_running
-- tests/utils/test_log_handlers.py::TestFileTaskLogHandler::test_file_task_handler_when_ti_value_is_invalid
-- tests/utils/test_log_handlers.py::TestFileTaskLogHandler::test_read_from_k8s_under_multi_namespace_mode
-- tests/utils/test_sqlalchemy.py::TestSqlAlchemyUtils::test_process_bind_param_naive
-- tests/utils/test_sqlalchemy.py::TestSqlAlchemyUtils::test_utc_transformations
-- tests/utils/test_state.py::test_dagrun_state_enum_escape
-- tests/utils/test_task_handler_with_custom_formatter.py::test_custom_formatter_custom_format_not_affected_by_config
-- tests/utils/test_task_handler_with_custom_formatter.py::test_custom_formatter_default_format
-- tests/utils/test_types.py::test_runtype_enum_escape
 
 
 # Operators

tests/utils/log/test_log_reader.py

@@ -79,7 +79,6 @@ def settings_folder(self):
     def configure_loggers(self, log_dir, settings_folder):
         logging_config = copy.deepcopy(DEFAULT_LOGGING_CONFIG)
         logging_config["handlers"]["task"]["base_log_folder"] = log_dir
-        logging_config["handlers"]["task"]["filename_template"] = self.FILENAME_TEMPLATE
         settings_file = os.path.join(settings_folder, "airflow_local_settings_test.py")
         with open(settings_file, "w") as handle:
             new_logging_file = f"LOGGING_CONFIG = {logging_config}"

tests/utils/test_dates.py

@@ -27,7 +27,9 @@
 
 
 class TestDates:
-    @pytest.mark.filterwarnings("ignore::DeprecationWarning")
+    @pytest.mark.filterwarnings(
+        "ignore:Function `days_ago` is deprecated.*:airflow.exceptions.RemovedInAirflow3Warning"
+    )
     def test_days_ago(self):
         today = pendulum.today()
         today_midnight = pendulum.instance(datetime.fromordinal(today.date().toordinal()))
@@ -103,7 +105,9 @@ def test_scale_time_units(self):
         assert arr4 == pytest.approx([2.3147, 1.1574], rel=1e-3)
 
 
-@pytest.mark.filterwarnings("ignore:`airflow.utils.dates.date_range:DeprecationWarning")
+@pytest.mark.filterwarnings(
+    r"ignore:`airflow.utils.dates.date_range\(\)` is deprecated:airflow.exceptions.RemovedInAirflow3Warning"
+)
 class TestUtilsDatesDateRange:
     def test_no_delta(self):
         assert dates.date_range(datetime(2016, 1, 1), datetime(2016, 1, 3)) == []

tests/utils/test_db_cleanup.py

@@ -297,6 +297,10 @@ def test__skip_archive(self, skip_archive, expected_archives):
             assert len(session.query(model).all()) == 5
             assert len(_get_archived_table_names(["dag_run"], session)) == expected_archives
 
+    @pytest.mark.filterwarnings(
+        # This test case might import some deprecated modules, ignore it
+        "ignore:This module is deprecated.*:airflow.exceptions.RemovedInAirflow3Warning"
+    )
     def test_no_models_missing(self):
         """
         1. Verify that for all tables in `airflow.models`, we either have them enabled in db cleanup,

tests/utils/test_email.py

@@ -17,6 +17,7 @@
 # under the License.
 from __future__ import annotations
 
+import json
 import os
 from email.mime.application import MIMEApplication
 from email.mime.multipart import MIMEMultipart
@@ -27,6 +28,7 @@
 import pytest
 
 from airflow.configuration import conf
+from airflow.exceptions import RemovedInAirflow3Warning
 from airflow.utils import email
 from tests.test_utils.config import conf_vars
 
@@ -147,6 +149,13 @@ def test_build_mime_message(self):
 
 @pytest.mark.db_test
 class TestEmailSmtp:
+    @pytest.fixture(autouse=True)
+    def setup_test_cases(self, monkeypatch):
+        monkeypatch.setenv(  # Set the default smtp connection for all test cases
+            "AIRFLOW_CONN_SMTP_DEFAULT",
+            json.dumps({"conn_type": "smtp", "login": "user", "password": "p@$$word"}),
+        )
+
     @mock.patch("airflow.utils.email.send_mime_email")
     def test_send_smtp(self, mock_send_mime, tmp_path):
         path = tmp_path / "testfile"
@@ -202,10 +211,15 @@ def test_send_bcc_smtp(self, mock_send_mime, tmp_path):
 
     @mock.patch("smtplib.SMTP_SSL")
     @mock.patch("smtplib.SMTP")
-    def test_send_mime(self, mock_smtp, mock_smtp_ssl):
+    def test_send_mime_airflow_config(self, mock_smtp, mock_smtp_ssl, monkeypatch):
+        monkeypatch.delenv("AIRFLOW_CONN_SMTP_DEFAULT", raising=False)
         mock_smtp.return_value = mock.Mock()
         msg = MIMEMultipart()
-        email.send_mime_email("from", "to", msg, dryrun=False)
+        with pytest.warns(
+            RemovedInAirflow3Warning,
+            match="Fetching SMTP credentials from configuration variables.*deprecated",
+        ):
+            email.send_mime_email("from", "to", msg, dryrun=False)
         mock_smtp.assert_called_once_with(
             host=conf.get("smtp", "SMTP_HOST"),
             port=conf.getint("smtp", "SMTP_PORT"),
@@ -221,16 +235,14 @@ def test_send_mime(self, mock_smtp, mock_smtp_ssl):
         assert mock_smtp.return_value.quit.called
 
     @mock.patch("smtplib.SMTP")
-    @mock.patch("airflow.hooks.base.BaseHook")
-    def test_send_mime_conn_id(self, mock_hook, mock_smtp):
+    def test_send_mime_conn_id(self, mock_smtp, monkeypatch):
+        monkeypatch.setenv(
+            "AIRFLOW_CONN_SMTP_TEST_CONN",
+            json.dumps({"conn_type": "smtp", "login": "test-user", "password": "test-p@$$word"}),
+        )
         msg = MIMEMultipart()
-        mock_conn = mock.Mock()
-        mock_conn.login = "user"
-        mock_conn.password = "password"
-        mock_hook.get_connection.return_value = mock_conn
-        email.send_mime_email("from", "to", msg, dryrun=False, conn_id="smtp_default")
-        mock_hook.get_connection.assert_called_with("smtp_default")
-        mock_smtp.return_value.login.assert_called_once_with("user", "password")
+        email.send_mime_email("from", "to", msg, dryrun=False, conn_id="smtp_test_conn")
+        mock_smtp.return_value.login.assert_called_once_with("test-user", "test-p@$$word")
         mock_smtp.return_value.sendmail.assert_called_once_with("from", "to", msg.as_string())
         assert mock_smtp.return_value.quit.called
 

tests/utils/test_log_handlers.py

@@ -31,6 +31,7 @@
 from kubernetes.client import models as k8s
 
 from airflow.config_templates.airflow_local_settings import DEFAULT_LOGGING_CONFIG
+from airflow.exceptions import RemovedInAirflow3Warning
 from airflow.executors import executor_loader
 from airflow.jobs.job import Job
 from airflow.jobs.triggerer_job_runner import TriggererJobRunner
@@ -75,6 +76,13 @@ def setup_method(self):
     def teardown_method(self):
         self.clean_up()
 
+    def test_deprecated_filename_template(self):
+        with pytest.warns(
+            RemovedInAirflow3Warning,
+            match="Passing filename_template to a log handler is deprecated and has no effect",
+        ):
+            FileTaskHandler("", filename_template="/foo/bar")
+
     def test_default_task_logging_setup(self):
         # file task handler is used by default.
         logger = logging.getLogger(TASK_LOGGER)
@@ -92,6 +100,7 @@ def task_callable(ti):
             run_type=DagRunType.MANUAL,
             state=State.RUNNING,
             execution_date=DEFAULT_DATE,
+            data_interval=dag.timetable.infer_manual_data_interval(run_after=DEFAULT_DATE),
         )
         task = PythonOperator(
             task_id="task_for_testing_file_log_handler",
@@ -144,6 +153,7 @@ def task_callable(ti):
             run_type=DagRunType.MANUAL,
             state=State.RUNNING,
             execution_date=DEFAULT_DATE,
+            data_interval=dag.timetable.infer_manual_data_interval(run_after=DEFAULT_DATE),
         )
         task = PythonOperator(
             task_id="task_for_testing_file_log_handler",
@@ -203,6 +213,7 @@ def task_callable(ti):
             run_type=DagRunType.MANUAL,
             state=State.RUNNING,
             execution_date=DEFAULT_DATE,
+            data_interval=dag.timetable.infer_manual_data_interval(run_after=DEFAULT_DATE),
         )
         ti = TaskInstance(task=task, run_id=dagrun.run_id)
 
@@ -413,6 +424,7 @@ def task_callable(ti):
             run_type=DagRunType.MANUAL,
             state=State.RUNNING,
             execution_date=DEFAULT_DATE,
+            data_interval=dag.timetable.infer_manual_data_interval(run_after=DEFAULT_DATE),
         )
         ti = TaskInstance(task=task, run_id=dagrun.run_id)
         ti.try_number = 3

tests/utils/test_sqlalchemy.py

@@ -82,6 +82,7 @@ def test_utc_transformations(self):
             execution_date=execution_date,
             start_date=start_date,
             session=self.session,
+            data_interval=dag.timetable.infer_manual_data_interval(run_after=execution_date),
         )
 
         assert execution_date == run.execution_date
@@ -113,6 +114,7 @@ def test_process_bind_param_naive(self):
                 execution_date=start_date,
                 start_date=start_date,
                 session=self.session,
+                data_interval=dag.timetable.infer_manual_data_interval(run_after=start_date),
             )
         dag.clear()
 

tests/utils/test_state.py

@@ -40,6 +40,7 @@ def test_dagrun_state_enum_escape():
             state=DagRunState.QUEUED,
             execution_date=DEFAULT_DATE,
             start_date=DEFAULT_DATE,
+            data_interval=dag.timetable.infer_manual_data_interval(run_after=DEFAULT_DATE),
             session=session,
         )
 

tests/utils/test_task_handler_with_custom_formatter.py

@@ -62,7 +62,12 @@ def custom_task_log_handler_config():
 def task_instance():
     dag = DAG(DAG_ID, start_date=DEFAULT_DATE)
     task = EmptyOperator(task_id=TASK_ID, dag=dag)
-    dagrun = dag.create_dagrun(DagRunState.RUNNING, execution_date=DEFAULT_DATE, run_type=DagRunType.MANUAL)
+    dagrun = dag.create_dagrun(
+        DagRunState.RUNNING,
+        execution_date=DEFAULT_DATE,
+        run_type=DagRunType.MANUAL,
+        data_interval=dag.timetable.infer_manual_data_interval(run_after=DEFAULT_DATE),
+    )
     ti = TaskInstance(task=task, run_id=dagrun.run_id)
     ti.log.disabled = False
     yield ti

tests/utils/test_types.py

@@ -35,12 +35,14 @@ def test_runtype_enum_escape():
     """
     with create_session() as session:
         dag = DAG(dag_id="test_enum_dags", start_date=DEFAULT_DATE)
+        data_interval = dag.timetable.infer_manual_data_interval(run_after=DEFAULT_DATE)
         dag.create_dagrun(
             run_type=DagRunType.SCHEDULED,
             state=State.RUNNING,
             execution_date=DEFAULT_DATE,
             start_date=DEFAULT_DATE,
             session=session,
+            data_interval=data_interval,
         )
 
         query = session.query(