github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Remove imap_email_content integration (#101233)

This commit is contained in:
Jan Bouwhuis 2023-10-02 08:57:16 +02:00 committed by GitHub
parent c1cfce116d
commit 78f827697e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 1 additions and 1193 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.coveragerc
View File

@ -563,7 +563,6 @@ omit =
homeassistant/components/ifttt/alarm_control_panel.py
homeassistant/components/iglo/light.py
homeassistant/components/ihc/*
homeassistant/components/imap_email_content/sensor.py
homeassistant/components/incomfort/*
homeassistant/components/insteon/binary_sensor.py
homeassistant/components/insteon/climate.py

30
homeassistant/components/imap/config_flow.py
View File

@ -10,13 +10,7 @@ from aioimaplib import AioImapException
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import (
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
from homeassistant.const import CONF_PASSWORD, CONF_PORT, CONF_USERNAME, CONF_VERIFY_SSL
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import AbortFlow, FlowResult
from homeassistant.helpers import config_validation as cv
@ -132,28 +126,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
_reauth_entry: config_entries.ConfigEntry | None
async def async_step_import(self, user_input: dict[str, Any]) -> FlowResult:
"""Handle the import from imap_email_content integration."""
data = CONFIG_SCHEMA(
{
CONF_SERVER: user_input[CONF_SERVER],
CONF_PORT: user_input[CONF_PORT],
CONF_USERNAME: user_input[CONF_USERNAME],
CONF_PASSWORD: user_input[CONF_PASSWORD],
CONF_FOLDER: user_input[CONF_FOLDER],
}
)
self._async_abort_entries_match(
{
key: data[key]
for key in (CONF_USERNAME, CONF_SERVER, CONF_FOLDER, CONF_SEARCH)
}
)
title = user_input[CONF_NAME]
if await validate_input(self.hass, data):
raise AbortFlow("cannot_connect")
return self.async_create_entry(title=title, data=data)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:

17
homeassistant/components/imap_email_content/__init__.py
View File

@ -1,17 +0,0 @@
"""The imap_email_content component."""
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN
PLATFORMS = [Platform.SENSOR]
CONFIG_SCHEMA = cv.deprecated(DOMAIN, raise_if_present=False)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up imap_email_content."""
return True

13
homeassistant/components/imap_email_content/const.py
View File

@ -1,13 +0,0 @@
"""Constants for the imap email content integration."""
DOMAIN = "imap_email_content"
CONF_SERVER = "server"
CONF_SENDERS = "senders"
CONF_FOLDER = "folder"
ATTR_FROM = "from"
ATTR_BODY = "body"
ATTR_SUBJECT = "subject"
DEFAULT_PORT = 993

8
homeassistant/components/imap_email_content/manifest.json
View File

@ -1,8 +0,0 @@
{
"domain": "imap_email_content",
"name": "IMAP Email Content",
"codeowners": [],
"dependencies": ["imap"],
"documentation": "https://www.home-assistant.io/integrations/imap_email_content",
"iot_class": "cloud_push"
}

173
homeassistant/components/imap_email_content/repairs.py
View File

@ -1,173 +0,0 @@
"""Repair flow for imap email content integration."""
from typing import Any
import voluptuous as vol
import yaml
from homeassistant import data_entry_flow
from homeassistant.components.imap import DOMAIN as IMAP_DOMAIN
from homeassistant.components.repairs import RepairsFlow
from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
CONF_VALUE_TEMPLATE,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import issue_registry as ir
from homeassistant.helpers.typing import ConfigType
from .const import CONF_FOLDER, CONF_SENDERS, CONF_SERVER, DOMAIN
async def async_process_issue(hass: HomeAssistant, config: ConfigType) -> None:
"""Register an issue and suggest new config."""
name: str = config.get(CONF_NAME) or config[CONF_USERNAME]
issue_id = (
f"{name}_{config[CONF_USERNAME]}_{config[CONF_SERVER]}_{config[CONF_FOLDER]}"
)
if CONF_VALUE_TEMPLATE in config:
template: str = config[CONF_VALUE_TEMPLATE].template
template = template.replace("subject", 'trigger.event.data["subject"]')
template = template.replace("from", 'trigger.event.data["sender"]')
template = template.replace("date", 'trigger.event.data["date"]')
template = template.replace("body", 'trigger.event.data["text"]')
else:
template = '{{ trigger.event.data["subject"] }}'
template_sensor_config: ConfigType = {
"template": [
{
"trigger": [
{
"id": "custom_event",
"platform": "event",
"event_type": "imap_content",
"event_data": {"sender": config[CONF_SENDERS][0]},
}
],
"sensor": [
{
"state": template,
"name": name,
}
],
}
]
}
data = {
CONF_SERVER: config[CONF_SERVER],
CONF_PORT: config[CONF_PORT],
CONF_USERNAME: config[CONF_USERNAME],
CONF_PASSWORD: config[CONF_PASSWORD],
CONF_FOLDER: config[CONF_FOLDER],
}
data[CONF_VALUE_TEMPLATE] = template
data[CONF_NAME] = name
placeholders = {"yaml_example": yaml.dump(template_sensor_config)}
placeholders.update(data)
ir.async_create_issue(
hass,
DOMAIN,
issue_id,
breaks_in_ha_version="2023.11.0",
is_fixable=True,
severity=ir.IssueSeverity.WARNING,
translation_key="migration",
translation_placeholders=placeholders,
data=data,
)
class DeprecationRepairFlow(RepairsFlow):
"""Handler for an issue fixing flow."""
def __init__(self, issue_id: str, config: ConfigType) -> None:
"""Create flow."""
self._name: str = config[CONF_NAME]
self._config: dict[str, Any] = config
self._issue_id = issue_id
super().__init__()
async def async_step_init(
self, user_input: dict[str, str] | None = None
) -> data_entry_flow.FlowResult:
"""Handle the first step of a fix flow."""
return await self.async_step_start()
@callback
def _async_get_placeholders(self) -> dict[str, str] | None:
issue_registry = ir.async_get(self.hass)
description_placeholders = None
if issue := issue_registry.async_get_issue(self.handler, self.issue_id):
description_placeholders = issue.translation_placeholders
return description_placeholders
async def async_step_start(
self, user_input: dict[str, str] | None = None
) -> data_entry_flow.FlowResult:
"""Wait for the user to start the config migration."""
placeholders = self._async_get_placeholders()
if user_input is None:
return self.async_show_form(
step_id="start",
data_schema=vol.Schema({}),
description_placeholders=placeholders,
)
return await self.async_step_confirm()
async def async_step_confirm(
self, user_input: dict[str, str] | None = None
) -> data_entry_flow.FlowResult:
"""Handle the confirm step of a fix flow."""
placeholders = self._async_get_placeholders()
if user_input is not None:
user_input[CONF_NAME] = self._name
result = await self.hass.config_entries.flow.async_init(
IMAP_DOMAIN, context={"source": SOURCE_IMPORT}, data=self._config
)
if result["type"] == FlowResultType.ABORT:
ir.async_delete_issue(self.hass, DOMAIN, self._issue_id)
ir.async_create_issue(
self.hass,
DOMAIN,
self._issue_id,
breaks_in_ha_version="2023.11.0",
is_fixable=False,
severity=ir.IssueSeverity.WARNING,
translation_key="deprecation",
translation_placeholders=placeholders,
data=self._config,
learn_more_url="https://www.home-assistant.io/integrations/imap/#using-events",
)
return self.async_abort(reason=result["reason"])
return self.async_create_entry(
title="",
data={},
)
return self.async_show_form(
step_id="confirm",
data_schema=vol.Schema({}),
description_placeholders=placeholders,
)
async def async_create_fix_flow(
hass: HomeAssistant,
issue_id: str,
data: dict[str, str | int | float | None],
) -> RepairsFlow:
"""Create flow."""
return DeprecationRepairFlow(issue_id, data)

302
homeassistant/components/imap_email_content/sensor.py
View File

@ -1,302 +0,0 @@
"""Email sensor support."""
from __future__ import annotations
from collections import deque
import datetime
import email
import imaplib
import logging
import voluptuous as vol
from homeassistant.components.sensor import PLATFORM_SCHEMA, SensorEntity
from homeassistant.const import (
ATTR_DATE,
CONF_NAME,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
CONF_VALUE_TEMPLATE,
CONF_VERIFY_SSL,
CONTENT_TYPE_TEXT_PLAIN,
)
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util.ssl import client_context
from .const import (
ATTR_BODY,
ATTR_FROM,
ATTR_SUBJECT,
CONF_FOLDER,
CONF_SENDERS,
CONF_SERVER,
DEFAULT_PORT,
)
from .repairs import async_process_issue
_LOGGER = logging.getLogger(__name__)
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Optional(CONF_NAME): cv.string,
vol.Required(CONF_USERNAME): cv.string,
vol.Required(CONF_PASSWORD): cv.string,
vol.Required(CONF_SERVER): cv.string,
vol.Required(CONF_SENDERS): [cv.string],
vol.Optional(CONF_PORT, default=DEFAULT_PORT): cv.port,
vol.Optional(CONF_VALUE_TEMPLATE): cv.template,
vol.Optional(CONF_FOLDER, default="INBOX"): cv.string,
vol.Optional(CONF_VERIFY_SSL, default=True): cv.boolean,
}
)
def setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the Email sensor platform."""
reader = EmailReader(
config[CONF_USERNAME],
config[CONF_PASSWORD],
config[CONF_SERVER],
config[CONF_PORT],
config[CONF_FOLDER],
config[CONF_VERIFY_SSL],
)
if (value_template := config.get(CONF_VALUE_TEMPLATE)) is not None:
value_template.hass = hass
sensor = EmailContentSensor(
hass,
reader,
config.get(CONF_NAME) or config[CONF_USERNAME],
config[CONF_SENDERS],
value_template,
)
hass.add_job(async_process_issue, hass, config)
if sensor.connected:
add_entities([sensor], True)
class EmailReader:
"""A class to read emails from an IMAP server."""
def __init__(self, user, password, server, port, folder, verify_ssl):
"""Initialize the Email Reader."""
self._user = user
self._password = password
self._server = server
self._port = port
self._folder = folder
self._verify_ssl = verify_ssl
self._last_id = None
self._last_message = None
self._unread_ids = deque([])
self.connection = None
@property
def last_id(self) -> int | None:
"""Return last email uid that was processed."""
return self._last_id
@property
def last_unread_id(self) -> int | None:
"""Return last email uid received."""
# We assume the last id in the list is the last unread id
# We cannot know if that is the newest one, because it could arrive later
# https://stackoverflow.com/questions/12409862/python-imap-the-order-of-uids
if self._unread_ids:
return int(self._unread_ids[-1])
return self._last_id
def connect(self):
"""Login and setup the connection."""
ssl_context = client_context() if self._verify_ssl else None
try:
self.connection = imaplib.IMAP4_SSL(
self._server, self._port, ssl_context=ssl_context
)
self.connection.login(self._user, self._password)
return True
except imaplib.IMAP4.error:
_LOGGER.error("Failed to login to %s", self._server)
return False
def _fetch_message(self, message_uid):
"""Get an email message from a message id."""
_, message_data = self.connection.uid("fetch", message_uid, "(RFC822)")
if message_data is None:
return None
if message_data[0] is None:
return None
raw_email = message_data[0][1]
email_message = email.message_from_bytes(raw_email)
return email_message
def read_next(self):
"""Read the next email from the email server."""
try:
self.connection.select(self._folder, readonly=True)
if self._last_id is None:
# search for today and yesterday
time_from = datetime.datetime.now() - datetime.timedelta(days=1)
search = f"SINCE {time_from:%d-%b-%Y}"
else:
search = f"UID {self._last_id}:*"
_, data = self.connection.uid("search", None, search)
self._unread_ids = deque(data[0].split())
while self._unread_ids:
message_uid = self._unread_ids.popleft()
if self._last_id is None or int(message_uid) > self._last_id:
self._last_id = int(message_uid)
self._last_message = self._fetch_message(message_uid)
return self._last_message
except imaplib.IMAP4.error:
_LOGGER.info("Connection to %s lost, attempting to reconnect", self._server)
try:
self.connect()
_LOGGER.info(
"Reconnect to %s succeeded, trying last message", self._server
)
if self._last_id is not None:
return self._fetch_message(str(self._last_id))
except imaplib.IMAP4.error:
_LOGGER.error("Failed to reconnect")
return None
class EmailContentSensor(SensorEntity):
"""Representation of an EMail sensor."""
def __init__(self, hass, email_reader, name, allowed_senders, value_template):
"""Initialize the sensor."""
self.hass = hass
self._email_reader = email_reader
self._name = name
self._allowed_senders = [sender.upper() for sender in allowed_senders]
self._value_template = value_template
self._last_id = None
self._message = None
self._state_attributes = None
self.connected = self._email_reader.connect()
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def native_value(self):
"""Return the current email state."""
return self._message
@property
def extra_state_attributes(self):
"""Return other state attributes for the message."""
return self._state_attributes
def render_template(self, email_message):
"""Render the message template."""
variables = {
ATTR_FROM: EmailContentSensor.get_msg_sender(email_message),
ATTR_SUBJECT: EmailContentSensor.get_msg_subject(email_message),
ATTR_DATE: email_message["Date"],
ATTR_BODY: EmailContentSensor.get_msg_text(email_message),
}
return self._value_template.render(variables, parse_result=False)
def sender_allowed(self, email_message):
"""Check if the sender is in the allowed senders list."""
return EmailContentSensor.get_msg_sender(email_message).upper() in (
sender for sender in self._allowed_senders
)
@staticmethod
def get_msg_sender(email_message):
"""Get the parsed message sender from the email."""
return str(email.utils.parseaddr(email_message["From"])[1])
@staticmethod
def get_msg_subject(email_message):
"""Decode the message subject."""
decoded_header = email.header.decode_header(email_message["Subject"])
header = email.header.make_header(decoded_header)
return str(header)
@staticmethod
def get_msg_text(email_message):
"""Get the message text from the email.
Will look for text/plain or use text/html if not found.
"""
message_text = None
message_html = None
message_untyped_text = None
for part in email_message.walk():
if part.get_content_type() == CONTENT_TYPE_TEXT_PLAIN:
if message_text is None:
message_text = part.get_payload()
elif part.get_content_type() == "text/html":
if message_html is None:
message_html = part.get_payload()
elif (
part.get_content_type().startswith("text")
and message_untyped_text is None
):
message_untyped_text = part.get_payload()
if message_text is not None:
return message_text
if message_html is not None:
return message_html
if message_untyped_text is not None:
return message_untyped_text
return email_message.get_payload()
def update(self) -> None:
"""Read emails and publish state change."""
email_message = self._email_reader.read_next()
while (
self._last_id is None or self._last_id != self._email_reader.last_unread_id
):
if email_message is None:
self._message = None
self._state_attributes = {}
return
self._last_id = self._email_reader.last_id
if self.sender_allowed(email_message):
message = EmailContentSensor.get_msg_subject(email_message)
if self._value_template is not None:
message = self.render_template(email_message)
self._message = message
self._state_attributes = {
ATTR_FROM: EmailContentSensor.get_msg_sender(email_message),
ATTR_SUBJECT: EmailContentSensor.get_msg_subject(email_message),
ATTR_DATE: email_message["Date"],
ATTR_BODY: EmailContentSensor.get_msg_text(email_message),
}
if self._last_id == self._email_reader.last_unread_id:
break
email_message = self._email_reader.read_next()

27
homeassistant/components/imap_email_content/strings.json
View File

@ -1,27 +0,0 @@
{
"issues": {
"deprecation": {
"title": "The IMAP email content integration is deprecated",
"description": "The IMAP email content integration is deprecated. Your IMAP server configuration was already migrated to the [imap integration](https://my.home-assistant.io/redirect/config_flow_start?domain=imap). To set up a sensor for the IMAP email content, set up a template sensor with the config:\n\n```yaml\n{yaml_example}```\n\nPlease remove the deprecated `imap_email_plaform` sensor configuration from your `configuration.yaml`.\n\nNote that the event filter only filters on the first of the configured allowed senders, customize the filter if needed.\n\nYou can skip this part if you have already set up a template sensor."
},
"migration": {
"title": "The IMAP email content integration needs attention",
"fix_flow": {
"step": {
"start": {
"title": "Migrate your IMAP email configuration",
"description": "The IMAP email content integration is deprecated. Your IMAP server configuration can be migrated automatically to the [imap integration](https://my.home-assistant.io/redirect/config_flow_start?domain=imap), this will enable using a custom `imap` event trigger. To set up a sensor that has an IMAP content state, a template sensor can be used. Remove the `imap_email_plaform` sensor configuration from your `configuration.yaml` after migration.\n\nSubmit to start migration of your IMAP server configuration to the `imap` integration."
},
"confirm": {
"title": "Your IMAP server settings will be migrated",
"description": "In this step an `imap` config entry will be set up with the following configuration:\n\n```text\nServer\t{server}\nPort\t{port}\nUsername\t{username}\nPassword\t*****\nFolder\t{folder}\n```\n\nSee also: (https://www.home-assistant.io/integrations/imap/)\n\nFitering configuration on allowed `sender` is part of the template sensor config that can copied and placed in your `configuration.yaml.\n\nNote that the event filter only filters on the first of the configured allowed senders, customize the filter if needed.\n\n```yaml\n{yaml_example}```\nDo not forget to cleanup the your `configuration.yaml` after migration.\n\nSubmit to migrate your IMAP server configuration to an `imap` configuration entry."
}
},
"abort": {
"already_configured": "The IMAP server config was already migrated to the imap integration. Remove the `imap_email_plaform` sensor configuration from your `configuration.yaml`.",
"cannot_connect": "Migration failed. Failed to connect to the IMAP server. Perform a manual migration."
}
}
}
}
}

6
homeassistant/generated/integrations.json
View File

@ -2604,12 +2604,6 @@
"config_flow": true,
"iot_class": "cloud_push"
},
"imap_email_content": {
"name": "IMAP Email Content",
"integration_type": "hub",
"config_flow": false,
"iot_class": "cloud_push"
},
"incomfort": {
"name": "Intergas InComfort/Intouch Lan2RF gateway",
"integration_type": "hub",

67
tests/components/imap/test_config_flow.py
View File

@ -469,73 +469,6 @@ async def test_advanced_options_form(
assert assert_result == data_entry_flow.FlowResultType.FORM
async def test_import_flow_success(hass: HomeAssistant) -> None:
"""Test a successful import of yaml."""
with patch(
"homeassistant.components.imap.config_flow.connect_to_server"
) as mock_client, patch(
"homeassistant.components.imap.async_setup_entry",
return_value=True,
) as mock_setup_entry:
mock_client.return_value.search.return_value = (
"OK",
[b""],
)
result2 = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
"name": "IMAP",
"username": "email@email.com",
"password": "password",
"server": "imap.server.com",
"port": 993,
"folder": "INBOX",
},
)
await hass.async_block_till_done()
assert result2["type"] == FlowResultType.CREATE_ENTRY
assert result2["title"] == "IMAP"
assert result2["data"] == {
"username": "email@email.com",
"password": "password",
"server": "imap.server.com",
"port": 993,
"charset": "utf-8",
"folder": "INBOX",
"search": "UnSeen UnDeleted",
}
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_flow_connection_error(hass: HomeAssistant) -> None:
"""Test a successful import of yaml."""
with patch(
"homeassistant.components.imap.config_flow.connect_to_server",
side_effect=AioImapException("Unexpected error"),
), patch(
"homeassistant.components.imap.async_setup_entry",
return_value=True,
):
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_IMPORT},
data={
"name": "IMAP",
"username": "email@email.com",
"password": "password",
"server": "imap.server.com",
"port": 993,
"folder": "INBOX",
},
)
await hass.async_block_till_done()
assert result["type"] == FlowResultType.ABORT
assert result["reason"] == "cannot_connect"
@pytest.mark.parametrize("cipher_list", ["python_default", "modern", "intermediate"])
@pytest.mark.parametrize("verify_ssl", [False, True])
async def test_config_flow_with_cipherlist_and_ssl_verify(

1
tests/components/imap_email_content/__init__.py
View File

@ -1 +0,0 @@
"""Tests for the imap_email_content component."""

296
tests/components/imap_email_content/test_repairs.py
View File

@ -1,296 +0,0 @@
"""Test repairs for imap_email_content."""
from collections.abc import Generator
from http import HTTPStatus
from unittest.mock import MagicMock, patch
import pytest
from homeassistant.components.repairs.websocket_api import (
RepairsFlowIndexView,
RepairsFlowResourceView,
)
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
from tests.typing import ClientSessionGenerator, WebSocketGenerator
@pytest.fixture
def mock_client() -> Generator[MagicMock, None, None]:
"""Mock the imap client."""
with patch(
"homeassistant.components.imap_email_content.sensor.EmailReader.read_next",
return_value=None,
), patch("imaplib.IMAP4_SSL") as mock_imap_client:
yield mock_imap_client
CONFIG = {
"platform": "imap_email_content",
"name": "Notifications",
"server": "imap.example.com",
"port": 993,
"username": "john.doe@example.com",
"password": "**SECRET**",
"folder": "INBOX.Notifications",
"value_template": "{{ body }}",
"senders": ["company@example.com"],
}
DESCRIPTION_PLACEHOLDERS = {
"yaml_example": ""
"template:\n"
"- sensor:\n"
" - name: Notifications\n"
" state: '{{ trigger.event.data[\"text\"] }}'\n"
" trigger:\n - event_data:\n"
" sender: company@example.com\n"
" event_type: imap_content\n"
" id: custom_event\n"
" platform: event\n",
"server": "imap.example.com",
"port": 993,
"username": "john.doe@example.com",
"password": "**SECRET**",
"folder": "INBOX.Notifications",
"value_template": '{{ trigger.event.data["text"] }}',
"name": "Notifications",
}
CONFIG_DEFAULT = {
"platform": "imap_email_content",
"name": "Notifications",
"server": "imap.example.com",
"port": 993,
"username": "john.doe@example.com",
"password": "**SECRET**",
"folder": "INBOX.Notifications",
"senders": ["company@example.com"],
}
DESCRIPTION_PLACEHOLDERS_DEFAULT = {
"yaml_example": ""
"template:\n"
"- sensor:\n"
" - name: Notifications\n"
" state: '{{ trigger.event.data[\"subject\"] }}'\n"
" trigger:\n - event_data:\n"
" sender: company@example.com\n"
" event_type: imap_content\n"
" id: custom_event\n"
" platform: event\n",
"server": "imap.example.com",
"port": 993,
"username": "john.doe@example.com",
"password": "**SECRET**",
"folder": "INBOX.Notifications",
"value_template": '{{ trigger.event.data["subject"] }}',
"name": "Notifications",
}
@pytest.mark.parametrize(
("config", "description_placeholders"),
[
(CONFIG, DESCRIPTION_PLACEHOLDERS),
(CONFIG_DEFAULT, DESCRIPTION_PLACEHOLDERS_DEFAULT),
],
ids=["with_value_template", "default_subject"],
)
async def test_deprecation_repair_flow(
hass: HomeAssistant,
mock_client: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
config: str | None,
description_placeholders: str,
) -> None:
"""Test the deprecation repair flow."""
# setup config
await async_setup_component(hass, "sensor", {"sensor": config})
await hass.async_block_till_done()
state = hass.states.get("sensor.notifications")
assert state is not None
ws_client = await hass_ws_client(hass)
client = await hass_client()
await ws_client.send_json({"id": 1, "type": "repairs/list_issues"})
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["issues"]) > 0
issue = None
for i in msg["result"]["issues"]:
if i["domain"] == "imap_email_content":
issue = i
assert issue is not None
assert (
issue["issue_id"]
== "Notifications_john.doe@example.com_imap.example.com_INBOX.Notifications"
)
assert issue["is_fixable"]
url = RepairsFlowIndexView.url
resp = await client.post(
url, json={"handler": "imap_email_content", "issue_id": issue["issue_id"]}
)
assert resp.status == HTTPStatus.OK
data = await resp.json()
flow_id = data["flow_id"]
assert data["description_placeholders"] == description_placeholders
assert data["step_id"] == "start"
# Apply fix
url = RepairsFlowResourceView.url.format(flow_id=flow_id)
resp = await client.post(url)
assert resp.status == HTTPStatus.OK
data = await resp.json()
flow_id = data["flow_id"]
assert data["description_placeholders"] == description_placeholders
assert data["step_id"] == "confirm"
url = RepairsFlowResourceView.url.format(flow_id=flow_id)
with patch(
"homeassistant.components.imap.config_flow.connect_to_server"
) as mock_client, patch(
"homeassistant.components.imap.async_setup_entry",
return_value=True,
):
mock_client.return_value.search.return_value = (
"OK",
[b""],
)
resp = await client.post(url)
assert resp.status == HTTPStatus.OK
data = await resp.json()
assert data["type"] == "create_entry"
# Assert the issue is resolved
await ws_client.send_json({"id": 2, "type": "repairs/list_issues"})
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["issues"]) == 0
@pytest.mark.parametrize(
("config", "description_placeholders"),
[
(CONFIG, DESCRIPTION_PLACEHOLDERS),
(CONFIG_DEFAULT, DESCRIPTION_PLACEHOLDERS_DEFAULT),
],
ids=["with_value_template", "default_subject"],
)
async def test_repair_flow_where_entry_already_exists(
hass: HomeAssistant,
mock_client: MagicMock,
hass_client: ClientSessionGenerator,
hass_ws_client: WebSocketGenerator,
config: str | None,
description_placeholders: str,
) -> None:
"""Test the deprecation repair flow and an entry already exists."""
await async_setup_component(hass, "sensor", {"sensor": config})
await hass.async_block_till_done()
state = hass.states.get("sensor.notifications")
assert state is not None
existing_imap_entry_config = {
"username": "john.doe@example.com",
"password": "password",
"server": "imap.example.com",
"port": 993,
"charset": "utf-8",
"folder": "INBOX.Notifications",
"search": "UnSeen UnDeleted",
}
with patch("homeassistant.components.imap.async_setup_entry", return_value=True):
imap_entry = MockConfigEntry(domain="imap", data=existing_imap_entry_config)
imap_entry.add_to_hass(hass)
await hass.config_entries.async_setup(imap_entry.entry_id)
ws_client = await hass_ws_client(hass)
client = await hass_client()
await ws_client.send_json({"id": 1, "type": "repairs/list_issues"})
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["issues"]) > 0
issue = None
for i in msg["result"]["issues"]:
if i["domain"] == "imap_email_content":
issue = i
assert issue is not None
assert (
issue["issue_id"]
== "Notifications_john.doe@example.com_imap.example.com_INBOX.Notifications"
)
assert issue["is_fixable"]
assert issue["translation_key"] == "migration"
url = RepairsFlowIndexView.url
resp = await client.post(
url, json={"handler": "imap_email_content", "issue_id": issue["issue_id"]}
)
assert resp.status == HTTPStatus.OK
data = await resp.json()
flow_id = data["flow_id"]
assert data["description_placeholders"] == description_placeholders
assert data["step_id"] == "start"
url = RepairsFlowResourceView.url.format(flow_id=flow_id)
resp = await client.post(url)
assert resp.status == HTTPStatus.OK
data = await resp.json()
flow_id = data["flow_id"]
assert data["description_placeholders"] == description_placeholders
assert data["step_id"] == "confirm"
url = RepairsFlowResourceView.url.format(flow_id=flow_id)
with patch(
"homeassistant.components.imap.config_flow.connect_to_server"
) as mock_client, patch(
"homeassistant.components.imap.async_setup_entry",
return_value=True,
):
mock_client.return_value.search.return_value = (
"OK",
[b""],
)
resp = await client.post(url)
assert resp.status == HTTPStatus.OK
data = await resp.json()
assert data["type"] == "abort"
assert data["reason"] == "already_configured"
# We should now have a non_fixable issue left since there is still
# a config in configuration.yaml
await ws_client.send_json({"id": 2, "type": "repairs/list_issues"})
msg = await ws_client.receive_json()
assert msg["success"]
assert len(msg["result"]["issues"]) > 0
issue = None
for i in msg["result"]["issues"]:
if i["domain"] == "imap_email_content":
issue = i
assert issue is not None
assert (
issue["issue_id"]
== "Notifications_john.doe@example.com_imap.example.com_INBOX.Notifications"
)
assert not issue["is_fixable"]
assert issue["translation_key"] == "deprecation"

253
tests/components/imap_email_content/test_sensor.py
View File

@ -1,253 +0,0 @@
"""The tests for the IMAP email content sensor platform."""
from collections import deque
import datetime
import email
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from homeassistant.components.imap_email_content import sensor as imap_email_content
from homeassistant.core import HomeAssistant
from homeassistant.helpers.event import async_track_state_change
from homeassistant.helpers.template import Template
from homeassistant.setup import async_setup_component
class FakeEMailReader:
"""A test class for sending test emails."""
def __init__(self, messages) -> None:
"""Set up the fake email reader."""
self._messages = messages
self.last_id = 0
self.last_unread_id = len(messages)
def add_test_message(self, message):
"""Add a new message."""
self.last_unread_id += 1
self._messages.append(message)
def connect(self):
"""Stay always Connected."""
return True
def read_next(self):
"""Get the next email."""
if len(self._messages) == 0:
return None
self.last_id += 1
return self._messages.popleft()
async def test_integration_setup_(hass: HomeAssistant) -> None:
"""Test the integration component setup is successful."""
assert await async_setup_component(hass, "imap_email_content", {})
async def test_allowed_sender(hass: HomeAssistant) -> None:
"""Test emails from allowed sender."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
test_message["Subject"] = "Test"
test_message["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message.set_payload("Test Message")
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([test_message])),
"test_emails_sensor",
["sender@test.com"],
None,
)
sensor.entity_id = "sensor.emailtest"
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert sensor.state == "Test"
assert sensor.extra_state_attributes["body"] == "Test Message"
assert sensor.extra_state_attributes["from"] == "sender@test.com"
assert sensor.extra_state_attributes["subject"] == "Test"
assert (
datetime.datetime(2016, 1, 1, 12, 44, 57)
== sensor.extra_state_attributes["date"]
)
async def test_multi_part_with_text(hass: HomeAssistant) -> None:
"""Test multi part emails."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
msg["From"] = "sender@test.com"
text = "Test Message"
html = "<html><head></head><body>Test Message</body></html>"
textPart = MIMEText(text, "plain")
htmlPart = MIMEText(html, "html")
msg.attach(textPart)
msg.attach(htmlPart)
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)
sensor.entity_id = "sensor.emailtest"
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert sensor.state == "Link"
assert sensor.extra_state_attributes["body"] == "Test Message"
async def test_multi_part_only_html(hass: HomeAssistant) -> None:
"""Test multi part emails with only HTML."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
msg["From"] = "sender@test.com"
html = "<html><head></head><body>Test Message</body></html>"
htmlPart = MIMEText(html, "html")
msg.attach(htmlPart)
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)
sensor.entity_id = "sensor.emailtest"
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert sensor.state == "Link"
assert (
sensor.extra_state_attributes["body"]
== "<html><head></head><body>Test Message</body></html>"
)
async def test_multi_part_only_other_text(hass: HomeAssistant) -> None:
"""Test multi part emails with only other text."""
msg = MIMEMultipart("alternative")
msg["Subject"] = "Link"
msg["From"] = "sender@test.com"
other = "Test Message"
htmlPart = MIMEText(other, "other")
msg.attach(htmlPart)
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([msg])),
"test_emails_sensor",
["sender@test.com"],
None,
)
sensor.entity_id = "sensor.emailtest"
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert sensor.state == "Link"
assert sensor.extra_state_attributes["body"] == "Test Message"
async def test_multiple_emails(hass: HomeAssistant) -> None:
"""Test multiple emails, discarding stale states."""
states = []
test_message1 = email.message.Message()
test_message1["From"] = "sender@test.com"
test_message1["Subject"] = "Test"
test_message1["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message1.set_payload("Test Message")
test_message2 = email.message.Message()
test_message2["From"] = "sender@test.com"
test_message2["Subject"] = "Test 2"
test_message2["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 58)
test_message2.set_payload("Test Message 2")
test_message3 = email.message.Message()
test_message3["From"] = "sender@test.com"
test_message3["Subject"] = "Test 3"
test_message3["Date"] = datetime.datetime(2016, 1, 1, 12, 50, 1)
test_message3.set_payload("Test Message 2")
def state_changed_listener(entity_id, from_s, to_s):
states.append(to_s)
async_track_state_change(hass, ["sensor.emailtest"], state_changed_listener)
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([test_message1, test_message2])),
"test_emails_sensor",
["sender@test.com"],
None,
)
sensor.entity_id = "sensor.emailtest"
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
# Fake a new received message
sensor._email_reader.add_test_message(test_message3)
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert states[0].state == "Test 2"
assert states[1].state == "Test 3"
assert sensor.extra_state_attributes["body"] == "Test Message 2"
async def test_sender_not_allowed(hass: HomeAssistant) -> None:
"""Test not whitelisted emails."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
test_message["Subject"] = "Test"
test_message["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message.set_payload("Test Message")
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([test_message])),
"test_emails_sensor",
["other@test.com"],
None,
)
sensor.entity_id = "sensor.emailtest"
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert sensor.state is None
async def test_template(hass: HomeAssistant) -> None:
"""Test value template."""
test_message = email.message.Message()
test_message["From"] = "sender@test.com"
test_message["Subject"] = "Test"
test_message["Date"] = datetime.datetime(2016, 1, 1, 12, 44, 57)
test_message.set_payload("Test Message")
sensor = imap_email_content.EmailContentSensor(
hass,
FakeEMailReader(deque([test_message])),
"test_emails_sensor",
["sender@test.com"],
Template("{{ subject }} from {{ from }} with message {{ body }}", hass),
)
sensor.entity_id = "sensor.emailtest"
sensor.async_schedule_update_ha_state(True)
await hass.async_block_till_done()
assert sensor.state == "Test from sender@test.com with message Test Message"