Fix SMTP connection from_email being ignored by email_on_failure and …

providers/smtp/src/airflow/providers/smtp/notifications/smtp.py

@@ -124,7 +124,7 @@ def _build_email_content(self, smtp: SmtpHook, context: Context):
             if smtp.from_email is not None:
                 self.from_email = smtp.from_email
             else:
-                raise ValueError("You should provide `from_email` or define it in the connection")
+                self.from_email = conf.get("email", "from_email", fallback="airflow@airflow")
             fields_to_re_render.append("from_email")
         if self.subject is None:
             smtp_default_templated_subject_path: str

providers/smtp/tests/unit/smtp/notifications/test_smtp.py

@@ -175,6 +175,8 @@ def test_notifier_with_defaults(self, mock_smtphook_hook, create_dag_without_db,
     def test_notifier_with_nondefault_connection_extra(
         self, mock_smtphook_hook, create_dag_without_db, mock_task_instance
     ):
+        """When the connection extra defines `from_email` (and no config `[email] from_email` is set),
+        the connection's from_email is used, including template rendering."""
         # TODO: we can use create_runtime_ti fixture in place of mock_task_instance once provider has minimum AF to Airflow 3.0+
         ti = mock_task_instance(
             dag_id=TEST_DAG_ID,
@@ -211,6 +213,70 @@ def test_notifier_with_nondefault_connection_extra(
                 **DEFAULT_EMAIL_PARAMS,
             )
 
+    @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook")
+    def test_notifier_from_email_falls_back_to_config(self, mock_smtphook_hook, create_dag_without_db):
+        """When config [email] from_email is set and the SMTP connection extra `from_email` is NOT set, `[email] from_email` config is used."""
+        mock_smtphook_hook.return_value.__enter__.return_value.from_email = None
+        mock_smtphook_hook.return_value.__enter__.return_value.subject_template = None
+        mock_smtphook_hook.return_value.__enter__.return_value.html_content_template = None
+
+        with conf_vars({("email", "from_email"): "config@example.com"}):
+            notifier = SmtpNotifier(to=TEST_RECEIVER, subject=TEST_SUBJECT, html_content=TEST_BODY)
+            notifier({"dag": create_dag_without_db(TEST_DAG_ID)})
+
+        mock_smtphook_hook.return_value.__enter__().send_email_smtp.assert_called_once_with(
+            from_email="config@example.com",
+            to=TEST_RECEIVER,
+            subject=TEST_SUBJECT,
+            html_content=TEST_BODY,
+            smtp_conn_id=SMTP_CONN_ID,
+            **DEFAULT_EMAIL_PARAMS,
+        )
+
+    @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook")
+    def test_notifier_connection_extra_takes_precedence_over_config(
+        self, mock_smtphook_hook, create_dag_without_db
+    ):
+        """The SMTP connection extra `from_email` must win over the `[email] from_email` config."""
+        mock_smtphook_hook.return_value.__enter__.return_value.from_email = "conn@example.com"
+        mock_smtphook_hook.return_value.__enter__.return_value.subject_template = None
+        mock_smtphook_hook.return_value.__enter__.return_value.html_content_template = None
+
+        with conf_vars({("email", "from_email"): "config@example.com"}):
+            notifier = SmtpNotifier(to=TEST_RECEIVER, subject=TEST_SUBJECT, html_content=TEST_BODY)
+            notifier({"dag": create_dag_without_db(TEST_DAG_ID)})
+
+        mock_smtphook_hook.return_value.__enter__().send_email_smtp.assert_called_once_with(
+            from_email="conn@example.com",
+            to=TEST_RECEIVER,
+            subject=TEST_SUBJECT,
+            html_content=TEST_BODY,
+            smtp_conn_id=SMTP_CONN_ID,
+            **DEFAULT_EMAIL_PARAMS,
+        )
+
+    @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook")
+    def test_notifier_from_email_default_when_nothing_configured(
+        self, mock_smtphook_hook, create_dag_without_db
+    ):
+        """Falls back to the hard-coded default when neither the connection nor config define from_email."""
+        mock_smtphook_hook.return_value.__enter__.return_value.from_email = None
+        mock_smtphook_hook.return_value.__enter__.return_value.subject_template = None
+        mock_smtphook_hook.return_value.__enter__.return_value.html_content_template = None
+
+        with conf_vars({("email", "from_email"): None}):
+            notifier = SmtpNotifier(to=TEST_RECEIVER, subject=TEST_SUBJECT, html_content=TEST_BODY)
+            notifier({"dag": create_dag_without_db(TEST_DAG_ID)})
+
+        mock_smtphook_hook.return_value.__enter__().send_email_smtp.assert_called_once_with(
+            from_email="airflow@airflow",
+            to=TEST_RECEIVER,
+            subject=TEST_SUBJECT,
+            html_content=TEST_BODY,
+            smtp_conn_id=SMTP_CONN_ID,
+            **DEFAULT_EMAIL_PARAMS,
+        )
+
     @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook")
     def test_notifier_with_custom_smtp_conn_id(self, mock_smtphook_hook, create_dag_without_db):
         """Test that a custom smtp_conn_id is correctly passed to SmtpHook."""

task-sdk/src/airflow/sdk/execution_time/task_runner.py

@@ -2035,6 +2035,7 @@ def _send_error_email_notification(
     if email_backend and email_backend != _DEFAULT_EMAIL_BACKEND:
         notifier_class: _ErrorEmailNotifier = _LegacyEmailBackendNotifier
         notifier_description = f"configured email_backend {email_backend!r}"
+        using_smtp_notifier = False
     else:
         try:
             from airflow.providers.smtp.notifications.smtp import SmtpNotifier
@@ -2046,6 +2047,7 @@ def _send_error_email_notification(
             )
             return
         notifier_class = SmtpNotifier
+        using_smtp_notifier = True
 
     subject_template_file = conf.get("email", "subject_template", fallback=None)
 
@@ -2093,7 +2095,9 @@ def _send_error_email_notification(
             to=to_emails,
             subject=subject,
             html_content=html_content,
-            from_email=conf.get("email", "from_email", fallback="airflow@airflow"),
+            from_email=None
+            if using_smtp_notifier
+            else conf.get("email", "from_email", fallback="airflow@airflow"),
         )
         notifier(email_context)
     except Exception:

task-sdk/tests/task_sdk/execution_time/test_task_runner.py

@@ -3920,7 +3920,7 @@ def execute(self, context):
                 else:
                     mock_smtp_notifier.assert_called_once()
                     kwargs = mock_smtp_notifier.call_args.kwargs
-                    assert kwargs["from_email"] == self.FROM
+                    assert kwargs["from_email"] is None
                     assert kwargs["to"] == emails
                     assert (
                         kwargs["html_content"]
@@ -3978,7 +3978,7 @@ def execute(self, context):
                 else:
                     mock_smtp_notifier.assert_called_once()
                     kwargs = mock_smtp_notifier.call_args.kwargs
-                    assert kwargs["from_email"] == self.FROM
+                    assert kwargs["from_email"] is None
                     assert kwargs["to"] == emails
                     assert (
                         kwargs["html_content"]
@@ -4030,7 +4030,7 @@ def execute(self, context):
                     kwargs["html_content"]
                     == "<h1>Custom Template</h1><p>Task: {{ti.task_id}}</p><p>Error: {{exception_html}}</p>"
                 )
-                assert kwargs["from_email"] == self.FROM
+                assert kwargs["from_email"] is None
 
     def test_custom_email_backend_is_used(self, create_runtime_ti, mock_supervisor_comms):
         """A custom ``[email] email_backend`` is wrapped and invoked with rendered fields."""