jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 01:08:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Set up smtp integration via the UI (#110817)

Browse Source 
* Set up smtp integration via the UI

* Update homeassistant/components/smtp/config_flow.py

Co-authored-by: Sid <27780930+autinerd@users.noreply.github.com>

* Update homeassistant/components/smtp/notify.py

Co-authored-by: Jan-Philipp Benecke <jan-philipp@bnck.me>

* Update homeassistant/components/smtp/notify.py

Co-authored-by: Jan-Philipp Benecke <jan-philipp@bnck.me>

* Update homeassistant/components/smtp/notify.py

Co-authored-by: Jan-Philipp Benecke <jan-philipp@bnck.me>

* Update homeassistant/components/smtp/notify.py

Co-authored-by: Jan-Philipp Benecke <jan-philipp@bnck.me>

* ruff

---------

Co-authored-by: Sid <27780930+autinerd@users.noreply.github.com>
Co-authored-by: Jan-Philipp Benecke <jan-philipp@bnck.me>
This commit is contained in:
Jan Bouwhuis 2024-02-18 13:20:45 +01:00 committed by GitHub
parent e2ab44903c
commit 66a31407f9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 652 additions and 105 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

99
homeassistant/components/smtp/__init__.py
View File

@ -1 +1,98 @@
"""The smtp component."""
"""Set up the smtp component."""
from __future__ import annotations
import logging
import voluptuous as vol
from homeassistant.components.notify import DOMAIN as NOTIFY_DOMAIN, PLATFORM_SCHEMA
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_PASSWORD,
CONF_PORT,
CONF_RECIPIENT,
CONF_SENDER,
CONF_TIMEOUT,
CONF_USERNAME,
CONF_VERIFY_SSL,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import discovery
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_DEBUG,
CONF_ENCRYPTION,
CONF_SENDER_NAME,
CONF_SERVER,
DEFAULT_DEBUG,
DEFAULT_ENCRYPTION,
DEFAULT_HOST,
DEFAULT_PORT,
DEFAULT_TIMEOUT,
DOMAIN,
ENCRYPTION_OPTIONS,
)
from .notify import MailNotificationService
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_RECIPIENT): vol.All(cv.ensure_list, [vol.Email()]),
vol.Required(CONF_SENDER): vol.Email(),
vol.Optional(CONF_SERVER, default=DEFAULT_HOST): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
vol.Optional(CONF_ENCRYPTION, default=DEFAULT_ENCRYPTION): vol.In(
ENCRYPTION_OPTIONS
),
vol.Optional(CONF_USERNAME): cv.string,
vol.Optional(CONF_PASSWORD): cv.string,
vol.Optional(CONF_SENDER_NAME): cv.string,
vol.Optional(CONF_DEBUG, default=DEFAULT_DEBUG): cv.boolean,
vol.Optional(CONF_VERIFY_SSL, default=True): cv.boolean,
}
)
PLATFORMS = [Platform.NOTIFY]
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the SMTP notify component from configuration.yaml."""
hass.data.setdefault(DOMAIN, {})["hass_config"] = config
if NOTIFY_DOMAIN in config:
for platform_config in config[NOTIFY_DOMAIN]:
hass.async_create_task(
hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=platform_config
)
)
return True
async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Set up SMTP config entry."""
hass_config: ConfigType = hass.data[DOMAIN]["hass_config"]
hass.data[DOMAIN][config_entry.entry_id] = None
config = dict(config_entry.data) | {"entry_id": config_entry.entry_id}
discovery.load_platform(hass, Platform.NOTIFY.value, DOMAIN, config, hass_config)
return True
async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:
"""Unload SMTP config entry."""
mail_service: MailNotificationService | None = hass.data[DOMAIN][
config_entry.entry_id
]
if mail_service is not None:
await mail_service.async_unregister_services()
return True

197
homeassistant/components/smtp/config_flow.py Normal file
View File

@ -0,0 +1,197 @@
"""Config flow for SMTP integration."""
from __future__ import annotations
import logging
from typing import Any
import voluptuous as vol
from homeassistant.config_entries import ConfigFlow
from homeassistant.const import (
CONF_NAME,
CONF_PASSWORD,
CONF_PLATFORM,
CONF_PORT,
CONF_RECIPIENT,
CONF_SENDER,
CONF_TIMEOUT,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN
from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.selector import (
BooleanSelector,
BooleanSelectorConfig,
NumberSelector,
NumberSelectorConfig,
NumberSelectorMode,
SelectSelector,
SelectSelectorConfig,
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from .const import (
CONF_DEBUG,
CONF_ENCRYPTION,
CONF_SENDER_NAME,
CONF_SERVER,
DEFAULT_DEBUG,
DEFAULT_ENCRYPTION,
DEFAULT_HOST,
DEFAULT_PORT,
DEFAULT_TIMEOUT,
DOMAIN,
ENCRYPTION_OPTIONS,
)
from .notify import MailNotificationService
_LOGGER = logging.getLogger(__name__)
BASE_DATA_FIELDS = {
vol.Optional(CONF_NAME): TextSelector(
TextSelectorConfig(type=TextSelectorType.TEXT)
),
vol.Required(CONF_RECIPIENT): SelectSelector(
SelectSelectorConfig(options=[], multiple=True, custom_value=True)
),
vol.Required(CONF_SENDER): TextSelector(
TextSelectorConfig(type=TextSelectorType.EMAIL)
),
vol.Required(CONF_SERVER, default=DEFAULT_HOST): TextSelector(
TextSelectorConfig(type=TextSelectorType.TEXT)
),
vol.Required(CONF_PORT, default=DEFAULT_PORT): NumberSelector(
NumberSelectorConfig(min=1, max=65535, mode=NumberSelectorMode.BOX, step=1)
),
vol.Optional(CONF_ENCRYPTION, default=DEFAULT_ENCRYPTION): SelectSelector(
SelectSelectorConfig(options=ENCRYPTION_OPTIONS, translation_key="encryption")
),
vol.Optional(CONF_USERNAME): TextSelector(
TextSelectorConfig(type=TextSelectorType.TEXT)
),
vol.Optional(CONF_PASSWORD): TextSelector(
TextSelectorConfig(type=TextSelectorType.PASSWORD)
),
vol.Optional(CONF_SENDER_NAME): TextSelector(
TextSelectorConfig(type=TextSelectorType.TEXT)
),
}
ADVANCED_DATA_FIELDS = {
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): NumberSelector(
NumberSelectorConfig(min=1, max=60, mode=NumberSelectorMode.BOX)
),
vol.Optional(CONF_DEBUG, default=DEFAULT_DEBUG): BooleanSelector(
BooleanSelectorConfig()
),
vol.Optional(CONF_VERIFY_SSL, default=True): BooleanSelector(
BooleanSelectorConfig()
),
}
CONFIG_SCHEMA = vol.Schema(BASE_DATA_FIELDS | ADVANCED_DATA_FIELDS)
UNIQUE_ENTRY_KEYS = [CONF_RECIPIENT, CONF_SENDER, CONF_SERVER, CONF_NAME]
def validate_smtp_settings(settings: dict[str, Any], errors: dict[str, str]) -> None:
"""Validate SMTP connection settings."""
try:
for recipient in cv.ensure_list(settings[CONF_RECIPIENT]):
vol.Email()(recipient)
except vol.Invalid:
errors[CONF_RECIPIENT] = "invalid_email_address"
try:
vol.Email()(settings[CONF_SENDER])
except vol.Invalid:
errors[CONF_SENDER] = "invalid_email_address"
if settings.get(CONF_USERNAME) and not settings.get(CONF_PASSWORD):
errors[CONF_PASSWORD] = "username_and_password"
if settings.get(CONF_PASSWORD) and not settings.get(CONF_USERNAME):
errors[CONF_USERNAME] = "username_and_password"
if errors:
return
settings[CONF_PORT] = cv.positive_int(settings[CONF_PORT])
settings[CONF_TIMEOUT] = cv.positive_int(settings[CONF_TIMEOUT])
service_class = MailNotificationService(
settings[CONF_SERVER],
settings[CONF_PORT],
settings.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
settings[CONF_SENDER],
settings[CONF_ENCRYPTION],
settings.get(CONF_USERNAME),
settings.get(CONF_PASSWORD),
settings[CONF_RECIPIENT],
settings.get(CONF_SENDER_NAME),
settings.get(CONF_DEBUG, DEFAULT_DEBUG),
settings.get(CONF_VERIFY_SSL, True),
)
service_class.connection_is_valid(errors=errors)
class SMTPConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a config flow for SMTP."""
VERSION = 1
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step."""
errors: dict[str, str] = {}
if user_input is not None:
self._async_abort_entries_match(
{key: user_input[key] for key in UNIQUE_ENTRY_KEYS if key in user_input}
)
data: dict[str, Any] = CONFIG_SCHEMA(user_input)
await self.hass.async_add_executor_job(validate_smtp_settings, data, errors)
if not errors:
name = data.get(CONF_NAME, "SMTP")
return self.async_create_entry(title=name, data=data)
fields = BASE_DATA_FIELDS
if self.show_advanced_options:
fields |= ADVANCED_DATA_FIELDS
schema = self.add_suggested_values_to_schema(vol.Schema(fields), user_input)
return self.async_show_form(
step_id="user",
data_schema=schema,
errors=errors,
)
async def async_step_import(self, config: dict[str, Any]) -> FlowResult:
"""Import configuration from yaml."""
async_create_issue(
self.hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2023.9.0",
is_fixable=False,
is_persistent=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "SMTP",
},
)
self._async_abort_entries_match(
{key: config[key] for key in UNIQUE_ENTRY_KEYS if key in config}
)
config.pop(CONF_PLATFORM)
config[CONF_RECIPIENT] = cv.ensure_list(config[CONF_RECIPIENT])
config = CONFIG_SCHEMA(config)
return self.async_create_entry(
title=config.get(CONF_NAME, "SMTP"),
data=config,
)

1
homeassistant/components/smtp/manifest.json
View File

@ -2,6 +2,7 @@
"domain": "smtp",
"name": "SMTP",
"codeowners": [],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/smtp",
"iot_class": "cloud_push"
}

98
homeassistant/components/smtp/notify.py
View File

@ -10,15 +10,13 @@ import logging
import os
from pathlib import Path
import smtplib
import voluptuous as vol
import socket
from homeassistant.components.notify import (
ATTR_DATA,
ATTR_TARGET,
ATTR_TITLE,
ATTR_TITLE_DEFAULT,
PLATFORM_SCHEMA,
BaseNotificationService,
)
from homeassistant.const import (
@ -29,12 +27,9 @@ from homeassistant.const import (
CONF_TIMEOUT,
CONF_USERNAME,
CONF_VERIFY_SSL,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.reload import setup_reload_service
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
import homeassistant.util.dt as dt_util
from homeassistant.util.ssl import client_context
@ -52,31 +47,10 @@ from .const import (
DEFAULT_PORT,
DEFAULT_TIMEOUT,
DOMAIN,
ENCRYPTION_OPTIONS,
)
PLATFORMS = [Platform.NOTIFY]
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_RECIPIENT): vol.All(cv.ensure_list, [vol.Email()]),
vol.Required(CONF_SENDER): vol.Email(),
vol.Optional(CONF_SERVER, default=DEFAULT_HOST): cv.string,
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_TIMEOUT, default=DEFAULT_TIMEOUT): cv.positive_int,
vol.Optional(CONF_ENCRYPTION, default=DEFAULT_ENCRYPTION): vol.In(
ENCRYPTION_OPTIONS
),
vol.Optional(CONF_USERNAME): cv.string,
vol.Optional(CONF_PASSWORD): cv.string,
vol.Optional(CONF_SENDER_NAME): cv.string,
vol.Optional(CONF_DEBUG, default=DEFAULT_DEBUG): cv.boolean,
vol.Optional(CONF_VERIFY_SSL, default=True): cv.boolean,
}
)
def get_service(
hass: HomeAssistant,
@ -84,21 +58,30 @@ def get_service(
discovery_info: DiscoveryInfoType | None = None,
) -> MailNotificationService | None:
"""Get the mail notification service."""
setup_reload_service(hass, DOMAIN, PLATFORMS)
if discovery_info is None:
_LOGGER.warning(
"The notify platform setup the smtp integration via configuration.yaml "
"is deprecated. Your config has been migrated to a config entry and "
"should be removed from your configuration.yaml. "
"Canceling setup via configuration.yaml"
)
return None
entry_id = discovery_info["entry_id"]
mail_service = MailNotificationService(
config[CONF_SERVER],
config[CONF_PORT],
config[CONF_TIMEOUT],
config[CONF_SENDER],
config[CONF_ENCRYPTION],
config.get(CONF_USERNAME),
config.get(CONF_PASSWORD),
config[CONF_RECIPIENT],
config.get(CONF_SENDER_NAME),
config[CONF_DEBUG],
config[CONF_VERIFY_SSL],
discovery_info.get(CONF_SERVER, DEFAULT_HOST),
discovery_info.get(CONF_PORT, DEFAULT_PORT),
discovery_info.get(CONF_TIMEOUT, DEFAULT_TIMEOUT),
discovery_info[CONF_SENDER],
discovery_info.get(CONF_ENCRYPTION, DEFAULT_ENCRYPTION),
discovery_info.get(CONF_USERNAME),
discovery_info.get(CONF_PASSWORD),
discovery_info[CONF_RECIPIENT],
discovery_info.get(CONF_SENDER_NAME),
discovery_info.get(CONF_DEBUG, DEFAULT_DEBUG),
discovery_info.get(CONF_VERIFY_SSL, True),
)
hass.data[DOMAIN][entry_id] = mail_service
if mail_service.connection_is_valid():
return mail_service
@ -121,7 +104,7 @@ class MailNotificationService(BaseNotificationService):
sender_name,
debug,
verify_ssl,
):
) -> None:
"""Initialize the SMTP service."""
self._server = server
self._port = port
@ -157,25 +140,32 @@ class MailNotificationService(BaseNotificationService):
mail.login(self.username, self.password)
return mail
def connection_is_valid(self):
def connection_is_valid(self, errors: dict[str, str] | None = None) -> bool:
"""Check for valid config, verify connectivity."""
server = None
try:
server = self.connect()
except (smtplib.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"
)
if errors is None:
_LOGGER.exception(
"Login not possible. Please check your setting and/or your credentials"
)
else:
errors["base"] = "authentication_failed"
return False
except (socket.gaierror, ConnectionRefusedError, OSError):
if errors is None:
_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,
)
else:
errors["base"] = "connection_refused"
return False
finally:

1
homeassistant/components/smtp/services.yaml
View File

@ -1 +0,0 @@
reload:

52
homeassistant/components/smtp/strings.json
View File

@ -1,8 +1,52 @@
{
"services": {
"reload": {
"name": "[%key:common::action::reload%]",
"description": "Reloads smtp notify services."
"config": {
"step": {
"user": {
"data": {
"name": "Name",
"recipient": "Recipients",
"sender": "Sender",
"sender_name": "Sender name",
"server": "[%key:common::config_flow::data::host%]",
"port": "[%key:common::config_flow::data::port%]",
"timeout": "Request Timeout (seconds)",
"encryption": "Encryption",
"username": "[%key:common::config_flow::data::username%]",
"password": "[%key:common::config_flow::data::password%]",
"debug": "Enable SMTP debug",
"verify_ssl": "[%key:common::config_flow::data::verify_ssl%]"
},
"data_description": {
"name": "When set it will create a specific notify service with this name, defaults to `notify.notify`",
"recipient": "The email addres(ses) of the receipient(s) that should receive an email notification",
"sender": "The email address of the sender",
"sender_name": "The name of the sender",
"server": "The SMTP server hostname or IP-adress",
"port": "The port of the SMTP server",
"encryption": "The encryption type of the email server.",
"username": "The username for authentication to the SMTP server",
"password": "The password for authentication to the SMTP server"
}
}
},
"abort": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"already_configured": "[%key:common::config_flow::abort::already_configured_service%]"
},
"error": {
"authentication_failed": "Login not possible. Please check your setting and/or your credentials",
"connection_refused": "SMTP server not found or refused connection. Please check the IP address, hostname, and availability of your SMTP server",
"invalid_email_address": "Invalid email address specified",
"username_and_password": "Username and password should be configured together"
}
},
"selector": {
"encryption": {
"options": {
"tls": "TLS (usually over port 465)",
"starttls": "STARTTLS (usually over port 587 or 25)",
"none": "No encryption (usually over port 25)"
}
}
},
"exceptions": {

1
homeassistant/generated/config_flows.py
View File

@ -474,6 +474,7 @@ FLOWS = {
"smarttub",
"smhi",
"sms",
"smtp",
"snapcast",
"snooz",
"solaredge",

2
homeassistant/generated/integrations.json
View File

@ -5461,7 +5461,7 @@
"smtp": {
"name": "SMTP",
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "cloud_push"
},
"snapcast": {

15
tests/components/smtp/conftest.py Normal file
View File

@ -0,0 +1,15 @@
"""Fixtures for smtp tests."""
from collections.abc import Generator
from unittest.mock import AsyncMock, patch
import pytest
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock, None, None]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.smtp.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry

36
tests/components/smtp/const.py Normal file
View File

@ -0,0 +1,36 @@
""""Shared constants for SMTP tests."""
from homeassistant.components.smtp.const import DOMAIN
MOCKED_CONFIG_ENTRY_DATA = {
"name": DOMAIN,
"recipient": ["test@example.com"],
"sender": "test@example.com",
"server": "localhost",
"port": 587,
"encryption": "starttls",
"debug": False,
"verify_ssl": True,
"timeout": 5,
}
MOCKED_USER_ADVANCED_DATA = {
"name": DOMAIN,
"recipient": ["test@example.com"],
"sender": "test@example.com",
"server": "localhost",
"port": 587,
"encryption": "starttls",
"debug": False,
"verify_ssl": True,
"timeout": 5,
}
MOCKED_USER_BASIC_DATA = {
"name": DOMAIN,
"recipient": ["test@example.com"],
"sender": "test@example.com",
"server": "localhost",
"port": 587,
"encryption": "starttls",
}

178
tests/components/smtp/test_config_flow.py Normal file
View File

@ -0,0 +1,178 @@
"""Tests for the SMTP config flow."""
from copy import deepcopy
from typing import Any
from unittest.mock import AsyncMock, patch
import pytest
from homeassistant import config_entries
import homeassistant.components.notify as notify
from homeassistant.components.smtp.const import DOMAIN
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.setup import async_setup_component
from .const import (
MOCKED_CONFIG_ENTRY_DATA,
MOCKED_USER_ADVANCED_DATA,
MOCKED_USER_BASIC_DATA,
)
from tests.common import MockConfigEntry
@patch(
"homeassistant.components.smtp.notify.MailNotificationService.connection_is_valid",
lambda x: True,
)
async def test_import_entry(hass: HomeAssistant) -> None:
"""Test import of a confif entry from yaml."""
assert await async_setup_component(
hass,
notify.DOMAIN,
{
notify.DOMAIN: [
{
"name": DOMAIN,
"platform": DOMAIN,
"recipient": "test@example.com",
"sender": "test@example.com",
},
]
},
)
# Wait for discovery to finish
await hass.async_block_till_done()
assert hass.services.has_service(notify.DOMAIN, DOMAIN)
async def test_no_import(hass: HomeAssistant) -> None:
"""Test platform setup without config succeeds."""
assert await async_setup_component(hass, DOMAIN, {DOMAIN: {}})
@pytest.mark.parametrize(
("user_input", "advanced_settings"),
[(MOCKED_USER_BASIC_DATA, False), (MOCKED_USER_ADVANCED_DATA, True)],
)
async def test_form(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
user_input: dict[str, Any],
advanced_settings: bool,
) -> None:
"""Test we get the form."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={
"source": config_entries.SOURCE_USER,
"show_advanced_options": advanced_settings,
},
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {}
def _connection_is_valid(errors: dict[str, str] | None = None) -> bool:
"""Check for valid config, verify connectivity."""
return True
with patch(
"homeassistant.components.smtp.notify.MailNotificationService.connection_is_valid",
side_effect=_connection_is_valid,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "smtp"
assert result2["data"] == MOCKED_CONFIG_ENTRY_DATA
assert len(mock_setup_entry.mock_calls) == 1
@pytest.mark.parametrize(
("mocked_user_input", "errors"),
[
(
{"recipient": ["test@example.com", "not_a_valid_email"]},
{"recipient": "invalid_email_address"},
),
({"sender": "not_a_valid_email"}, {"sender": "invalid_email_address"}),
({"username": "someuser"}, {"password": "username_and_password"}),
({"password": "somepassword"}, {"username": "username_and_password"}),
],
)
async def test_invalid_form(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
mocked_user_input: dict[str, Any],
errors: dict[str, str],
) -> None:
"""Test form validation works."""
user_input = deepcopy(MOCKED_USER_BASIC_DATA)
user_input.update(mocked_user_input)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
)
assert result["type"] == FlowResultType.FORM
assert result["errors"] == {}
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], user_input
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == errors
assert len(mock_setup_entry.mock_calls) == 0
async def test_entry_already_configured(hass: HomeAssistant) -> None:
"""Test aborting if the entry is already configured."""
entry = MockConfigEntry(domain=DOMAIN, data=MOCKED_CONFIG_ENTRY_DATA)
entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
assert result["type"] == FlowResultType.FORM
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
MOCKED_USER_BASIC_DATA,
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.ABORT
assert result2["reason"] == "already_configured"
@pytest.mark.parametrize("error", ["authentication_failed", "connection_refused"])
async def test_form_invalid_auth_or_connection_refused(
hass: HomeAssistant, error: str
) -> None:
"""Test we handle invalid auth."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
def _connection_is_valid(errors: dict[str, str] | None = None) -> bool:
"""Check for valid config, verify connectivity."""
errors["base"] = error
return False
with patch(
"homeassistant.components.smtp.notify.MailNotificationService.connection_is_valid",
side_effect=_connection_is_valid,
):
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"], MOCKED_USER_BASIC_DATA
)
assert result2["type"] == FlowResultType.FORM
assert result2["errors"] == {
"base": error,
}

77
tests/components/smtp/test_notify.py
View File

@ -1,20 +1,20 @@
"""The tests for the notify smtp platform."""
from copy import deepcopy
from pathlib import Path
import re
from unittest.mock import patch
import pytest
from homeassistant import config as hass_config
import homeassistant.components.notify as notify
from homeassistant.components.smtp.const import DOMAIN
from homeassistant.components.smtp.notify import MailNotificationService
from homeassistant.const import SERVICE_RELOAD
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError
from homeassistant.setup import async_setup_component
from tests.common import get_fixture_path
from .const import MOCKED_CONFIG_ENTRY_DATA
from tests.common import MockConfigEntry
class MockSMTP(MailNotificationService):
@ -25,46 +25,6 @@ class MockSMTP(MailNotificationService):
return msg.as_string(), recipients
async def test_reload_notify(hass: HomeAssistant) -> None:
"""Verify we can reload the notify service."""
with patch(
"homeassistant.components.smtp.notify.MailNotificationService.connection_is_valid"
):
assert await async_setup_component(
hass,
notify.DOMAIN,
{
notify.DOMAIN: [
{
"name": DOMAIN,
"platform": DOMAIN,
"recipient": "test@example.com",
"sender": "test@example.com",
},
]
},
)
await hass.async_block_till_done()
assert hass.services.has_service(notify.DOMAIN, DOMAIN)
yaml_path = get_fixture_path("configuration.yaml", "smtp")
with patch.object(hass_config, "YAML_CONFIG_FILE", yaml_path), patch(
"homeassistant.components.smtp.notify.MailNotificationService.connection_is_valid"
):
await hass.services.async_call(
DOMAIN,
SERVICE_RELOAD,
{},
blocking=True,
)
await hass.async_block_till_done()
assert not hass.services.has_service(notify.DOMAIN, DOMAIN)
assert hass.services.has_service(notify.DOMAIN, "smtp_reloaded")
@pytest.fixture
def message():
"""Return MockSMTP object with test data."""
@ -123,6 +83,35 @@ EMAIL_DATA = [
]
@patch(
"homeassistant.components.smtp.notify.MailNotificationService.connection_is_valid",
lambda x: True,
)
async def test_reload_smtp(hass: HomeAssistant) -> None:
"""Verify we can reload a smtp config entry."""
data = deepcopy(MOCKED_CONFIG_ENTRY_DATA)
entry = MockConfigEntry(domain=DOMAIN, data=data)
entry.add_to_hass(hass)
await hass.config_entries.async_setup(entry.entry_id)
# Wait for discovery to finish
await hass.async_block_till_done()
assert hass.services.has_service(notify.DOMAIN, DOMAIN)
await hass.config_entries.async_reload(entry.entry_id)
assert not hass.services.has_service(notify.DOMAIN, DOMAIN)
await hass.async_block_till_done()
# Wait for discovery to finish
assert hass.services.has_service(notify.DOMAIN, DOMAIN)
# Unloading the entry should remove the service
await hass.config_entries.async_unload(entry.entry_id)
assert not hass.services.has_service(notify.DOMAIN, DOMAIN)
await hass.config_entries.async_setup(entry.entry_id)
# Wait for discovery to finish
await hass.async_block_till_done()
assert hass.services.has_service(notify.DOMAIN, DOMAIN)
@pytest.mark.parametrize(
("message_data", "data", "content_type"),
EMAIL_DATA,