Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-07-07 23:06:34 +01:00
Code Activity

Add notify entities to SMTP integration (#173557)

Browse Source
This commit is contained in:
Manu
2026-06-15 16:32:24 +02:00
committed by GitHub
parent d8f125dfe9
commit 2dc1870ecd
11 changed files with 688 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/smtp/__init__.py
+67 -6
View File
@@ -1,14 +1,43 @@
"""The smtp integration."""
import logging
from smtplib import SMTPAuthenticationError
from socket import gaierror
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME, CONF_RECIPIENT, Platform
from homeassistant.const import (
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_RECIPIENT,
CONF_SENDER,
CONF_TIMEOUT,
CONF_USERNAME,
CONF_VERIFY_SSL,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.helpers import discovery
from homeassistant.util.ssl import create_client_context
from .const import DOMAIN
from .const import (
CONF_ENCRYPTION,
CONF_SENDER_NAME,
CONF_SERVER,
DEFAULT_TIMEOUT,
DOMAIN,
)
from .helpers import SmtpClient
_LOGGER = logging.getLogger(__name__)
type SmtpConfigEntry = ConfigEntry[SmtpClient]
PLATFORMS: list[Platform] = [Platform.NOTIFY]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: SmtpConfigEntry) -> bool:
"""Set up SMTP from a config entry."""
hass.async_create_task(
@@ -27,17 +56,49 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
{},
)
)
client = SmtpClient(
server=entry.data[CONF_SERVER],
port=entry.data[CONF_PORT],
timeout=entry.options.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
sender=entry.data[CONF_SENDER],
encryption=entry.data[CONF_ENCRYPTION],
username=entry.data.get(CONF_USERNAME),
password=entry.data.get(CONF_PASSWORD),
sender_name=entry.data.get(CONF_SENDER_NAME),
verify_ssl=entry.data[CONF_VERIFY_SSL],
ssl_context=(
await hass.async_add_executor_job(create_client_context)
if entry.data[CONF_VERIFY_SSL]
else None
),
)
try:
await hass.async_add_executor_job(lambda: client.connect().quit())
except SMTPAuthenticationError as e:
raise ConfigEntryError(
translation_domain=DOMAIN,
translation_key="authentication_error",
) from e
except (gaierror, ConnectionRefusedError) as e:
_LOGGER.debug("Full exception:", exc_info=True)
raise ConfigEntryNotReady(
translation_domain=DOMAIN,
translation_key="connection_error",
) from e
entry.runtime_data = client
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
entry.async_on_unload(entry.add_update_listener(_async_update_listener))
return True
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
async def _async_update_listener(hass: HomeAssistant, entry: SmtpConfigEntry) -> None:
"""Handle update."""
hass.config_entries.async_schedule_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: SmtpConfigEntry) -> bool:
"""Unload a config entry."""
return True
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
homeassistant/components/smtp/config_flow.py
+3 -3
View File
@@ -10,7 +10,6 @@ import voluptuous as vol
from homeassistant.config_entries import (
SOURCE_USER,
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
ConfigSubentryData,
@@ -46,6 +45,7 @@ from homeassistant.helpers.selector import (
)
from homeassistant.util.ssl import create_client_context
from . import SmtpConfigEntry
from .const import (
CONF_ENCRYPTION,
CONF_SENDER_NAME,
@@ -120,14 +120,14 @@ class MailConfigFlow(ConfigFlow, domain=DOMAIN):
@classmethod
@callback
def async_get_supported_subentry_types(
cls, config_entry: ConfigEntry
cls, config_entry: SmtpConfigEntry
) -> dict[str, type[ConfigSubentryFlow]]:
"""Return subentries supported by this integration."""
return {SUBENTRY_TYPE_RECIPIENT: RecipientSubentryFlowHandler}
@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlowHandler:
def async_get_options_flow(config_entry: SmtpConfigEntry) -> OptionsFlowHandler:
"""Get the options flow for this handler."""
return OptionsFlowHandler()
homeassistant/components/smtp/helpers.py
+194
View File
@@ -0,0 +1,194 @@
"""Helpers for SMTP integration."""
from email.mime.application import MIMEApplication
from email.mime.image import MIMEImage
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import logging
import os
from pathlib import Path
import smtplib
import socket
import ssl
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from .const import ATTR_HTML, DOMAIN
_LOGGER = logging.getLogger(__name__)
class SmtpClient:
"""Mailer class."""
def __init__(
self,
server: str,
port: int,
timeout: int,
sender: str,
encryption: str,
username: str | None,
password: str | None,
sender_name: str | None,
verify_ssl: bool,
ssl_context: ssl.SSLContext | None,
) -> None:
"""Initialize the SMTP service."""
self._server = server
self._port = port
self._timeout = timeout
self._sender = sender
self.encryption = encryption
self.username = username
self.password = password
self._sender_name = sender_name
self._verify_ssl = verify_ssl
self.tries = 2
self._ssl_context = ssl_context
def connect(self) -> smtplib.SMTP_SSL | smtplib.SMTP:
"""Connect/authenticate to SMTP Server."""
mail: smtplib.SMTP_SSL | smtplib.SMTP
if self.encryption == "tls":
mail = smtplib.SMTP_SSL(
self._server,
self._port,
timeout=self._timeout,
context=self._ssl_context,
)
else:
mail = smtplib.SMTP(self._server, self._port, timeout=self._timeout)
mail.ehlo_or_helo_if_needed()
if self.encryption == "starttls":
mail.starttls(context=self._ssl_context)
mail.ehlo()
if self.username and self.password:
mail.login(self.username, self.password)
return mail
def connection_is_valid(self) -> bool:
"""Check for valid config, verify connectivity."""
server = None
try:
server = self.connect()
except socket.gaierror, ConnectionRefusedError:
_LOGGER.exception(
(
"SMTP server not found or refused connection (%s:%s). Please check"
" the IP address, hostname, and availability of your SMTP server"
),
self._server,
self._port,
)
except smtplib.SMTPAuthenticationError:
_LOGGER.exception(
"Login not possible. Please check your setting and/or your credentials"
)
return False
finally:
if server:
server.quit()
return True
def _build_text_msg(message: str) -> MIMEText:
"""Build plaintext email."""
_LOGGER.debug("Building plain text email")
return MIMEText(message)
def _attach_file(
hass: HomeAssistant, atch_name: str, content_id: str | None = None
) -> MIMEImage | MIMEApplication | None:
"""Create a message attachment.
If MIMEImage is successful and content_id is passed (HTML), add images in-line.
Otherwise add them as attachments.
"""
try:
file_path = Path(atch_name).parent
if os.path.exists(file_path) and not hass.config.is_allowed_path(
str(file_path)
):
allow_list = "allowlist_external_dirs"
file_name = os.path.basename(atch_name)
url = "https://www.home-assistant.io/docs/configuration/basic/"
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="remote_path_not_allowed",
translation_placeholders={
"allow_list": allow_list,
"file_path": str(file_path),
"file_name": file_name,
"url": url,
},
)
with open(atch_name, "rb") as attachment_file:
file_bytes = attachment_file.read()
except FileNotFoundError:
_LOGGER.warning("Attachment %s not found. Skipping", atch_name)
return None
attachment: MIMEImage | MIMEApplication
try:
attachment = MIMEImage(file_bytes)
except TypeError:
_LOGGER.warning(
"Attachment %s has an unknown MIME type. Falling back to file",
atch_name,
)
attachment = MIMEApplication(file_bytes, Name=os.path.basename(atch_name))
attachment["Content-Disposition"] = (
f'attachment; filename="{os.path.basename(atch_name)}"'
)
else:
if content_id:
attachment.add_header("Content-ID", f"<{content_id}>")
else:
attachment.add_header(
"Content-Disposition",
f"attachment; filename={os.path.basename(atch_name)}",
)
return attachment
def _build_multipart_msg(
hass: HomeAssistant, message: str, images: list[str]
) -> MIMEMultipart:
"""Build Multipart message with images as attachments."""
_LOGGER.debug("Building multipart email with image attachment(s)")
msg = MIMEMultipart()
body_txt = MIMEText(message)
msg.attach(body_txt)
for atch_name in images:
attachment = _attach_file(hass, atch_name)
if attachment:
msg.attach(attachment)
return msg
def _build_html_msg(
hass: HomeAssistant, text: str, html: str, images: list[str]
) -> MIMEMultipart:
"""Build Multipart message with in-line images and rich HTML (UTF-8)."""
_LOGGER.debug("Building HTML rich email")
msg = MIMEMultipart("related")
alternative = MIMEMultipart("alternative")
alternative.attach(MIMEText(text, _charset="utf-8"))
alternative.attach(MIMEText(html, ATTR_HTML, _charset="utf-8"))
msg.attach(alternative)
for atch_name in images:
name = os.path.basename(atch_name)
attachment = _attach_file(hass, atch_name, name)
if attachment:
msg.attach(attachment)
return msg
homeassistant/components/smtp/icons.json
+9
View File
@@ -0,0 +1,9 @@
{
"entity": {
"notify": {
"mailto": {
"default": "mdi:email-outline"
}
}
}
}
homeassistant/components/smtp/notify.py
+152 -199
View File
@@ -1,17 +1,19 @@
"""Mail (SMTP) notification service."""
from email.mime.application import MIMEApplication
from email.mime.image import MIMEImage
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import email.utils
import logging
import os
from pathlib import Path
import smtplib
import socket
import ssl
from typing import Any
from smtplib import (
SMTP,
SMTP_SSL,
SMTPAuthenticationError,
SMTPException,
SMTPServerDisconnected,
)
from socket import gaierror
from ssl import SSLContext
from typing import TYPE_CHECKING, Any
import voluptuous as vol
@@ -22,8 +24,10 @@ from homeassistant.components.notify import (
ATTR_TITLE_DEFAULT,
PLATFORM_SCHEMA as NOTIFY_PLATFORM_SCHEMA,
BaseNotificationService,
NotifyEntity,
NotifyEntityFeature,
)
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.config_entries import SOURCE_IMPORT, ConfigSubentry
from homeassistant.const import (
CONF_DEBUG,
CONF_PASSWORD,
@@ -37,12 +41,15 @@ from homeassistant.const import (
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers import config_validation as cv
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, entity_registry as er
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import dt as dt_util
from homeassistant.util.ssl import create_client_context
from . import SmtpConfigEntry
from .const import (
ATTR_HTML,
ATTR_IMAGES,
@@ -57,12 +64,15 @@ from .const import (
DOMAIN,
ENCRYPTION_OPTIONS,
)
from .helpers import SmtpClient, _build_html_msg, _build_multipart_msg, _build_text_msg
from .issue import async_deprecate_yaml_issue
PLATFORMS = [Platform.NOTIFY]
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 1
PLATFORM_SCHEMA = NOTIFY_PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_RECIPIENT): vol.All(cv.ensure_list, [vol.Email()]),
@@ -109,20 +119,7 @@ async def async_get_service(
if discovery_info[CONF_VERIFY_SSL]
else None
)
mail_service = MailNotificationService(
discovery_info[CONF_SERVER],
discovery_info[CONF_PORT],
discovery_info.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
discovery_info[CONF_SENDER],
discovery_info[CONF_ENCRYPTION],
discovery_info.get(CONF_USERNAME),
discovery_info.get(CONF_PASSWORD),
discovery_info[CONF_RECIPIENT],
discovery_info.get(CONF_SENDER_NAME),
DEFAULT_DEBUG,
discovery_info[CONF_VERIFY_SSL],
ssl_context,
)
mail_service = MailNotificationService(discovery_info, ssl_context)
if await hass.async_add_executor_job(mail_service.connection_is_valid):
return mail_service
@@ -130,86 +127,140 @@ async def async_get_service(
return None
class MailNotificationService(BaseNotificationService):
async def async_setup_entry(
hass: HomeAssistant,
config_entry: SmtpConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the notification entity platform."""
client = config_entry.runtime_data
async_add_entities(
[
MailNotifyEntity(config_entry, subentry, client)
for subentry in config_entry.subentries.values()
],
)
entity_registry = er.async_get(hass)
entity_entries = er.async_entries_for_config_entry(
entity_registry, config_entry.entry_id
)
current_recipients = {
subentry.unique_id for subentry in config_entry.subentries.values()
}
for entity in entity_entries:
if (
entity.unique_id.removeprefix(f"{config_entry.entry_id}_")
not in current_recipients
):
entity_registry.async_remove(entity.entity_id)
class MailNotifyEntity(NotifyEntity):
"""Representation of an SMTP notify entity."""
_attr_has_entity_name = True
_attr_translation_key = "mailto"
_attr_supported_features = NotifyEntityFeature.TITLE
def __init__(
self,
entry: SmtpConfigEntry,
subentry: ConfigSubentry,
client: SmtpClient,
) -> None:
"""Initialize the notify entity."""
self._entry = entry
self._subentry = subentry
self._client = client
self._attr_unique_id = f"{entry.entry_id}_{subentry.unique_id}"
self._attr_device_info = DeviceInfo(
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, entry.entry_id)},
)
self._attr_name = subentry.title
def send_message(self, message: str, title: str | None = None) -> None:
"""Send an email message via notify.send_message action."""
msg = MIMEText(message)
msg["Subject"] = title or ATTR_TITLE_DEFAULT
self._send_email(msg=msg)
def _send_email(self, msg: MIMEMultipart | MIMEText) -> None:
"""Send the message."""
if TYPE_CHECKING:
assert self._subentry.unique_id
msg["From"] = email.utils.formataddr(
(self._entry.data.get(CONF_SENDER_NAME), self._entry.data[CONF_SENDER])
)
msg["X-Mailer"] = "Home Assistant"
msg["Date"] = email.utils.format_datetime(dt_util.now())
msg["Message-Id"] = email.utils.make_msgid()
client: SMTP_SSL | SMTP | None = None
for attempt in range(self._client.tries):
try:
client = self._client.connect()
except SMTPAuthenticationError as e:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="authentication_error",
) from e
except (gaierror, ConnectionRefusedError, SMTPException) as e:
_LOGGER.debug("Full exception:", exc_info=True)
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="send_mail_connection_error",
) from e
try:
client.sendmail(
self._entry.data[CONF_SENDER],
self._subentry.unique_id,
msg.as_string(),
)
break
except SMTPException as e:
_LOGGER.debug(
"Error sending mail at attempt %s:", attempt + 1, exc_info=True
)
if attempt == self._client.tries - 1:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="send_mail_connection_error",
) from e
finally:
client.quit()
class MailNotificationService(SmtpClient, BaseNotificationService):
"""Implement the notification service for E-mail messages."""
def __init__(
self,
server: str,
port: int,
timeout: int,
sender: str,
encryption: str,
username: str | None,
password: str | None,
recipients: list[str],
sender_name: str | None,
debug: bool,
verify_ssl: bool,
ssl_context: ssl.SSLContext | None,
config: DiscoveryInfoType,
ssl_context: SSLContext | None,
) -> None:
"""Initialize the SMTP service."""
self._server = server
self._port = port
self._timeout = timeout
self._sender = sender
self.encryption = encryption
self.username = username
self.password = password
self.recipients = recipients
self._sender_name = sender_name
self.debug = debug
self._verify_ssl = verify_ssl
self.tries = 2
self._ssl_context = ssl_context
def connect(self) -> smtplib.SMTP_SSL | smtplib.SMTP:
"""Connect/authenticate to SMTP Server."""
mail: smtplib.SMTP_SSL | smtplib.SMTP
if self.encryption == "tls":
mail = smtplib.SMTP_SSL(
self._server,
self._port,
timeout=self._timeout,
context=self._ssl_context,
)
else:
mail = smtplib.SMTP(self._server, self._port, timeout=self._timeout)
mail.set_debuglevel(self.debug)
mail.ehlo_or_helo_if_needed()
if self.encryption == "starttls":
mail.starttls(context=self._ssl_context)
mail.ehlo()
if self.username and self.password:
mail.login(self.username, self.password)
return mail
def connection_is_valid(self) -> bool:
"""Check for valid config, verify connectivity."""
server = None
try:
server = self.connect()
except socket.gaierror, ConnectionRefusedError:
_LOGGER.exception(
(
"SMTP server not found or refused connection (%s:%s). Please check"
" the IP address, hostname, and availability of your SMTP server"
),
self._server,
self._port,
)
except smtplib.SMTPAuthenticationError:
_LOGGER.exception(
"Login not possible. Please check your setting and/or your credentials"
)
return False
finally:
if server:
server.quit()
return True
self.recipients = config[CONF_RECIPIENT]
super().__init__(
server=config[CONF_SERVER],
port=config[CONF_PORT],
timeout=config.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
sender=config[CONF_SENDER],
encryption=config[CONF_ENCRYPTION],
username=config.get(CONF_USERNAME),
password=config.get(CONF_PASSWORD),
sender_name=config.get(CONF_SENDER_NAME),
verify_ssl=config[CONF_VERIFY_SSL],
ssl_context=ssl_context,
)
def send_message(self, message: str, **kwargs: Any) -> None:
"""Build and send a message to a user.
@@ -261,112 +312,14 @@ class MailNotificationService(BaseNotificationService):
try:
mail.sendmail(self._sender, recipients, msg.as_string())
break
except smtplib.SMTPServerDisconnected:
except SMTPServerDisconnected:
_LOGGER.warning(
"SMTPServerDisconnected sending mail: retrying connection"
)
mail.quit()
mail = self.connect()
except smtplib.SMTPException:
except SMTPException:
_LOGGER.warning("SMTPException sending mail: retrying connection")
mail.quit()
mail = self.connect()
mail.quit()
def _build_text_msg(message: str) -> MIMEText:
"""Build plaintext email."""
_LOGGER.debug("Building plain text email")
return MIMEText(message)
def _attach_file(
hass: HomeAssistant, atch_name: str, content_id: str | None = None
) -> MIMEImage | MIMEApplication | None:
"""Create a message attachment.
If MIMEImage is successful and content_id is passed (HTML), add images in-line.
Otherwise add them as attachments.
"""
try:
file_path = Path(atch_name).parent
if os.path.exists(file_path) and not hass.config.is_allowed_path(
str(file_path)
):
allow_list = "allowlist_external_dirs"
file_name = os.path.basename(atch_name)
url = "https://www.home-assistant.io/docs/configuration/basic/"
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="remote_path_not_allowed",
translation_placeholders={
"allow_list": allow_list,
"file_path": str(file_path),
"file_name": file_name,
"url": url,
},
)
with open(atch_name, "rb") as attachment_file:
file_bytes = attachment_file.read()
except FileNotFoundError:
_LOGGER.warning("Attachment %s not found. Skipping", atch_name)
return None
attachment: MIMEImage | MIMEApplication
try:
attachment = MIMEImage(file_bytes)
except TypeError:
_LOGGER.warning(
"Attachment %s has an unknown MIME type. Falling back to file",
atch_name,
)
attachment = MIMEApplication(file_bytes, Name=os.path.basename(atch_name))
attachment["Content-Disposition"] = (
f'attachment; filename="{os.path.basename(atch_name)}"'
)
else:
if content_id:
attachment.add_header("Content-ID", f"<{content_id}>")
else:
attachment.add_header(
"Content-Disposition",
f"attachment; filename={os.path.basename(atch_name)}",
)
return attachment
def _build_multipart_msg(
hass: HomeAssistant, message: str, images: list[str]
) -> MIMEMultipart:
"""Build Multipart message with images as attachments."""
_LOGGER.debug("Building multipart email with image attachme_build_html_msgnt(s)")
msg = MIMEMultipart()
body_txt = MIMEText(message)
msg.attach(body_txt)
for atch_name in images:
attachment = _attach_file(hass, atch_name)
if attachment:
msg.attach(attachment)
return msg
def _build_html_msg(
hass: HomeAssistant, text: str, html: str, images: list[str]
) -> MIMEMultipart:
"""Build Multipart message with in-line images and rich HTML (UTF-8)."""
_LOGGER.debug("Building HTML rich email")
msg = MIMEMultipart("related")
alternative = MIMEMultipart("alternative")
alternative.attach(MIMEText(text, _charset="utf-8"))
alternative.attach(MIMEText(html, ATTR_HTML, _charset="utf-8"))
msg.attach(alternative)
for atch_name in images:
name = os.path.basename(atch_name)
attachment = _attach_file(hass, atch_name, name)
if attachment:
msg.attach(attachment)
return msg
homeassistant/components/smtp/strings.json
+9
View File
@@ -84,8 +84,17 @@
}
},
"exceptions": {
"authentication_error": {
"message": "Failed to authenticate with SMTP server. Please verify your credentials."
},
"connection_error": {
"message": "Failed to connect to SMTP server."
},
"remote_path_not_allowed": {
"message": "Cannot send email with attachment \"{file_name}\" from directory \"{file_path}\" which is not secure to load data from. Only folders added to `{allow_list}` are accessible. See {url} for more information."
},
"send_mail_connection_error": {
"message": "Failed to send email message due to a connection error."
}
},
"issues": {
tests/components/smtp/conftest.py
+12 -1
View File
@@ -40,7 +40,7 @@ def mock_smtp() -> Generator[MagicMock]:
with (
patch(
"homeassistant.components.smtp.notify.smtplib.SMTP", autospec=True
"homeassistant.components.smtp.helpers.smtplib.SMTP", autospec=True
) as mock_client,
patch("homeassistant.components.smtp.config_flow.SMTP", new=mock_client),
):
@@ -48,6 +48,17 @@ def mock_smtp() -> Generator[MagicMock]:
yield client
@pytest.fixture(name="make_msgid")
def mock_make_msgid() -> Generator[None]:
"""Mock email.utils.make_msgid."""
with patch(
"homeassistant.components.smtp.notify.email.utils.make_msgid",
return_value="<177777777700.12345.12345678901234567890@mock>",
):
yield
@pytest.fixture(name="smtp_ssl")
def mock_smtp_ssl() -> Generator[MagicMock]:
"""Mock SMTP."""
tests/components/smtp/snapshots/test_notify.ambr
+52
View File
@@ -0,0 +1,52 @@
# serializer version: 1
# name: test_notify_platform[notify.home_assistant_recipient-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'notify',
'entity_category': None,
'entity_id': 'notify.home_assistant_recipient',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Recipient',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Recipient',
'platform': 'smtp',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <NotifyEntityFeature: 1>,
'translation_key': 'mailto',
'unique_id': '123456789_recipient@example.com',
'unit_of_measurement': None,
})
# ---
# name: test_notify_platform[notify.home_assistant_recipient-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Home Assistant Recipient',
'supported_features': <NotifyEntityFeature: 1>,
}),
'context': <ANY>,
'entity_id': 'notify.home_assistant_recipient',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
tests/components/smtp/test_config_flow.py
+1
View File
@@ -224,6 +224,7 @@ async def test_form_recipient_already_configured(
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("smtp")
async def test_options_flow(
hass: HomeAssistant,
config_entry: MockConfigEntry,
tests/components/smtp/test_init.py
+28
View File
@@ -1,5 +1,7 @@
"""Tests for the SMTP integration."""
from smtplib import SMTPAuthenticationError
from socket import gaierror
from unittest.mock import AsyncMock, MagicMock
import pytest
@@ -48,6 +50,32 @@ async def test_entry_setup_unload(
assert config_entry.state is ConfigEntryState.NOT_LOADED
@pytest.mark.parametrize(
("exception", "state"),
[
(ConnectionRefusedError, ConfigEntryState.SETUP_RETRY),
(gaierror, ConfigEntryState.SETUP_RETRY),
(SMTPAuthenticationError(0, ""), ConfigEntryState.SETUP_ERROR),
],
)
async def test_config_entry_not_ready(
hass: HomeAssistant,
config_entry: MockConfigEntry,
smtp: MagicMock,
exception: Exception,
state: ConfigEntryState,
) -> None:
"""Test config entry not ready."""
smtp.login.side_effect = exception
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is state
@pytest.mark.usefixtures("smtp")
async def test_import(
hass: HomeAssistant,
tests/components/smtp/test_notify.py
+161 -15
View File
@@ -2,16 +2,49 @@
from pathlib import Path
import re
from unittest.mock import patch
from smtplib import (
SMTPAuthenticationError,
SMTPHeloError,
SMTPSenderRefused,
SMTPServerDisconnected,
)
from socket import gaierror
from unittest.mock import MagicMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.smtp.const import DOMAIN
from homeassistant.components.notify import (
ATTR_MESSAGE,
DOMAIN as NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
)
from homeassistant.components.smtp.const import (
CONF_ENCRYPTION,
CONF_SENDER_NAME,
CONF_SERVER,
DOMAIN,
)
from homeassistant.components.smtp.notify import MailNotificationService
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
ATTR_ENTITY_ID,
CONF_PASSWORD,
CONF_PORT,
CONF_RECIPIENT,
CONF_SENDER,
CONF_TIMEOUT,
CONF_USERNAME,
CONF_VERIFY_SSL,
STATE_UNKNOWN,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_registry as er
from homeassistant.util.ssl import create_client_context
from tests.common import MockConfigEntry, snapshot_platform
class MockSMTP(MailNotificationService):
"""Test SMTP object that doesn't need a working server."""
@@ -25,18 +58,19 @@ class MockSMTP(MailNotificationService):
def message():
"""Return MockSMTP object with test data."""
return MockSMTP(
"localhost",
25,
5,
"test@test.com",
1,
"testuser",
"testpass",
["recip1@example.com", "testrecip@test.com"],
"Home Assistant",
0,
True,
create_client_context(),
config={
CONF_SERVER: "localhost",
CONF_PORT: 25,
CONF_TIMEOUT: 5,
CONF_SENDER: "test@test.com",
CONF_ENCRYPTION: 1,
CONF_USERNAME: "testuser",
CONF_PASSWORD: "testpass",
CONF_RECIPIENT: ["recip1@example.com", "testrecip@test.com"],
CONF_SENDER_NAME: "Home Assistant",
CONF_VERIFY_SSL: True,
},
ssl_context=create_client_context(),
)
@@ -182,3 +216,115 @@ def test_send_target_message(target, hass: HomeAssistant, message) -> None:
_, recipient = message.send_message(message_data, target=target)
assert recipient == expected_recipient
@pytest.mark.usefixtures("smtp")
async def test_notify_platform(
hass: HomeAssistant,
config_entry: MockConfigEntry,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
) -> None:
"""Test setup of the SMTP notify platform."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
await snapshot_platform(hass, entity_registry, snapshot, config_entry.entry_id)
@pytest.mark.usefixtures("make_msgid")
@pytest.mark.freeze_time("2026-05-03T03:09:37+00:00")
async def test_notify_send_message(
hass: HomeAssistant,
config_entry: MockConfigEntry,
smtp: MagicMock,
) -> None:
"""Test sending an email message via notify.send_message action."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
state = hass.states.get("notify.home_assistant_recipient")
assert state
assert state.state == STATE_UNKNOWN
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
{
ATTR_ENTITY_ID: "notify.home_assistant_recipient",
ATTR_MESSAGE: "Hello World",
},
blocking=True,
)
state = hass.states.get("notify.home_assistant_recipient")
assert state
assert state.state == "2026-05-03T03:09:37+00:00"
smtp.sendmail.assert_called_once_with(
"email@example.com",
"recipient@example.com",
(
'Content-Type: text/plain; charset="us-ascii"\n'
"MIME-Version: 1.0\n"
"Content-Transfer-Encoding: 7bit\n"
"Subject: Home Assistant\n"
"From: Home Assistant <email@example.com>\n"
"X-Mailer: Home Assistant\n"
"Date: Sat, 02 May 2026 20:09:37 -0700\n"
"Message-Id: <177777777700.12345.12345678901234567890@mock>\n\n"
"Hello World"
),
)
@pytest.mark.parametrize(
("call_method", "exception", "translation_key"),
[
("login", SMTPAuthenticationError(0, ""), "authentication_error"),
("login", gaierror, "send_mail_connection_error"),
("login", ConnectionRefusedError, "send_mail_connection_error"),
("login", SMTPHeloError(0, ""), "send_mail_connection_error"),
("sendmail", SMTPServerDisconnected, "send_mail_connection_error"),
("sendmail", SMTPSenderRefused(0, b"", ""), "send_mail_connection_error"),
],
)
@pytest.mark.freeze_time("2026-05-03T03:09:37+00:00")
async def test_notify_send_message_exceptions(
hass: HomeAssistant,
config_entry: MockConfigEntry,
smtp: MagicMock,
call_method: str,
exception: Exception,
translation_key: str,
) -> None:
"""Test exceptions via notify.send_message action."""
config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.LOADED
getattr(smtp, call_method).side_effect = exception
with pytest.raises(HomeAssistantError) as e:
await hass.services.async_call(
NOTIFY_DOMAIN,
SERVICE_SEND_MESSAGE,
{
ATTR_ENTITY_ID: "notify.home_assistant_recipient",
ATTR_MESSAGE: "Hello World",
},
blocking=True,
)
assert e.value.translation_key == translation_key