Don't allow listening to state_reported in event triggers (#114191)

Co-authored-by: Robert Svensson <Kane610@users.noreply.github.com>
This commit is contained in:
Erik Montnemery 2024-03-26 10:35:47 +01:00 committed by GitHub
parent 788813aad6
commit bac527f289
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 84 additions and 2 deletions

View File

@ -7,8 +7,9 @@ from typing import Any
import voluptuous as vol
from homeassistant.const import CONF_EVENT_DATA, CONF_PLATFORM
from homeassistant.const import CONF_EVENT_DATA, CONF_PLATFORM, EVENT_STATE_REPORTED
from homeassistant.core import CALLBACK_TYPE, Event, HassJob, HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, template
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType
@ -16,10 +17,24 @@ from homeassistant.helpers.typing import ConfigType
CONF_EVENT_TYPE = "event_type"
CONF_EVENT_CONTEXT = "context"
def _validate_event_types(value: Any) -> Any:
"""Validate the event types.
If the event types are templated, we check when attaching the trigger.
"""
templates: list[template.Template] = value
if any(tpl.is_static and tpl.template == EVENT_STATE_REPORTED for tpl in templates):
raise vol.Invalid(f"Can't listen to {EVENT_STATE_REPORTED} in event trigger")
return value
TRIGGER_SCHEMA = cv.TRIGGER_BASE_SCHEMA.extend(
{
vol.Required(CONF_PLATFORM): "event",
vol.Required(CONF_EVENT_TYPE): vol.All(cv.ensure_list, [cv.template]),
vol.Required(CONF_EVENT_TYPE): vol.All(
cv.ensure_list, [cv.template], _validate_event_types
),
vol.Optional(CONF_EVENT_DATA): vol.All(dict, cv.template_complex),
vol.Optional(CONF_EVENT_CONTEXT): vol.All(dict, cv.template_complex),
}
@ -49,6 +64,10 @@ async def async_attach_trigger(
event_types = template.render_complex(
config[CONF_EVENT_TYPE], variables, limited=True
)
if EVENT_STATE_REPORTED in event_types:
raise HomeAssistantError(
f"Can't listen to {EVENT_STATE_REPORTED} in event trigger"
)
event_data_schema: vol.Schema | None = None
event_data_items: ItemsView | None = None
if CONF_EVENT_DATA in config:

View File

@ -505,3 +505,66 @@ async def test_event_data_with_list(hass: HomeAssistant, calls) -> None:
hass.bus.async_fire("test_event", {"some_attr": [1, 2, 3]})
await hass.async_block_till_done()
assert len(calls) == 1
@pytest.mark.parametrize(
"event_type", ["state_reported", ["test_event", "state_reported"]]
)
async def test_state_reported_event(
hass: HomeAssistant, calls, caplog, event_type: list[str]
) -> None:
"""Test triggering on state reported event."""
context = Context()
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger": {"platform": "event", "event_type": event_type},
"action": {
"service": "test.automation",
"data_template": {"id": "{{ trigger.id}}"},
},
}
},
)
hass.bus.async_fire("test_event", context=context)
await hass.async_block_till_done()
assert len(calls) == 0
assert (
"Unnamed automation failed to setup triggers and has been disabled: Can't "
"listen to state_reported in event trigger for dictionary value @ "
"data['event_type']. Got None" in caplog.text
)
async def test_templated_state_reported_event(
hass: HomeAssistant, calls, caplog
) -> None:
"""Test triggering on state reported event."""
context = Context()
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: {
"trigger_variables": {"event_type": "state_reported"},
"trigger": {"platform": "event", "event_type": "{{event_type}}"},
"action": {
"service": "test.automation",
"data_template": {"id": "{{ trigger.id}}"},
},
}
},
)
hass.bus.async_fire("test_event", context=context)
await hass.async_block_till_done()
assert len(calls) == 0
assert (
"Got error 'Can't listen to state_reported in event trigger' "
"when setting up triggers for automation 0" in caplog.text
)