Revert "Set up smtp integration via the UI" (#110862)

Revert "Set up smtp integration via the UI (#110817)"

This reverts commit 66a31407f9ad0d2fa754cd168162cde5099cb215.
This commit is contained in:
Jan Bouwhuis 2024-02-18 16:21:27 +01:00 committed by GitHub
parent 67ac60d042
commit addc02fa86
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 105 additions and 652 deletions

View File

@ -1,98 +1 @@
"""Set up the smtp component.""" """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

View File

@ -1,197 +0,0 @@
"""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,
)

View File

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

View File

@ -10,13 +10,15 @@ import logging
import os import os
from pathlib import Path from pathlib import Path
import smtplib import smtplib
import socket
import voluptuous as vol
from homeassistant.components.notify import ( from homeassistant.components.notify import (
ATTR_DATA, ATTR_DATA,
ATTR_TARGET, ATTR_TARGET,
ATTR_TITLE, ATTR_TITLE,
ATTR_TITLE_DEFAULT, ATTR_TITLE_DEFAULT,
PLATFORM_SCHEMA,
BaseNotificationService, BaseNotificationService,
) )
from homeassistant.const import ( from homeassistant.const import (
@ -27,9 +29,12 @@ from homeassistant.const import (
CONF_TIMEOUT, CONF_TIMEOUT,
CONF_USERNAME, CONF_USERNAME,
CONF_VERIFY_SSL, CONF_VERIFY_SSL,
Platform,
) )
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError 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 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
import homeassistant.util.dt as dt_util import homeassistant.util.dt as dt_util
from homeassistant.util.ssl import client_context from homeassistant.util.ssl import client_context
@ -47,10 +52,31 @@ from .const import (
DEFAULT_PORT, DEFAULT_PORT,
DEFAULT_TIMEOUT, DEFAULT_TIMEOUT,
DOMAIN, DOMAIN,
ENCRYPTION_OPTIONS,
) )
PLATFORMS = [Platform.NOTIFY]
_LOGGER = logging.getLogger(__name__) _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( def get_service(
hass: HomeAssistant, hass: HomeAssistant,
@ -58,30 +84,21 @@ def get_service(
discovery_info: DiscoveryInfoType | None = None, discovery_info: DiscoveryInfoType | None = None,
) -> MailNotificationService | None: ) -> MailNotificationService | None:
"""Get the mail notification service.""" """Get the mail notification service."""
if discovery_info is None: setup_reload_service(hass, DOMAIN, PLATFORMS)
_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( mail_service = MailNotificationService(
discovery_info.get(CONF_SERVER, DEFAULT_HOST), config[CONF_SERVER],
discovery_info.get(CONF_PORT, DEFAULT_PORT), config[CONF_PORT],
discovery_info.get(CONF_TIMEOUT, DEFAULT_TIMEOUT), config[CONF_TIMEOUT],
discovery_info[CONF_SENDER], config[CONF_SENDER],
discovery_info.get(CONF_ENCRYPTION, DEFAULT_ENCRYPTION), config[CONF_ENCRYPTION],
discovery_info.get(CONF_USERNAME), config.get(CONF_USERNAME),
discovery_info.get(CONF_PASSWORD), config.get(CONF_PASSWORD),
discovery_info[CONF_RECIPIENT], config[CONF_RECIPIENT],
discovery_info.get(CONF_SENDER_NAME), config.get(CONF_SENDER_NAME),
discovery_info.get(CONF_DEBUG, DEFAULT_DEBUG), config[CONF_DEBUG],
discovery_info.get(CONF_VERIFY_SSL, True), config[CONF_VERIFY_SSL],
) )
hass.data[DOMAIN][entry_id] = mail_service
if mail_service.connection_is_valid(): if mail_service.connection_is_valid():
return mail_service return mail_service
@ -104,7 +121,7 @@ class MailNotificationService(BaseNotificationService):
sender_name, sender_name,
debug, debug,
verify_ssl, verify_ssl,
) -> None: ):
"""Initialize the SMTP service.""" """Initialize the SMTP service."""
self._server = server self._server = server
self._port = port self._port = port
@ -140,32 +157,25 @@ class MailNotificationService(BaseNotificationService):
mail.login(self.username, self.password) mail.login(self.username, self.password)
return mail return mail
def connection_is_valid(self, errors: dict[str, str] | None = None) -> bool: def connection_is_valid(self):
"""Check for valid config, verify connectivity.""" """Check for valid config, verify connectivity."""
server = None server = None
try: try:
server = self.connect() server = self.connect()
except smtplib.SMTPAuthenticationError: except (smtplib.socket.gaierror, ConnectionRefusedError):
if errors is None: _LOGGER.exception(
_LOGGER.exception( (
"Login not possible. Please check your setting and/or your credentials" "SMTP server not found or refused connection (%s:%s). Please check"
) " the IP address, hostname, and availability of your SMTP server"
else: ),
errors["base"] = "authentication_failed" self._server,
return False self._port,
)
except (socket.gaierror, ConnectionRefusedError, OSError): except smtplib.SMTPAuthenticationError:
if errors is None: _LOGGER.exception(
_LOGGER.exception( "Login not possible. Please check your setting and/or your credentials"
( )
"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 return False
finally: finally:

View File

@ -0,0 +1 @@
reload:

View File

@ -1,52 +1,8 @@
{ {
"config": { "services": {
"step": { "reload": {
"user": { "name": "[%key:common::action::reload%]",
"data": { "description": "Reloads smtp notify services."
"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": { "exceptions": {

View File

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

View File

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

View File

@ -1,15 +0,0 @@
"""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

View File

@ -1,36 +0,0 @@
""""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",
}

View File

@ -1,178 +0,0 @@
"""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,
}

View File

@ -1,20 +1,20 @@
"""The tests for the notify smtp platform.""" """The tests for the notify smtp platform."""
from copy import deepcopy
from pathlib import Path from pathlib import Path
import re import re
from unittest.mock import patch from unittest.mock import patch
import pytest import pytest
from homeassistant import config as hass_config
import homeassistant.components.notify as notify import homeassistant.components.notify as notify
from homeassistant.components.smtp.const import DOMAIN from homeassistant.components.smtp.const import DOMAIN
from homeassistant.components.smtp.notify import MailNotificationService from homeassistant.components.smtp.notify import MailNotificationService
from homeassistant.const import SERVICE_RELOAD
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ServiceValidationError from homeassistant.exceptions import ServiceValidationError
from homeassistant.setup import async_setup_component
from .const import MOCKED_CONFIG_ENTRY_DATA from tests.common import get_fixture_path
from tests.common import MockConfigEntry
class MockSMTP(MailNotificationService): class MockSMTP(MailNotificationService):
@ -25,6 +25,46 @@ class MockSMTP(MailNotificationService):
return msg.as_string(), recipients 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 @pytest.fixture
def message(): def message():
"""Return MockSMTP object with test data.""" """Return MockSMTP object with test data."""
@ -83,35 +123,6 @@ 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( @pytest.mark.parametrize(
("message_data", "data", "content_type"), ("message_data", "data", "content_type"),
EMAIL_DATA, EMAIL_DATA,