evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add imap custom event data template (#93423)

Browse source 
* Add imap custom event template

* Add template validation
This commit is contained in:
Jan Bouwhuis 2023-05-25 11:05:25 +02:00 • committed by GitHub
parent 6cd766ef1f
commit 1b5d207984
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
7 changed files with 155 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

2
homeassistant/components/imap/__init__.py
View file

@ -45,7 +45,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
coordinator_class = ImapPollingDataUpdateCoordinator
coordinator: ImapPushDataUpdateCoordinator | ImapPollingDataUpdateCoordinator = (
coordinator_class(hass, imap_client)
coordinator_class(hass, imap_client, entry)
)
await coordinator.async_config_entry_first_refresh()

32
homeassistant/components/imap/config_flow.py
View file

@ -11,18 +11,24 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import AbortFlow, FlowResult
from homeassistant.exceptions import TemplateError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.selector import (
SelectSelector,
SelectSelectorConfig,
SelectSelectorMode,
TextSelector,
TextSelectorConfig,
TextSelectorType,
)
from homeassistant.helpers.template import Template
from homeassistant.util.ssl import SSLCipherList
from .const import (
CONF_CHARSET,
CONF_CUSTOM_EVENT_DATA_TEMPLATE,
CONF_FOLDER,
CONF_MAX_MESSAGE_SIZE,
CONF_SEARCH,
@ -43,6 +49,9 @@ CIPHER_SELECTOR = SelectSelector(
translation_key=CONF_SSL_CIPHER_LIST,
)
)
TEMPLATE_SELECTOR = TextSelector(
TextSelectorConfig(type=TextSelectorType.TEXT, multiline=True)
)
CONFIG_SCHEMA = vol.Schema(
{
@ -69,14 +78,17 @@ OPTIONS_SCHEMA = vol.Schema(
)
OPTIONS_SCHEMA_ADVANCED = {
vol.Optional(CONF_CUSTOM_EVENT_DATA_TEMPLATE): TEMPLATE_SELECTOR,
vol.Optional(CONF_MAX_MESSAGE_SIZE, default=DEFAULT_MAX_MESSAGE_SIZE): vol.All(
cv.positive_int,
vol.Range(min=DEFAULT_MAX_MESSAGE_SIZE, max=MAX_MESSAGE_SIZE_LIMIT),
)
),
}
async def validate_input(user_input: dict[str, Any]) -> dict[str, str]:
async def validate_input(
hass: HomeAssistant, user_input: dict[str, Any]
) -> dict[str, str]:
"""Validate user input."""
errors = {}
@ -104,6 +116,12 @@ async def validate_input(user_input: dict[str, Any]) -> dict[str, str]:
errors[CONF_CHARSET] = "invalid_charset"
else:
errors[CONF_SEARCH] = "invalid_search"
if template := user_input.get(CONF_CUSTOM_EVENT_DATA_TEMPLATE):
try:
Template(template, hass=hass).ensure_valid()
except TemplateError:
errors[CONF_CUSTOM_EVENT_DATA_TEMPLATE] = "invalid_template"
return errors
@ -131,7 +149,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
}
)
title = user_input[CONF_NAME]
if await validate_input(data):
if await validate_input(self.hass, data):
raise AbortFlow("cannot_connect")
return self.async_create_entry(title=title, data=data)
@ -154,7 +172,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
}
)
if not (errors := await validate_input(user_input)):
if not (errors := await validate_input(self.hass, user_input)):
title = user_input[CONF_USERNAME]
return self.async_create_entry(title=title, data=user_input)
@ -177,7 +195,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
assert self._reauth_entry
if user_input is not None:
user_input = {**self._reauth_entry.data, **user_input}
if not (errors := await validate_input(user_input)):
if not (errors := await validate_input(self.hass, user_input)):
self.hass.config_entries.async_update_entry(
self._reauth_entry, data=user_input
)
@ -231,7 +249,7 @@ class OptionsFlow(config_entries.OptionsFlowWithConfigEntry):
errors = {"base": err.reason}
else:
entry_data.update(user_input)
errors = await validate_input(entry_data)
errors = await validate_input(self.hass, entry_data)
if not errors:
self.hass.config_entries.async_update_entry(
self.config_entry, data=entry_data

1
homeassistant/components/imap/const.py
View file

@ -9,6 +9,7 @@ CONF_FOLDER: Final = "folder"
CONF_SEARCH: Final = "search"
CONF_CHARSET: Final = "charset"
CONF_MAX_MESSAGE_SIZE = "max_message_size"
CONF_CUSTOM_EVENT_DATA_TEMPLATE: Final = "custom_event_data_template"
CONF_SSL_CIPHER_LIST: Final = "ssl_cipher_list"
DEFAULT_PORT: Final = 993

60
homeassistant/components/imap/coordinator.py
View file

@ -19,13 +19,19 @@ from homeassistant.const import (
CONTENT_TYPE_TEXT_PLAIN,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryError
from homeassistant.exceptions import (
ConfigEntryAuthFailed,
ConfigEntryError,
TemplateError,
)
from homeassistant.helpers.json import json_bytes
from homeassistant.helpers.template import Template
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from homeassistant.util.ssl import SSLCipherList, client_context
from .const import (
CONF_CHARSET,
CONF_CUSTOM_EVENT_DATA_TEMPLATE,
CONF_FOLDER,
CONF_MAX_MESSAGE_SIZE,
CONF_SEARCH,
@ -145,16 +151,22 @@ class ImapDataUpdateCoordinator(DataUpdateCoordinator[int | None]):
"""Base class for imap client."""
config_entry: ConfigEntry
custom_event_template: Template | None
def __init__(
self,
hass: HomeAssistant,
imap_client: IMAP4_SSL,
entry: ConfigEntry,
update_interval: timedelta | None,
) -> None:
"""Initiate imap client."""
self.imap_client = imap_client
self._last_message_id: str | None = None
self.custom_event_template = None
_custom_event_template = entry.data.get(CONF_CUSTOM_EVENT_DATA_TEMPLATE)
if _custom_event_template is not None:
self.custom_event_template = Template(_custom_event_template, hass=hass)
super().__init__(
hass,
_LOGGER,
@ -181,15 +193,36 @@ class ImapDataUpdateCoordinator(DataUpdateCoordinator[int | None]):
"search": self.config_entry.data[CONF_SEARCH],
"folder": self.config_entry.data[CONF_FOLDER],
"date": message.date,
"text": message.text[
: self.config_entry.data.get(
CONF_MAX_MESSAGE_SIZE, DEFAULT_MAX_MESSAGE_SIZE
)
],
"text": message.text,
"sender": message.sender,
"subject": message.subject,
"headers": message.headers,
}
if self.custom_event_template is not None:
try:
data["custom"] = self.custom_event_template.async_render(
data, parse_result=True
)
_LOGGER.debug(
"imap custom template (%s) for msgid %s rendered to: %s",
self.custom_event_template,
last_message_id,
data["custom"],
)
except TemplateError as err:
data["custom"] = None
_LOGGER.error(
"Error rendering imap custom template (%s) for msgid %s "
"failed with message: %s",
self.custom_event_template,
last_message_id,
err,
)
data["text"] = message.text[
: self.config_entry.data.get(
CONF_MAX_MESSAGE_SIZE, DEFAULT_MAX_MESSAGE_SIZE
)
]
if (size := len(json_bytes(data))) > MAX_EVENT_DATA_BYTES:
_LOGGER.warning(
"Custom imap_content event skipped, size (%s) exceeds "
@ -203,7 +236,8 @@ class ImapDataUpdateCoordinator(DataUpdateCoordinator[int | None]):
self.hass.bus.fire(EVENT_IMAP, data)
_LOGGER.debug(
"Message processed, sender: %s, subject: %s",
"Message with id %s processed, sender: %s, subject: %s",
last_message_id,
message.sender,
message.subject,
)
@ -260,9 +294,11 @@ class ImapDataUpdateCoordinator(DataUpdateCoordinator[int | None]):
class ImapPollingDataUpdateCoordinator(ImapDataUpdateCoordinator):
"""Class for imap client."""
def __init__(self, hass: HomeAssistant, imap_client: IMAP4_SSL) -> None:
def __init__(
self, hass: HomeAssistant, imap_client: IMAP4_SSL, entry: ConfigEntry
) -> None:
"""Initiate imap client."""
super().__init__(hass, imap_client, timedelta(seconds=10))
super().__init__(hass, imap_client, entry, timedelta(seconds=10))
async def _async_update_data(self) -> int | None:
"""Update the number of unread emails."""
@ -291,9 +327,11 @@ class ImapPollingDataUpdateCoordinator(ImapDataUpdateCoordinator):
class ImapPushDataUpdateCoordinator(ImapDataUpdateCoordinator):
"""Class for imap client."""
def __init__(self, hass: HomeAssistant, imap_client: IMAP4_SSL) -> None:
def __init__(
self, hass: HomeAssistant, imap_client: IMAP4_SSL, entry: ConfigEntry
) -> None:
"""Initiate imap client."""
super().__init__(hass, imap_client, None)
super().__init__(hass, imap_client, entry, None)
self._push_wait_task: asyncio.Task[None] | None = None
async def _async_update_data(self) -> int | None:

4
homeassistant/components/imap/strings.json
View file

@ -40,6 +40,7 @@
"data": {
"folder": "[%key:component::imap::config::step::user::data::folder%]",
"search": "[%key:component::imap::config::step::user::data::search%]",
"custom_event_data_template": "Template to create custom event data",
"max_message_size": "Max message size (2048 < size < 30000)"
}
}
@ -50,7 +51,8 @@
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_charset": "[%key:component::imap::config::error::invalid_charset%]",
"invalid_folder": "[%key:component::imap::config::error::invalid_folder%]",
"invalid_search": "[%key:component::imap::config::error::invalid_search%]"
"invalid_search": "[%key:component::imap::config::error::invalid_search%]",
"invalid_template": "Invalid template"
}
},
"selector": {

25
tests/components/imap/test_config_flow.py
View file

@ -404,6 +404,21 @@ async def test_key_options_in_options_form(hass: HomeAssistant) -> None:
({"max_message_size": "8192"}, data_entry_flow.FlowResultType.CREATE_ENTRY),
({"max_message_size": "1024"}, data_entry_flow.FlowResultType.FORM),
({"max_message_size": "65536"}, data_entry_flow.FlowResultType.FORM),
(
{"custom_event_data_template": "{{ subject }}"},
data_entry_flow.FlowResultType.CREATE_ENTRY,
),
(
{"custom_event_data_template": "{{ invalid_syntax"},
data_entry_flow.FlowResultType.FORM,
),
],
ids=[
"valid_message_size",
"invalid_message_size_low",
"invalid_message_size_high",
"valid_template",
"invalid_template",
],
)
async def test_advanced_options_form(
@ -438,9 +453,13 @@ async def test_advanced_options_form(
result["flow_id"], new_config
)
assert result2["type"] == assert_result
# Check if entry was updated
for key, value in new_config.items():
assert str(entry.data[key]) == value
if result2.get("errors") is not None:
assert assert_result == data_entry_flow.FlowResultType.FORM
else:
# Check if entry was updated
for key, value in new_config.items():
assert str(entry.data[key]) == value
except vol.MultipleInvalid:
# Check if form was expected with these options
assert assert_result == data_entry_flow.FlowResultType.FORM

54
tests/components/imap/test_init.py
View file

@ -505,3 +505,57 @@ async def test_message_is_truncated(
event_data = event_called[0].data
assert len(event_data["text"]) == 3
@pytest.mark.parametrize(
("imap_search", "imap_fetch"),
[(TEST_SEARCH_RESPONSE, TEST_FETCH_RESPONSE_TEXT_PLAIN)],
ids=["plain"],
)
@pytest.mark.parametrize("imap_has_capability", [True, False], ids=["push", "poll"])
@pytest.mark.parametrize(
("custom_template", "result", "error"),
[
("{{ subject }}", "Test subject", None),
('{{ "@example.com" in sender }}', True, None),
("{% bad template }}", None, "Error rendering imap custom template"),
],
ids=["subject_test", "sender_filter", "template_error"],
)
async def test_custom_template(
hass: HomeAssistant,
mock_imap_protocol: MagicMock,
caplog: pytest.LogCaptureFixture,
custom_template: str,
result: str | bool | None,
error: str | None,
) -> None:
"""Test the custom template event data."""
event_called = async_capture_events(hass, "imap_content")
config = MOCK_CONFIG.copy()
config["custom_event_data_template"] = custom_template
config_entry = MockConfigEntry(domain=DOMAIN, data=config)
config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
# Make sure we have had one update (when polling)
async_fire_time_changed(hass, utcnow() + timedelta(seconds=5))
await hass.async_block_till_done()
state = hass.states.get("sensor.imap_email_email_com")
# we should have received one message
assert state is not None
assert state.state == "1"
# we should have received one event
assert len(event_called) == 1
data: dict[str, Any] = event_called[0].data
assert data["server"] == "imap.server.com"
assert data["username"] == "email@email.com"
assert data["search"] == "UnSeen UnDeleted"
assert data["folder"] == "INBOX"
assert data["sender"] == "john.doe@example.com"
assert data["subject"] == "Test subject"
assert data["text"]
assert data["custom"] == result
assert error in caplog.text if error is not None else True