This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 03bc8311d61 [v3-4-test] Add emailer utilities to the SMTP provider
(#74623) (#75704)
03bc8311d61 is described below
commit 03bc8311d6126dc04b44abe9b977fd03fa7c9414
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Oct 27 04:07:02 2026 +0000
[v3-4-test] Add emailer utilities to the SMTP provider (#74623) (#75704)
* Add core email utilities to the SMTP provider
Copy airflow.utils.email into airflow.providers.smtp.utils.emailer so that
core can later deprecate its email module and delegate to the provider,
once a provider release containing it is available.
The module is a copy of the core module. The only code changes are the
imports, which go through common.compat, and the Connection lookup, which
uses BaseHook.get_connection as other providers do instead of importing
airflow.models. SMTP settings still come from the [smtp] configuration
section and only the login and password from the Connection. Routing these
calls through SmtpHook instead would change behavior for existing setups,
because the hook reads all settings from the Connection and retries
differently.
Tests are copied from core, with additional edge cases. The module is
registered as an email backend.
* Describe the SMTP emailer without referring to core
The emailer documentation now explains what the module does and how it is
configured, without mentioning where the code came from.
(cherry picked from commit 31994976b54fa98bba8619dc97ec6d32db67ce92)
Co-authored-by: Shivam Rastogi <[email protected]>
---
docs/spelling_wordlist.txt | 1 +
providers/smtp/docs/index.rst | 1 +
providers/smtp/docs/utils/emailer.rst | 35 ++
providers/smtp/provider.yaml | 3 +
.../airflow/providers/smtp/get_provider_info.py | 1 +
.../src/airflow/providers/smtp/utils/__init__.py | 16 +
.../src/airflow/providers/smtp/utils/emailer.py | 340 +++++++++++++
providers/smtp/tests/unit/smtp/utils/__init__.py | 16 +
.../smtp/tests/unit/smtp/utils/test_emailer.py | 525 +++++++++++++++++++++
9 files changed, 938 insertions(+)
diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index c6b993e6b4c..8df92f5f9ed 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -549,6 +549,7 @@ Drillbit
driverless
dropdown
druidHook
+dryrun
ds
dsl
dsn
diff --git a/providers/smtp/docs/index.rst b/providers/smtp/docs/index.rst
index ef15a42b1b4..6f8e873d889 100644
--- a/providers/smtp/docs/index.rst
+++ b/providers/smtp/docs/index.rst
@@ -37,6 +37,7 @@
Configuration <configurations-ref>
Connection types <connections/smtp>
SMTP Notifications <notifications/smtp_notifier_howto_guide>
+ Email backend <utils/emailer>
Python API <_api/airflow/providers/smtp/index>
.. toctree::
diff --git a/providers/smtp/docs/utils/emailer.rst
b/providers/smtp/docs/utils/emailer.rst
new file mode 100644
index 00000000000..9f4345d5de1
--- /dev/null
+++ b/providers/smtp/docs/utils/emailer.rst
@@ -0,0 +1,35 @@
+.. Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+.. http://www.apache.org/licenses/LICENSE-2.0
+
+.. Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+Email backend
+=============
+
+:mod:`airflow.providers.smtp.utils.emailer` sends email over SMTP. SMTP server
+settings are read from the ``[smtp]`` configuration section, and the login and
+password from the Airflow Connection given by ``conn_id`` (``smtp_default`` by
+default). If that Connection does not exist, the email is sent without
authentication.
+
+To send Airflow's email notifications through this module, set:
+
+.. code-block:: ini
+
+ [email]
+ email_backend = airflow.providers.smtp.utils.emailer.send_email_smtp
+
+Unlike :class:`~airflow.providers.smtp.hooks.smtp.SmtpHook`, which takes all
of its
+settings from the SMTP Connection, this module reads the server settings from
the
+``[smtp]`` configuration section.
diff --git a/providers/smtp/provider.yaml b/providers/smtp/provider.yaml
index ed9b9ff7b60..fd9ab0906e8 100644
--- a/providers/smtp/provider.yaml
+++ b/providers/smtp/provider.yaml
@@ -187,3 +187,6 @@ connection-types:
notifications:
- airflow.providers.smtp.notifications.smtp.SmtpNotifier
+
+email-backends:
+ - airflow.providers.smtp.utils.emailer.send_email_smtp
diff --git a/providers/smtp/src/airflow/providers/smtp/get_provider_info.py
b/providers/smtp/src/airflow/providers/smtp/get_provider_info.py
index 46e46efb970..edb1e9c97b9 100644
--- a/providers/smtp/src/airflow/providers/smtp/get_provider_info.py
+++ b/providers/smtp/src/airflow/providers/smtp/get_provider_info.py
@@ -100,4 +100,5 @@ def get_provider_info():
}
],
"notifications":
["airflow.providers.smtp.notifications.smtp.SmtpNotifier"],
+ "email-backends":
["airflow.providers.smtp.utils.emailer.send_email_smtp"],
}
diff --git a/providers/smtp/src/airflow/providers/smtp/utils/__init__.py
b/providers/smtp/src/airflow/providers/smtp/utils/__init__.py
new file mode 100644
index 00000000000..13a83393a91
--- /dev/null
+++ b/providers/smtp/src/airflow/providers/smtp/utils/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git a/providers/smtp/src/airflow/providers/smtp/utils/emailer.py
b/providers/smtp/src/airflow/providers/smtp/utils/emailer.py
new file mode 100644
index 00000000000..c735735d800
--- /dev/null
+++ b/providers/smtp/src/airflow/providers/smtp/utils/emailer.py
@@ -0,0 +1,340 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Send email using the ``[email]`` and ``[smtp]`` Airflow configuration
sections."""
+
+from __future__ import annotations
+
+import collections.abc
+import logging
+import os
+import re
+import smtplib
+import ssl
+from collections.abc import Iterable
+from email.mime.application import MIMEApplication
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from email.utils import formatdate
+from typing import Any
+
+from airflow.providers.common.compat.sdk import AirflowException, BaseHook,
conf
+
+log = logging.getLogger(__name__)
+
+
+def send_email(
+ to: list[str] | Iterable[str],
+ subject: str,
+ html_content: str,
+ files: list[str] | None = None,
+ dryrun: bool = False,
+ cc: str | Iterable[str] | None = None,
+ bcc: str | Iterable[str] | None = None,
+ mime_subtype: str = "mixed",
+ mime_charset: str = "utf-8",
+ conn_id: str | None = None,
+ custom_headers: dict[str, Any] | None = None,
+ **kwargs,
+) -> None:
+ """
+ Send an email using the backend specified in the *EMAIL_BACKEND*
configuration option.
+
+ :param to: A list or iterable of email addresses to send the email to.
+ :param subject: The subject of the email.
+ :param html_content: The content of the email in HTML format.
+ :param files: A list of paths to files to attach to the email.
+ :param dryrun: If *True*, the email will not actually be sent. Default:
*False*.
+ :param cc: A string or iterable of strings containing email addresses to
send a copy of the email to.
+ :param bcc: A string or iterable of strings containing email addresses to
send a
+ blind carbon copy of the email to.
+ :param mime_subtype: The subtype of the MIME message. Default: "mixed".
+ :param mime_charset: The charset of the email. Default: "utf-8".
+ :param conn_id: The connection ID to use for the backend. If not provided,
the default connection
+ specified in the *EMAIL_CONN_ID* configuration option will be used.
+ :param custom_headers: A dictionary of additional headers to add to the
MIME message.
+ No validations are run on these values, and they should be able to be
encoded.
+ :param kwargs: Additional keyword arguments to pass to the backend.
+ """
+ backend = conf.getimport("email", "EMAIL_BACKEND")
+ backend_conn_id = conn_id or conf.get("email", "EMAIL_CONN_ID")
+ from_email = conf.get("email", "from_email", fallback=None)
+
+ to_list = get_email_address_list(to)
+ to_comma_separated = ", ".join(to_list)
+
+ return backend(
+ to_comma_separated,
+ subject,
+ html_content,
+ files=files,
+ dryrun=dryrun,
+ cc=cc,
+ bcc=bcc,
+ mime_subtype=mime_subtype,
+ mime_charset=mime_charset,
+ conn_id=backend_conn_id,
+ from_email=from_email,
+ custom_headers=custom_headers,
+ **kwargs,
+ )
+
+
+def send_email_smtp(
+ to: str | Iterable[str],
+ subject: str,
+ html_content: str,
+ files: list[str] | None = None,
+ dryrun: bool = False,
+ cc: str | Iterable[str] | None = None,
+ bcc: str | Iterable[str] | None = None,
+ mime_subtype: str = "mixed",
+ mime_charset: str = "utf-8",
+ conn_id: str = "smtp_default",
+ from_email: str | None = None,
+ custom_headers: dict[str, Any] | None = None,
+ **kwargs,
+) -> None:
+ """
+ Send an email with html content.
+
+ :param to: Recipient email address or list of addresses.
+ :param subject: Email subject.
+ :param html_content: Email body in HTML format.
+ :param files: List of file paths to attach to the email.
+ :param dryrun: If True, the email will not be sent, but all other actions
will be performed.
+ :param cc: Carbon copy recipient email address or list of addresses.
+ :param bcc: Blind carbon copy recipient email address or list of addresses.
+ :param mime_subtype: MIME subtype of the email.
+ :param mime_charset: MIME charset of the email.
+ :param conn_id: Connection ID of the SMTP server.
+ :param from_email: Sender email address.
+ :param custom_headers: Dictionary of custom headers to include in the
email.
+ :param kwargs: Additional keyword arguments.
+
+ >>> send_email("[email protected]", "foo", "<b>Foo</b> bar", ["/dev/null"],
dryrun=True)
+ """
+ smtp_mail_from = conf.get("smtp", "SMTP_MAIL_FROM")
+
+ if smtp_mail_from is not None:
+ mail_from = smtp_mail_from
+ else:
+ if from_email is None:
+ raise ValueError(
+ "You should set from email - either by smtp/smtp_mail_from
config or `from_email` parameter"
+ )
+ mail_from = from_email
+
+ msg, recipients = build_mime_message(
+ mail_from=mail_from,
+ to=to,
+ subject=subject,
+ html_content=html_content,
+ files=files,
+ cc=cc,
+ bcc=bcc,
+ mime_subtype=mime_subtype,
+ mime_charset=mime_charset,
+ custom_headers=custom_headers,
+ )
+
+ send_mime_email(e_from=mail_from, e_to=recipients, mime_msg=msg,
conn_id=conn_id, dryrun=dryrun)
+
+
+def build_mime_message(
+ mail_from: str | None,
+ to: str | Iterable[str],
+ subject: str,
+ html_content: str,
+ files: list[str] | None = None,
+ cc: str | Iterable[str] | None = None,
+ bcc: str | Iterable[str] | None = None,
+ mime_subtype: str = "mixed",
+ mime_charset: str = "utf-8",
+ custom_headers: dict[str, Any] | None = None,
+) -> tuple[MIMEMultipart, list[str]]:
+ """
+ Build a MIME message that can be used to send an email and returns a full
list of recipients.
+
+ :param mail_from: Email address to set as the email's "From" field.
+ :param to: A string or iterable of strings containing email addresses to
set as the email's "To" field.
+ :param subject: The subject of the email.
+ :param html_content: The content of the email in HTML format.
+ :param files: A list of paths to files to be attached to the email.
+ :param cc: A string or iterable of strings containing email addresses to
set as the email's "CC" field.
+ :param bcc: A string or iterable of strings containing email addresses to
set as the email's "BCC" field.
+ :param mime_subtype: The subtype of the MIME message. Default: "mixed".
+ :param mime_charset: The charset of the email. Default: "utf-8".
+ :param custom_headers: Additional headers to add to the MIME message. No
validations are run on these
+ values, and they should be able to be encoded.
+ :return: A tuple containing the email as a MIMEMultipart object and a list
of recipient email addresses.
+ """
+ to = get_email_address_list(to)
+
+ msg = MIMEMultipart(mime_subtype)
+ msg["Subject"] = subject
+ if mail_from:
+ msg["From"] = mail_from
+ msg["To"] = ", ".join(to)
+ recipients = to
+ if cc:
+ cc = get_email_address_list(cc)
+ msg["CC"] = ", ".join(cc)
+ recipients += cc
+
+ if bcc:
+ # don't add bcc in header
+ bcc = get_email_address_list(bcc)
+ recipients += bcc
+
+ msg["Date"] = formatdate(localtime=True)
+ mime_text = MIMEText(html_content, "html", mime_charset)
+ msg.attach(mime_text)
+
+ for fname in files or []:
+ basename = os.path.basename(fname)
+ with open(fname, "rb") as file:
+ part = MIMEApplication(file.read(), Name=basename)
+ part.add_header("Content-Disposition", "attachment",
filename=basename)
+ part["Content-ID"] = f"<{basename}>"
+ msg.attach(part)
+
+ if custom_headers:
+ for header_key, header_value in custom_headers.items():
+ msg[header_key] = header_value
+
+ return msg, recipients
+
+
+def send_mime_email(
+ e_from: str,
+ e_to: str | list[str],
+ mime_msg: MIMEMultipart,
+ conn_id: str = "smtp_default",
+ dryrun: bool = False,
+) -> None:
+ """
+ Send a MIME email.
+
+ :param e_from: The email address of the sender.
+ :param e_to: The email address or a list of email addresses of the
recipient(s).
+ :param mime_msg: The MIME message to send.
+ :param conn_id: The ID of the SMTP connection to use.
+ :param dryrun: If True, the email will not be sent, but a log message will
be generated.
+ """
+ smtp_host = conf.get_mandatory_value("smtp", "SMTP_HOST")
+ smtp_port = conf.getint("smtp", "SMTP_PORT")
+ smtp_starttls = conf.getboolean("smtp", "SMTP_STARTTLS")
+ smtp_ssl = conf.getboolean("smtp", "SMTP_SSL")
+ smtp_retry_limit = conf.getint("smtp", "SMTP_RETRY_LIMIT")
+ smtp_timeout = conf.getint("smtp", "SMTP_TIMEOUT")
+ smtp_user = None
+ smtp_password = None
+
+ if conn_id is not None:
+ try:
+ airflow_conn = BaseHook.get_connection(conn_id)
+ smtp_user = airflow_conn.login
+ smtp_password = airflow_conn.password
+ except AirflowException:
+ pass
+ if smtp_user is None or smtp_password is None:
+ log.debug("No user/password found for SMTP, so logging in with no
authentication.")
+
+ if not dryrun:
+ for attempt in range(smtp_retry_limit + 1):
+ log.info("Email alerting: attempt %s", str(attempt + 1))
+ try:
+ smtp_conn = _get_smtp_connection(smtp_host, smtp_port,
smtp_timeout, smtp_ssl)
+ except smtplib.SMTPServerDisconnected:
+ if attempt == smtp_retry_limit:
+ raise
+ else:
+ if smtp_starttls:
+ smtp_conn.starttls(context=_get_ssl_context())
+ if smtp_user and smtp_password:
+ smtp_conn.login(smtp_user, smtp_password)
+ log.info("Sent an alert email to %s", e_to)
+ smtp_conn.sendmail(e_from, e_to, mime_msg.as_string())
+ smtp_conn.quit()
+ break
+
+
+def get_email_address_list(addresses: str | Iterable[str]) -> list[str]:
+ """
+ Return a list of email addresses from the provided input.
+
+ :param addresses: A string or iterable of strings containing email
addresses.
+ :return: A list of email addresses.
+ :raises TypeError: If the input is not a string or iterable of strings.
+ """
+ if isinstance(addresses, str):
+ return _get_email_list_from_str(addresses)
+ if isinstance(addresses, collections.abc.Iterable):
+ if not all(isinstance(item, str) for item in addresses):
+ raise TypeError("The items in your iterable must be strings.")
+ return list(addresses)
+ raise TypeError(f"Unexpected argument type: Received
'{type(addresses).__name__}'.")
+
+
+def _get_ssl_context() -> ssl.SSLContext | None:
+ """
+ Return the SSL context configured via the ``email.ssl_context`` option.
+
+ ``"default"`` produces :func:`ssl.create_default_context`; ``"none"``
+ returns ``None`` so callers that explicitly want to skip certificate
+ validation (for example, against a self-signed SMTP server in a
+ lab environment) can still do so.
+ """
+ ssl_context_string = conf.get("email", "SSL_CONTEXT")
+ if ssl_context_string == "default":
+ return ssl.create_default_context()
+ if ssl_context_string == "none":
+ return None
+ raise RuntimeError(
+ f"The email.ssl_context configuration variable must "
+ f"be set to 'default' or 'none' and is '{ssl_context_string}."
+ )
+
+
+def _get_smtp_connection(host: str, port: int, timeout: int, with_ssl: bool)
-> smtplib.SMTP:
+ """
+ Return an SMTP connection to the specified host and port, with optional
SSL encryption.
+
+ :param host: The hostname or IP address of the SMTP server.
+ :param port: The port number to connect to on the SMTP server.
+ :param timeout: The timeout in seconds for the connection.
+ :param with_ssl: Whether to use SSL encryption for the connection.
+ :return: An SMTP connection to the specified host and port.
+ """
+ if not with_ssl:
+ return smtplib.SMTP(host=host, port=port, timeout=timeout)
+ return smtplib.SMTP_SSL(host=host, port=port, timeout=timeout,
context=_get_ssl_context())
+
+
+def _get_email_list_from_str(addresses: str) -> list[str]:
+ """
+ Extract a list of email addresses from a string.
+
+ The string can contain multiple email addresses separated
+ by any of the following delimiters: ',' or ';'.
+
+ :param addresses: A string containing one or more email addresses.
+ :return: A list of email addresses.
+ """
+ pattern = r"\s*[,;]\s*"
+ return re.split(pattern, addresses)
diff --git a/providers/smtp/tests/unit/smtp/utils/__init__.py
b/providers/smtp/tests/unit/smtp/utils/__init__.py
new file mode 100644
index 00000000000..13a83393a91
--- /dev/null
+++ b/providers/smtp/tests/unit/smtp/utils/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
diff --git a/providers/smtp/tests/unit/smtp/utils/test_emailer.py
b/providers/smtp/tests/unit/smtp/utils/test_emailer.py
new file mode 100644
index 00000000000..f2386216bb7
--- /dev/null
+++ b/providers/smtp/tests/unit/smtp/utils/test_emailer.py
@@ -0,0 +1,525 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+import os
+from email.mime.application import MIMEApplication
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from smtplib import SMTPAuthenticationError, SMTPServerDisconnected
+from unittest import mock
+
+import pytest
+
+from airflow.providers.common.compat.sdk import conf
+from airflow.providers.smtp.utils import emailer
+
+from tests_common.test_utils.config import conf_vars
+
+EMAILS = ["[email protected]", "[email protected]"]
+
+send_email_test = mock.MagicMock()
+
+
+class TestEmailer:
+ def setup_method(self):
+ conf.remove_option("email", "EMAIL_BACKEND")
+ conf.remove_option("email", "EMAIL_CONN_ID")
+
+ def test_get_email_address_single_email(self):
+ emails_string = "[email protected]"
+
+ assert emailer.get_email_address_list(emails_string) == [emails_string]
+
+ def test_get_email_address_comma_sep_string(self):
+ emails_string = "[email protected], [email protected]"
+
+ assert emailer.get_email_address_list(emails_string) == EMAILS
+
+ def test_get_email_address_colon_sep_string(self):
+ emails_string = "[email protected]; [email protected]"
+
+ assert emailer.get_email_address_list(emails_string) == EMAILS
+
+ def test_get_email_address_list(self):
+ emails_list = ["[email protected]", "[email protected]"]
+
+ assert emailer.get_email_address_list(emails_list) == EMAILS
+
+ def test_get_email_address_tuple(self):
+ emails_tuple = ("[email protected]", "[email protected]")
+
+ assert emailer.get_email_address_list(emails_tuple) == EMAILS
+
+ def test_get_email_address_invalid_type(self):
+ emails_string = 1
+
+ with pytest.raises(TypeError):
+ emailer.get_email_address_list(emails_string)
+
+ def test_get_email_address_invalid_type_in_iterable(self):
+ emails_list = ["[email protected]", 2]
+
+ with pytest.raises(TypeError):
+ emailer.get_email_address_list(emails_list)
+
+ @mock.patch("airflow.providers.smtp.utils.emailer.send_email")
+ def test_default_backend(self, mock_send_email):
+ res = emailer.send_email("to", "subject", "content")
+ mock_send_email.assert_called_once_with("to", "subject", "content")
+ assert mock_send_email.return_value == res
+
+ @mock.patch("airflow.providers.smtp.utils.emailer.send_email_smtp")
+ def test_custom_backend(self, mock_send_email):
+ with conf_vars(
+ {
+ ("email", "email_backend"):
"unit.smtp.utils.test_emailer.send_email_test",
+ ("email", "email_conn_id"): "smtp_default",
+ }
+ ):
+ emailer.send_email("to", "subject", "content")
+ send_email_test.assert_called_once_with(
+ "to",
+ "subject",
+ "content",
+ files=None,
+ dryrun=False,
+ cc=None,
+ bcc=None,
+ mime_charset="utf-8",
+ mime_subtype="mixed",
+ conn_id="smtp_default",
+ from_email=None,
+ custom_headers=None,
+ )
+ assert not mock_send_email.called
+
+ @mock.patch("airflow.providers.smtp.utils.emailer.send_email_smtp")
+ @conf_vars(
+ {
+ ("email", "email_backend"):
"unit.smtp.utils.test_emailer.send_email_test",
+ ("email", "email_conn_id"): "smtp_default",
+ ("email", "from_email"): "[email protected]",
+ }
+ )
+ def test_custom_backend_sender(self, mock_send_email_smtp):
+ emailer.send_email("to", "subject", "content")
+ _, call_kwargs = send_email_test.call_args
+ assert call_kwargs["from_email"] == "[email protected]"
+ assert not mock_send_email_smtp.called
+
+ def test_build_mime_message(self):
+ mail_from = "[email protected]"
+ mail_to = "[email protected]"
+ subject = "test subject"
+ html_content = "<html>Test</html>"
+ custom_headers = {"Reply-To": "[email protected]"}
+
+ msg, recipients = emailer.build_mime_message(
+ mail_from=mail_from,
+ to=mail_to,
+ subject=subject,
+ html_content=html_content,
+ custom_headers=custom_headers,
+ )
+
+ assert "From" in msg
+ assert "To" in msg
+ assert "Subject" in msg
+ assert "Reply-To" in msg
+ assert [mail_to] == recipients
+ assert msg["To"] == ",".join(recipients)
+
+ def test_build_mime_message_escapes_attachment_filename(self, tmp_path):
+ # A quote in the filename must not break out of the quoted
+ # Content-Disposition value and inject extra parameters.
+ malicious = 'report.txt"; x-evil="1'
+ attachment = tmp_path / malicious
+ attachment.write_bytes(b"data")
+
+ msg, _ = emailer.build_mime_message(
+ mail_from="[email protected]",
+ to="[email protected]",
+ subject="subject",
+ html_content="<html></html>",
+ files=[os.fspath(attachment)],
+ )
+
+ part = msg.get_payload()[-1]
+ assert part.get_filename() == malicious
+ assert "x-evil" not in
dict(part.get_params(header="Content-Disposition"))
+
+ def test_build_mime_message_cc_in_headers_bcc_only_in_recipients(self):
+ msg, recipients = emailer.build_mime_message(
+ mail_from="[email protected]",
+ to="[email protected]; [email protected]",
+ subject="subject",
+ html_content="content",
+ cc=["[email protected]"],
+ bcc="[email protected]",
+ )
+
+ assert recipients == ["[email protected]", "[email protected]",
"[email protected]", "[email protected]"]
+ assert msg["To"] == "[email protected], [email protected]"
+ assert msg["CC"] == "[email protected]"
+ assert "Bcc" not in msg
+
+ def test_build_mime_message_without_sender_omits_from_header(self):
+ msg, _ = emailer.build_mime_message(
+ mail_from=None, to="[email protected]", subject="s",
html_content="c"
+ )
+
+ assert "From" not in msg
+
+ def test_build_mime_message_missing_attachment_raises(self, tmp_path):
+ with pytest.raises(FileNotFoundError):
+ emailer.build_mime_message(
+ mail_from="[email protected]",
+ to="[email protected]",
+ subject="s",
+ html_content="c",
+ files=[os.fspath(tmp_path / "missing")],
+ )
+
+ @mock.patch("airflow.providers.smtp.utils.emailer.send_email_smtp")
+ def test_send_email_joins_recipients_and_prefers_conn_id_argument(self,
mock_send_email_smtp):
+ with conf_vars(
+ {
+ ("email", "email_backend"):
"airflow.providers.smtp.utils.emailer.send_email_smtp",
+ ("email", "email_conn_id"): "config_conn",
+ }
+ ):
+ emailer.send_email(["[email protected]", "[email protected]"],
"subject", "content", conn_id="arg_conn")
+
+ args, kwargs = mock_send_email_smtp.call_args
+ assert args[0] == "[email protected], [email protected]"
+ assert kwargs["conn_id"] == "arg_conn"
+
+
+class TestEmailerSmtp:
+ @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.providers.smtp.utils.emailer.send_mime_email")
+ def test_send_smtp(self, mock_send_mime, tmp_path):
+ path = tmp_path / "testfile"
+ path.write_text("attachment")
+ emailer.send_email_smtp("to", "subject", "content",
files=[os.fspath(path)])
+ assert mock_send_mime.called
+ _, call_args = mock_send_mime.call_args
+ assert conf.get("smtp", "SMTP_MAIL_FROM") == call_args["e_from"]
+ assert call_args["e_to"] == ["to"]
+ msg = call_args["mime_msg"]
+ assert msg["Subject"] == "subject"
+ assert conf.get("smtp", "SMTP_MAIL_FROM") == msg["From"]
+ assert len(msg.get_payload()) == 2
+ filename = f'attachment; filename="{path.name}"'
+ assert filename == msg.get_payload()[-1].get("Content-Disposition")
+ mimeapp = MIMEApplication("attachment")
+ assert mimeapp.get_payload() == msg.get_payload()[-1].get_payload()
+
+ @mock.patch("airflow.providers.smtp.utils.emailer.send_mime_email")
+ def test_send_smtp_with_multibyte_content(self, mock_send_mime):
+ emailer.send_email_smtp("to", "subject", "🔥", mime_charset="utf-8")
+ assert mock_send_mime.called
+ _, call_args = mock_send_mime.call_args
+ msg = call_args["mime_msg"]
+ mimetext = MIMEText("🔥", "mixed", "utf-8")
+ assert mimetext.get_payload() == msg.get_payload()[0].get_payload()
+
+ @mock.patch("airflow.providers.smtp.utils.emailer.send_mime_email")
+ def test_send_bcc_smtp(self, mock_send_mime, tmp_path):
+ path = tmp_path / "testfile"
+ path.write_text("attachment")
+ emailer.send_email_smtp(
+ "to",
+ "subject",
+ "content",
+ files=[os.fspath(path)],
+ cc="cc",
+ bcc="bcc",
+ custom_headers={"Reply-To": "[email protected]"},
+ )
+ assert mock_send_mime.called
+ _, call_args = mock_send_mime.call_args
+ assert conf.get("smtp", "SMTP_MAIL_FROM") == call_args["e_from"]
+ assert call_args["e_to"] == ["to", "cc", "bcc"]
+ msg = call_args["mime_msg"]
+ assert msg["Subject"] == "subject"
+ assert conf.get("smtp", "SMTP_MAIL_FROM") == msg["From"]
+ assert len(msg.get_payload()) == 2
+ assert f'attachment; filename="{path.name}"' ==
msg.get_payload()[-1].get("Content-Disposition")
+ mimeapp = MIMEApplication("attachment")
+ assert mimeapp.get_payload() == msg.get_payload()[-1].get_payload()
+ assert msg["Reply-To"] == "[email protected]"
+
+ @mock.patch("smtplib.SMTP_SSL")
+ @mock.patch("smtplib.SMTP")
+ 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()
+ emailer.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"),
+ timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+ )
+ assert not mock_smtp_ssl.called
+ assert mock_smtp.return_value.starttls.called
+ mock_smtp.return_value.sendmail.assert_called_once_with("from", "to",
msg.as_string())
+ assert mock_smtp.return_value.quit.called
+
+ @mock.patch("smtplib.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()
+ emailer.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
+
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_missing_conn_id_sends_without_auth(self, mock_smtp):
+ msg = MIMEMultipart()
+ emailer.send_mime_email("from", "to", msg, dryrun=False,
conn_id="smtp_missing_conn")
+ assert not mock_smtp.return_value.login.called
+ mock_smtp.return_value.sendmail.assert_called_once_with("from", "to",
msg.as_string())
+
+ @mock.patch("smtplib.SMTP_SSL")
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_ssl_none_context(self, mock_smtp, mock_smtp_ssl):
+ mock_smtp_ssl.return_value = mock.Mock()
+ with conf_vars({("smtp", "smtp_ssl"): "True", ("email",
"ssl_context"): "none"}):
+ emailer.send_mime_email("from", "to", MIMEMultipart(),
dryrun=False)
+ assert not mock_smtp.called
+ mock_smtp_ssl.assert_called_once_with(
+ host=conf.get("smtp", "SMTP_HOST"),
+ port=conf.getint("smtp", "SMTP_PORT"),
+ timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+ context=None,
+ )
+
+ @mock.patch("smtplib.SMTP_SSL")
+ @mock.patch("smtplib.SMTP")
+ @mock.patch("ssl.create_default_context")
+ def test_send_mime_ssl_default_context_if_not_set(self,
create_default_context, mock_smtp, mock_smtp_ssl):
+ mock_smtp_ssl.return_value = mock.Mock()
+ with conf_vars({("smtp", "smtp_ssl"): "True"}):
+ emailer.send_mime_email("from", "to", MIMEMultipart(),
dryrun=False)
+ assert not mock_smtp.called
+ assert create_default_context.called
+ mock_smtp_ssl.assert_called_once_with(
+ host=conf.get("smtp", "SMTP_HOST"),
+ port=conf.getint("smtp", "SMTP_PORT"),
+ timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+ context=create_default_context.return_value,
+ )
+
+ @mock.patch("smtplib.SMTP_SSL")
+ @mock.patch("smtplib.SMTP")
+ @mock.patch("ssl.create_default_context")
+ def test_send_mime_ssl_default_context_with_value_set_to_default(
+ self, create_default_context, mock_smtp, mock_smtp_ssl
+ ):
+ mock_smtp_ssl.return_value = mock.Mock()
+ with conf_vars({("smtp", "smtp_ssl"): "True", ("email",
"ssl_context"): "default"}):
+ emailer.send_mime_email("from", "to", MIMEMultipart(),
dryrun=False)
+ assert not mock_smtp.called
+ assert create_default_context.called
+ mock_smtp_ssl.assert_called_once_with(
+ host=conf.get("smtp", "SMTP_HOST"),
+ port=conf.getint("smtp", "SMTP_PORT"),
+ timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+ context=create_default_context.return_value,
+ )
+
+ @mock.patch("smtplib.SMTP_SSL")
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_noauth(self, mock_smtp, mock_smtp_ssl):
+ mock_smtp.return_value = mock.Mock()
+ with conf_vars(
+ {
+ ("smtp", "smtp_user"): None,
+ ("smtp", "smtp_password"): None,
+ }
+ ):
+ emailer.send_mime_email("from", "to", MIMEMultipart(),
dryrun=False)
+ assert not mock_smtp_ssl.called
+ mock_smtp.assert_called_once_with(
+ host=conf.get("smtp", "SMTP_HOST"),
+ port=conf.getint("smtp", "SMTP_PORT"),
+ timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+ )
+ assert not mock_smtp.login.called
+
+ @mock.patch("smtplib.SMTP_SSL")
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_dryrun(self, mock_smtp, mock_smtp_ssl):
+ emailer.send_mime_email("from", "to", MIMEMultipart(), dryrun=True)
+ assert not mock_smtp.called
+ assert not mock_smtp_ssl.called
+
+ @mock.patch("smtplib.SMTP_SSL")
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_complete_failure(self, mock_smtp: mock.Mock,
mock_smtp_ssl: mock.Mock):
+ mock_smtp.side_effect = SMTPServerDisconnected()
+ msg = MIMEMultipart()
+ with pytest.raises(SMTPServerDisconnected):
+ emailer.send_mime_email("from", "to", msg, dryrun=False)
+ mock_smtp.assert_any_call(
+ host=conf.get("smtp", "SMTP_HOST"),
+ port=conf.getint("smtp", "SMTP_PORT"),
+ timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+ )
+ assert mock_smtp.call_count == conf.getint("smtp", "SMTP_RETRY_LIMIT")
+ 1
+ assert not mock_smtp_ssl.called
+ assert not mock_smtp.return_value.starttls.called
+ assert not mock_smtp.return_value.login.called
+ assert not mock_smtp.return_value.sendmail.called
+ assert not mock_smtp.return_value.quit.called
+
+ @mock.patch("smtplib.SMTP_SSL")
+ @mock.patch("smtplib.SMTP")
+ @mock.patch("ssl.create_default_context")
+ def test_send_mime_ssl_complete_failure(self, create_default_context,
mock_smtp, mock_smtp_ssl):
+ mock_smtp_ssl.side_effect = SMTPServerDisconnected()
+ msg = MIMEMultipart()
+ with conf_vars({("smtp", "smtp_ssl"): "True"}):
+ with pytest.raises(SMTPServerDisconnected):
+ emailer.send_mime_email("from", "to", msg, dryrun=False)
+
+ mock_smtp_ssl.assert_any_call(
+ host=conf.get("smtp", "SMTP_HOST"),
+ port=conf.getint("smtp", "SMTP_PORT"),
+ timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+ context=create_default_context.return_value,
+ )
+ assert create_default_context.called
+ assert mock_smtp_ssl.call_count == conf.getint("smtp",
"SMTP_RETRY_LIMIT") + 1
+ assert not mock_smtp.called
+ assert not mock_smtp_ssl.return_value.starttls.called
+ assert not mock_smtp_ssl.return_value.login.called
+ assert not mock_smtp_ssl.return_value.sendmail.called
+ assert not mock_smtp_ssl.return_value.quit.called
+
+ @mock.patch("smtplib.SMTP_SSL")
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_custom_timeout_retrylimit(self, mock_smtp,
mock_smtp_ssl):
+ mock_smtp.side_effect = SMTPServerDisconnected()
+ msg = MIMEMultipart()
+
+ custom_retry_limit = 10
+ custom_timeout = 60
+
+ with conf_vars(
+ {
+ ("smtp", "smtp_retry_limit"): str(custom_retry_limit),
+ ("smtp", "smtp_timeout"): str(custom_timeout),
+ }
+ ):
+ with pytest.raises(SMTPServerDisconnected):
+ emailer.send_mime_email("from", "to", msg, dryrun=False)
+
+ mock_smtp.assert_any_call(
+ host=conf.get("smtp", "SMTP_HOST"), port=conf.getint("smtp",
"SMTP_PORT"), timeout=custom_timeout
+ )
+ assert not mock_smtp_ssl.called
+ assert mock_smtp.call_count == custom_retry_limit + 1
+
+ @mock.patch("smtplib.SMTP_SSL")
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_partial_failure(self, mock_smtp, mock_smtp_ssl):
+ final_mock = mock.Mock()
+ side_effects = [SMTPServerDisconnected(), SMTPServerDisconnected(),
final_mock]
+ mock_smtp.side_effect = side_effects
+ msg = MIMEMultipart()
+
+ emailer.send_mime_email("from", "to", msg, dryrun=False)
+
+ mock_smtp.assert_any_call(
+ host=conf.get("smtp", "SMTP_HOST"),
+ port=conf.getint("smtp", "SMTP_PORT"),
+ timeout=conf.getint("smtp", "SMTP_TIMEOUT"),
+ )
+ assert mock_smtp.call_count == side_effects.index(final_mock) + 1
+ assert not mock_smtp_ssl.called
+ assert final_mock.starttls.called
+ final_mock.sendmail.assert_called_once_with("from", "to",
msg.as_string())
+ assert final_mock.quit.called
+
+ @mock.patch("airflow.providers.smtp.utils.emailer.send_mime_email")
+ def test_send_smtp_mail_from_config_takes_precedence_over_from_email(self,
mock_send_mime):
+ with conf_vars({("smtp", "smtp_mail_from"): "[email protected]"}):
+ emailer.send_email_smtp("to", "subject", "content",
from_email="[email protected]")
+
+ _, call_args = mock_send_mime.call_args
+ assert call_args["e_from"] == "[email protected]"
+ assert call_args["mime_msg"]["From"] == "[email protected]"
+
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_conn_id_none_skips_connection_lookup(self, mock_smtp):
+ with mock.patch.object(emailer.BaseHook, "get_connection") as
mock_get_connection:
+ emailer.send_mime_email("from", "to", MIMEMultipart(),
conn_id=None)
+ assert not mock_get_connection.called
+ assert not mock_smtp.return_value.login.called
+ assert mock_smtp.return_value.sendmail.called
+
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_login_without_password_skips_auth(self, mock_smtp,
monkeypatch):
+ monkeypatch.setenv("AIRFLOW_CONN_SMTP_DEFAULT",
json.dumps({"conn_type": "smtp", "login": "user"}))
+ emailer.send_mime_email("from", "to", MIMEMultipart())
+ assert not mock_smtp.return_value.login.called
+ assert mock_smtp.return_value.sendmail.called
+
+ @mock.patch("smtplib.SMTP")
+ @mock.patch("ssl.create_default_context")
+ def test_send_mime_starttls_uses_configured_ssl_context(self,
create_default_context, mock_smtp):
+ with conf_vars({("smtp", "smtp_starttls"): "True", ("email",
"ssl_context"): "default"}):
+ emailer.send_mime_email("from", "to", MIMEMultipart())
+
mock_smtp.return_value.starttls.assert_called_once_with(context=create_default_context.return_value)
+
+ @mock.patch("smtplib.SMTP_SSL")
+ def test_send_mime_invalid_ssl_context_raises(self, mock_smtp_ssl):
+ with conf_vars({("smtp", "smtp_ssl"): "True", ("email",
"ssl_context"): "bogus"}):
+ with pytest.raises(RuntimeError, match="must be set to 'default'
or 'none'"):
+ emailer.send_mime_email("from", "to", MIMEMultipart())
+ assert not mock_smtp_ssl.called
+
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_failure_after_connect_is_not_retried(self, mock_smtp):
+ mock_smtp.return_value.login.side_effect =
SMTPAuthenticationError(535, b"bad credentials")
+ with pytest.raises(SMTPAuthenticationError):
+ emailer.send_mime_email("from", "to", MIMEMultipart())
+ assert mock_smtp.call_count == 1
+ assert not mock_smtp.return_value.sendmail.called
+
+ @mock.patch("smtplib.SMTP")
+ def test_send_mime_without_starttls(self, mock_smtp):
+ with conf_vars({("smtp", "smtp_starttls"): "False"}):
+ emailer.send_mime_email("from", "to", MIMEMultipart())
+ assert not mock_smtp.return_value.starttls.called
+ assert mock_smtp.return_value.sendmail.called
