mirror of
https://github.com/home-assistant/core.git
synced 2025-09-09 06:41:33 +02:00
Add event platform to templates (#145408)
Co-authored-by: Norbert Rittel <norbert@rittel.de>
This commit is contained in:
@@ -17,6 +17,7 @@ from homeassistant.components.blueprint import (
|
||||
)
|
||||
from homeassistant.components.button import DOMAIN as DOMAIN_BUTTON
|
||||
from homeassistant.components.cover import DOMAIN as DOMAIN_COVER
|
||||
from homeassistant.components.event import DOMAIN as DOMAIN_EVENT
|
||||
from homeassistant.components.fan import DOMAIN as DOMAIN_FAN
|
||||
from homeassistant.components.image import DOMAIN as DOMAIN_IMAGE
|
||||
from homeassistant.components.light import DOMAIN as DOMAIN_LIGHT
|
||||
@@ -53,6 +54,7 @@ from . import (
|
||||
binary_sensor as binary_sensor_platform,
|
||||
button as button_platform,
|
||||
cover as cover_platform,
|
||||
event as event_platform,
|
||||
fan as fan_platform,
|
||||
image as image_platform,
|
||||
light as light_platform,
|
||||
@@ -124,6 +126,9 @@ CONFIG_SECTION_SCHEMA = vol.All(
|
||||
vol.Optional(DOMAIN_COVER): vol.All(
|
||||
cv.ensure_list, [cover_platform.COVER_YAML_SCHEMA]
|
||||
),
|
||||
vol.Optional(DOMAIN_EVENT): vol.All(
|
||||
cv.ensure_list, [event_platform.EVENT_YAML_SCHEMA]
|
||||
),
|
||||
vol.Optional(DOMAIN_FAN): vol.All(
|
||||
cv.ensure_list, [fan_platform.FAN_YAML_SCHEMA]
|
||||
),
|
||||
|
@@ -12,6 +12,7 @@ from homeassistant.components import websocket_api
|
||||
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
|
||||
from homeassistant.components.button import ButtonDeviceClass
|
||||
from homeassistant.components.cover import CoverDeviceClass
|
||||
from homeassistant.components.event import EventDeviceClass
|
||||
from homeassistant.components.sensor import (
|
||||
CONF_STATE_CLASS,
|
||||
DEVICE_CLASS_STATE_CLASSES,
|
||||
@@ -72,6 +73,7 @@ from .cover import (
|
||||
STOP_ACTION,
|
||||
async_create_preview_cover,
|
||||
)
|
||||
from .event import CONF_EVENT_TYPE, CONF_EVENT_TYPES, async_create_preview_event
|
||||
from .fan import (
|
||||
CONF_OFF_ACTION,
|
||||
CONF_ON_ACTION,
|
||||
@@ -203,6 +205,24 @@ def generate_schema(domain: str, flow_type: str) -> vol.Schema:
|
||||
)
|
||||
}
|
||||
|
||||
if domain == Platform.EVENT:
|
||||
schema |= {
|
||||
vol.Required(CONF_EVENT_TYPE): selector.TemplateSelector(),
|
||||
vol.Required(CONF_EVENT_TYPES): selector.TemplateSelector(),
|
||||
}
|
||||
|
||||
if flow_type == "config":
|
||||
schema |= {
|
||||
vol.Optional(CONF_DEVICE_CLASS): selector.SelectSelector(
|
||||
selector.SelectSelectorConfig(
|
||||
options=[cls.value for cls in EventDeviceClass],
|
||||
mode=selector.SelectSelectorMode.DROPDOWN,
|
||||
translation_key="event_device_class",
|
||||
sort=True,
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
if domain == Platform.FAN:
|
||||
schema |= _SCHEMA_STATE | {
|
||||
vol.Required(CONF_ON_ACTION): selector.ActionSelector(),
|
||||
@@ -441,6 +461,7 @@ TEMPLATE_TYPES = [
|
||||
Platform.BINARY_SENSOR,
|
||||
Platform.BUTTON,
|
||||
Platform.COVER,
|
||||
Platform.EVENT,
|
||||
Platform.FAN,
|
||||
Platform.IMAGE,
|
||||
Platform.LIGHT,
|
||||
@@ -473,6 +494,11 @@ CONFIG_FLOW = {
|
||||
preview="template",
|
||||
validate_user_input=validate_user_input(Platform.COVER),
|
||||
),
|
||||
Platform.EVENT: SchemaFlowFormStep(
|
||||
config_schema(Platform.EVENT),
|
||||
preview="template",
|
||||
validate_user_input=validate_user_input(Platform.EVENT),
|
||||
),
|
||||
Platform.FAN: SchemaFlowFormStep(
|
||||
config_schema(Platform.FAN),
|
||||
preview="template",
|
||||
@@ -542,6 +568,11 @@ OPTIONS_FLOW = {
|
||||
preview="template",
|
||||
validate_user_input=validate_user_input(Platform.COVER),
|
||||
),
|
||||
Platform.EVENT: SchemaFlowFormStep(
|
||||
options_schema(Platform.EVENT),
|
||||
preview="template",
|
||||
validate_user_input=validate_user_input(Platform.EVENT),
|
||||
),
|
||||
Platform.FAN: SchemaFlowFormStep(
|
||||
options_schema(Platform.FAN),
|
||||
preview="template",
|
||||
@@ -596,6 +627,7 @@ CREATE_PREVIEW_ENTITY: dict[
|
||||
Platform.ALARM_CONTROL_PANEL: async_create_preview_alarm_control_panel,
|
||||
Platform.BINARY_SENSOR: async_create_preview_binary_sensor,
|
||||
Platform.COVER: async_create_preview_cover,
|
||||
Platform.EVENT: async_create_preview_event,
|
||||
Platform.FAN: async_create_preview_fan,
|
||||
Platform.LIGHT: async_create_preview_light,
|
||||
Platform.LOCK: async_create_preview_lock,
|
||||
|
@@ -38,6 +38,7 @@ PLATFORMS = [
|
||||
Platform.BINARY_SENSOR,
|
||||
Platform.BUTTON,
|
||||
Platform.COVER,
|
||||
Platform.EVENT,
|
||||
Platform.FAN,
|
||||
Platform.IMAGE,
|
||||
Platform.LIGHT,
|
||||
|
235
homeassistant/components/template/event.py
Normal file
235
homeassistant/components/template/event.py
Normal file
@@ -0,0 +1,235 @@
|
||||
"""Support for events which integrates with other components."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any, Final
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.event import (
|
||||
DOMAIN as EVENT_DOMAIN,
|
||||
ENTITY_ID_FORMAT,
|
||||
EventDeviceClass,
|
||||
EventEntity,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_DEVICE_CLASS
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import (
|
||||
AddConfigEntryEntitiesCallback,
|
||||
AddEntitiesCallback,
|
||||
)
|
||||
from homeassistant.helpers.restore_state import RestoreEntity
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from . import TriggerUpdateCoordinator
|
||||
from .entity import AbstractTemplateEntity
|
||||
from .helpers import (
|
||||
async_setup_template_entry,
|
||||
async_setup_template_platform,
|
||||
async_setup_template_preview,
|
||||
)
|
||||
from .template_entity import (
|
||||
TEMPLATE_ENTITY_COMMON_CONFIG_ENTRY_SCHEMA,
|
||||
TemplateEntity,
|
||||
make_template_entity_common_modern_attributes_schema,
|
||||
)
|
||||
from .trigger_entity import TriggerEntity
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_NAME = "Template Event"
|
||||
|
||||
CONF_EVENT_TYPE = "event_type"
|
||||
CONF_EVENT_TYPES = "event_types"
|
||||
|
||||
DEVICE_CLASS_SCHEMA: Final = vol.All(vol.Lower, vol.Coerce(EventDeviceClass))
|
||||
|
||||
EVENT_COMMON_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Optional(CONF_DEVICE_CLASS): DEVICE_CLASS_SCHEMA,
|
||||
vol.Required(CONF_EVENT_TYPE): cv.template,
|
||||
vol.Required(CONF_EVENT_TYPES): cv.template,
|
||||
}
|
||||
)
|
||||
|
||||
EVENT_YAML_SCHEMA = EVENT_COMMON_SCHEMA.extend(
|
||||
make_template_entity_common_modern_attributes_schema(DEFAULT_NAME).schema
|
||||
)
|
||||
|
||||
|
||||
EVENT_CONFIG_ENTRY_SCHEMA = EVENT_COMMON_SCHEMA.extend(
|
||||
TEMPLATE_ENTITY_COMMON_CONFIG_ENTRY_SCHEMA.schema
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_platform(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
discovery_info: DiscoveryInfoType | None = None,
|
||||
) -> None:
|
||||
"""Set up the template event."""
|
||||
await async_setup_template_platform(
|
||||
hass,
|
||||
EVENT_DOMAIN,
|
||||
config,
|
||||
StateEventEntity,
|
||||
TriggerEventEntity,
|
||||
async_add_entities,
|
||||
discovery_info,
|
||||
)
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
config_entry: ConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Initialize config entry."""
|
||||
await async_setup_template_entry(
|
||||
hass,
|
||||
config_entry,
|
||||
async_add_entities,
|
||||
StateEventEntity,
|
||||
EVENT_CONFIG_ENTRY_SCHEMA,
|
||||
True,
|
||||
)
|
||||
|
||||
|
||||
@callback
|
||||
def async_create_preview_event(
|
||||
hass: HomeAssistant, name: str, config: dict[str, Any]
|
||||
) -> StateEventEntity:
|
||||
"""Create a preview event."""
|
||||
return async_setup_template_preview(
|
||||
hass,
|
||||
name,
|
||||
config,
|
||||
StateEventEntity,
|
||||
EVENT_CONFIG_ENTRY_SCHEMA,
|
||||
)
|
||||
|
||||
|
||||
class AbstractTemplateEvent(AbstractTemplateEntity, EventEntity):
|
||||
"""Representation of a template event features."""
|
||||
|
||||
_entity_id_format = ENTITY_ID_FORMAT
|
||||
|
||||
# The super init is not called because TemplateEntity and TriggerEntity will call AbstractTemplateEntity.__init__.
|
||||
# This ensures that the __init__ on AbstractTemplateEntity is not called twice.
|
||||
def __init__(self, config: dict[str, Any]) -> None: # pylint: disable=super-init-not-called
|
||||
"""Initialize the features."""
|
||||
self._event_type_template = config[CONF_EVENT_TYPE]
|
||||
self._event_types_template = config[CONF_EVENT_TYPES]
|
||||
|
||||
self._attr_device_class = config.get(CONF_DEVICE_CLASS)
|
||||
|
||||
self._event_type = None
|
||||
self._attr_event_types = []
|
||||
|
||||
@callback
|
||||
def _update_event_types(self, event_types: Any) -> None:
|
||||
"""Update the event types from the template."""
|
||||
if event_types in (None, "None", ""):
|
||||
self._attr_event_types = []
|
||||
return
|
||||
|
||||
if not isinstance(event_types, list):
|
||||
_LOGGER.error(
|
||||
("Received invalid event_types list: %s for entity %s. Expected list"),
|
||||
event_types,
|
||||
self.entity_id,
|
||||
)
|
||||
self._attr_event_types = []
|
||||
return
|
||||
|
||||
self._attr_event_types = [str(event_type) for event_type in event_types]
|
||||
|
||||
@callback
|
||||
def _update_event_type(self, event_type: Any) -> None:
|
||||
"""Update the effect from the template."""
|
||||
try:
|
||||
self._trigger_event(event_type)
|
||||
except ValueError:
|
||||
_LOGGER.error(
|
||||
"Received invalid event_type: %s for entity %s. Expected one of: %s",
|
||||
event_type,
|
||||
self.entity_id,
|
||||
self._attr_event_types,
|
||||
)
|
||||
|
||||
|
||||
class StateEventEntity(TemplateEntity, AbstractTemplateEvent):
|
||||
"""Representation of a template event."""
|
||||
|
||||
_attr_should_poll = False
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
config: dict[str, Any],
|
||||
unique_id: str | None,
|
||||
) -> None:
|
||||
"""Initialize the select."""
|
||||
TemplateEntity.__init__(self, hass, config, unique_id)
|
||||
AbstractTemplateEvent.__init__(self, config)
|
||||
|
||||
@callback
|
||||
def _async_setup_templates(self) -> None:
|
||||
"""Set up templates."""
|
||||
self.add_template_attribute(
|
||||
"_attr_event_types",
|
||||
self._event_types_template,
|
||||
None,
|
||||
self._update_event_types,
|
||||
none_on_template_error=True,
|
||||
)
|
||||
self.add_template_attribute(
|
||||
"_event_type",
|
||||
self._event_type_template,
|
||||
None,
|
||||
self._update_event_type,
|
||||
none_on_template_error=True,
|
||||
)
|
||||
super()._async_setup_templates()
|
||||
|
||||
|
||||
class TriggerEventEntity(TriggerEntity, AbstractTemplateEvent, RestoreEntity):
|
||||
"""Event entity based on trigger data."""
|
||||
|
||||
domain = EVENT_DOMAIN
|
||||
extra_template_keys_complex = (
|
||||
CONF_EVENT_TYPE,
|
||||
CONF_EVENT_TYPES,
|
||||
)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
coordinator: TriggerUpdateCoordinator,
|
||||
config: ConfigType,
|
||||
) -> None:
|
||||
"""Initialize the entity."""
|
||||
TriggerEntity.__init__(self, hass, coordinator, config)
|
||||
AbstractTemplateEvent.__init__(self, config)
|
||||
|
||||
@callback
|
||||
def _handle_coordinator_update(self) -> None:
|
||||
"""Handle update of the data."""
|
||||
self._process_data()
|
||||
|
||||
if not self.available:
|
||||
self.async_write_ha_state()
|
||||
return
|
||||
|
||||
for key, updater in (
|
||||
(CONF_EVENT_TYPES, self._update_event_types),
|
||||
(CONF_EVENT_TYPE, self._update_event_type),
|
||||
):
|
||||
updater(self._rendered[key])
|
||||
|
||||
self.async_set_context(self.coordinator.data["context"])
|
||||
self.async_write_ha_state()
|
@@ -136,6 +136,32 @@
|
||||
},
|
||||
"title": "Template cover"
|
||||
},
|
||||
"event": {
|
||||
"data": {
|
||||
"device_id": "[%key:common::config_flow::data::device%]",
|
||||
"name": "[%key:common::config_flow::data::name%]",
|
||||
"device_class": "[%key:component::template::common::device_class%]",
|
||||
"event_type": "Last fired event type",
|
||||
"event_types": "[%key:component::event::entity_component::_::state_attributes::event_types::name%]"
|
||||
},
|
||||
"data_description": {
|
||||
"device_id": "[%key:component::template::common::device_id_description%]",
|
||||
"event_type": "Defines a template for the type of the event.",
|
||||
"event_types": "Defines a template for a list of available event types."
|
||||
},
|
||||
"sections": {
|
||||
"advanced_options": {
|
||||
"name": "[%key:component::template::common::advanced_options%]",
|
||||
"data": {
|
||||
"availability": "[%key:component::template::common::availability%]"
|
||||
},
|
||||
"data_description": {
|
||||
"availability": "[%key:component::template::common::availability_description%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"title": "Template event"
|
||||
},
|
||||
"fan": {
|
||||
"data": {
|
||||
"device_id": "[%key:common::config_flow::data::device%]",
|
||||
@@ -358,6 +384,7 @@
|
||||
"binary_sensor": "Template a binary sensor",
|
||||
"button": "Template a button",
|
||||
"cover": "Template a cover",
|
||||
"event": "Template an event",
|
||||
"fan": "Template a fan",
|
||||
"image": "Template an image",
|
||||
"light": "Template a light",
|
||||
@@ -565,6 +592,31 @@
|
||||
},
|
||||
"title": "[%key:component::template::config::step::cover::title%]"
|
||||
},
|
||||
"event": {
|
||||
"data": {
|
||||
"device_id": "[%key:common::config_flow::data::device%]",
|
||||
"name": "[%key:common::config_flow::data::name%]",
|
||||
"event_type": "[%key:component::template::config::step::event::data::event_type%]",
|
||||
"event_types": "[%component::event::entity_component::_::state_attributes::event_types::name%]"
|
||||
},
|
||||
"data_description": {
|
||||
"device_id": "[%key:component::template::common::device_id_description%]",
|
||||
"event_type": "[%key:component::template::config::step::event::data_description::event_type%]",
|
||||
"event_types": "[%key:component::template::config::step::event::data_description::event_types%]"
|
||||
},
|
||||
"sections": {
|
||||
"advanced_options": {
|
||||
"name": "[%key:component::template::common::advanced_options%]",
|
||||
"data": {
|
||||
"availability": "[%key:component::template::common::availability%]"
|
||||
},
|
||||
"data_description": {
|
||||
"availability": "[%key:component::template::common::availability_description%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"title": "[%key:component::template::config::step::event::title%]"
|
||||
},
|
||||
"fan": {
|
||||
"data": {
|
||||
"device_id": "[%key:common::config_flow::data::device%]",
|
||||
@@ -905,6 +957,13 @@
|
||||
"window": "[%key:component::cover::entity_component::window::name%]"
|
||||
}
|
||||
},
|
||||
"event_device_class": {
|
||||
"options": {
|
||||
"doorbell": "[%key:component::event::entity_component::doorbell::name%]",
|
||||
"button": "[%key:component::event::entity_component::button::name%]",
|
||||
"motion": "[%key:component::event::entity_component::motion::name%]"
|
||||
}
|
||||
},
|
||||
"sensor_device_class": {
|
||||
"options": {
|
||||
"absolute_humidity": "[%key:component::sensor::entity_component::absolute_humidity::name%]",
|
||||
|
20
tests/components/template/snapshots/test_event.ambr
Normal file
20
tests/components/template/snapshots/test_event.ambr
Normal file
@@ -0,0 +1,20 @@
|
||||
# serializer version: 1
|
||||
# name: test_setup_config_entry
|
||||
StateSnapshot({
|
||||
'attributes': ReadOnlyDict({
|
||||
'event_type': 'single',
|
||||
'event_types': list([
|
||||
'single',
|
||||
'double',
|
||||
'hold',
|
||||
]),
|
||||
'friendly_name': 'template_event',
|
||||
}),
|
||||
'context': <ANY>,
|
||||
'entity_id': 'event.template_event',
|
||||
'last_changed': <ANY>,
|
||||
'last_reported': <ANY>,
|
||||
'last_updated': <ANY>,
|
||||
'state': '2024-07-09T00:00:00.000+00:00',
|
||||
})
|
||||
# ---
|
@@ -149,6 +149,16 @@ BINARY_SENSOR_OPTIONS = {
|
||||
},
|
||||
{},
|
||||
),
|
||||
(
|
||||
"event",
|
||||
{"event_type": "{{ states('event.one') }}"},
|
||||
"2024-07-09T00:00:00.000+00:00",
|
||||
{"one": "single", "two": "double"},
|
||||
{},
|
||||
{"event_types": "{{ ['single', 'double'] }}"},
|
||||
{"event_types": "{{ ['single', 'double'] }}"},
|
||||
{},
|
||||
),
|
||||
(
|
||||
"fan",
|
||||
{"state": "{{ states('fan.one') }}"},
|
||||
@@ -362,6 +372,12 @@ async def test_config_flow(
|
||||
{"set_cover_position": []},
|
||||
{"set_cover_position": []},
|
||||
),
|
||||
(
|
||||
"event",
|
||||
{"event_type": "{{ 'single' }}"},
|
||||
{"event_types": "{{ ['single', 'double'] }}"},
|
||||
{"event_types": "{{ ['single', 'double'] }}"},
|
||||
),
|
||||
(
|
||||
"fan",
|
||||
{"state": "{{ states('fan.one') }}"},
|
||||
@@ -582,6 +598,16 @@ async def test_config_flow_device(
|
||||
{"set_cover_position": []},
|
||||
"state",
|
||||
),
|
||||
(
|
||||
"event",
|
||||
{"event_type": "{{ states('event.one') }}"},
|
||||
{"event_type": "{{ states('event.two') }}"},
|
||||
["2024-07-09T00:00:00.000+00:00", "2024-07-09T00:00:00.000+00:00"],
|
||||
{"one": "single", "two": "double"},
|
||||
{"event_types": "{{ ['single', 'double'] }}"},
|
||||
{"event_types": "{{ ['single', 'double'] }}"},
|
||||
"event_type",
|
||||
),
|
||||
(
|
||||
"fan",
|
||||
{"state": "{{ states('fan.one') }}"},
|
||||
@@ -1469,6 +1495,12 @@ async def test_option_flow_sensor_preview_config_entry_removed(
|
||||
{"set_cover_position": []},
|
||||
{"set_cover_position": []},
|
||||
),
|
||||
(
|
||||
"event",
|
||||
{"event_type": "{{ 'single' }}"},
|
||||
{"event_types": "{{ ['single', 'double'] }}"},
|
||||
{"event_types": "{{ ['single', 'double'] }}"},
|
||||
),
|
||||
(
|
||||
"fan",
|
||||
{"state": "{{ states('fan.one') }}"},
|
||||
|
823
tests/components/template/test_event.py
Normal file
823
tests/components/template/test_event.py
Normal file
@@ -0,0 +1,823 @@
|
||||
"""The tests for the Template event platform."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components import event, template
|
||||
from homeassistant.const import (
|
||||
ATTR_ENTITY_PICTURE,
|
||||
ATTR_ICON,
|
||||
STATE_UNAVAILABLE,
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, State
|
||||
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||
from homeassistant.setup import async_setup_component
|
||||
|
||||
from .conftest import ConfigurationStyle, async_get_flow_preview_state
|
||||
|
||||
from tests.common import (
|
||||
MockConfigEntry,
|
||||
assert_setup_component,
|
||||
mock_restore_cache_with_extra_data,
|
||||
)
|
||||
from tests.conftest import WebSocketGenerator
|
||||
|
||||
TEST_OBJECT_ID = "template_event"
|
||||
TEST_ENTITY_ID = f"event.{TEST_OBJECT_ID}"
|
||||
TEST_SENSOR = "sensor.event"
|
||||
TEST_STATE_TRIGGER = {
|
||||
"trigger": {"trigger": "state", "entity_id": TEST_SENSOR},
|
||||
"variables": {"triggering_entity": "{{ trigger.entity_id }}"},
|
||||
"action": [
|
||||
{"event": "action_event", "event_data": {"what": "{{ triggering_entity }}"}}
|
||||
],
|
||||
}
|
||||
TEST_EVENT_TYPES_TEMPLATE = "{{ ['single', 'double', 'hold'] }}"
|
||||
TEST_EVENT_TYPE_TEMPLATE = "{{ 'single' }}"
|
||||
|
||||
TEST_EVENT_CONFIG = {
|
||||
"event_types": TEST_EVENT_TYPES_TEMPLATE,
|
||||
"event_type": TEST_EVENT_TYPE_TEMPLATE,
|
||||
}
|
||||
TEST_UNIQUE_ID_CONFIG = {
|
||||
**TEST_EVENT_CONFIG,
|
||||
"unique_id": "not-so-unique-anymore",
|
||||
}
|
||||
TEST_FROZEN_INPUT = "2024-07-09 00:00:00+00:00"
|
||||
TEST_FROZEN_STATE = "2024-07-09T00:00:00.000+00:00"
|
||||
|
||||
|
||||
async def async_setup_modern_format(
|
||||
hass: HomeAssistant,
|
||||
count: int,
|
||||
event_config: dict[str, Any],
|
||||
extra_config: dict[str, Any] | None,
|
||||
) -> None:
|
||||
"""Do setup of event integration via new format."""
|
||||
extra = extra_config if extra_config else {}
|
||||
config = {**event_config, **extra}
|
||||
|
||||
with assert_setup_component(count, template.DOMAIN):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
template.DOMAIN,
|
||||
{"template": {"event": config}},
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def async_setup_trigger_format(
|
||||
hass: HomeAssistant,
|
||||
count: int,
|
||||
event_config: dict[str, Any],
|
||||
extra_config: dict[str, Any] | None,
|
||||
) -> None:
|
||||
"""Do setup of event integration via trigger format."""
|
||||
extra = extra_config if extra_config else {}
|
||||
config = {
|
||||
"template": {
|
||||
**TEST_STATE_TRIGGER,
|
||||
"event": {**event_config, **extra},
|
||||
}
|
||||
}
|
||||
|
||||
with assert_setup_component(count, template.DOMAIN):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
template.DOMAIN,
|
||||
config,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
|
||||
async def async_setup_event_config(
|
||||
hass: HomeAssistant,
|
||||
count: int,
|
||||
style: ConfigurationStyle,
|
||||
event_config: dict[str, Any],
|
||||
extra_config: dict[str, Any] | None,
|
||||
) -> None:
|
||||
"""Do setup of event integration."""
|
||||
if style == ConfigurationStyle.MODERN:
|
||||
await async_setup_modern_format(hass, count, event_config, extra_config)
|
||||
elif style == ConfigurationStyle.TRIGGER:
|
||||
await async_setup_trigger_format(hass, count, event_config, extra_config)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def setup_base_event(
|
||||
hass: HomeAssistant,
|
||||
count: int,
|
||||
style: ConfigurationStyle,
|
||||
event_config: dict[str, Any],
|
||||
) -> None:
|
||||
"""Do setup of event integration."""
|
||||
await async_setup_event_config(
|
||||
hass,
|
||||
count,
|
||||
style,
|
||||
event_config,
|
||||
None,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def setup_event(
|
||||
hass: HomeAssistant,
|
||||
count: int,
|
||||
style: ConfigurationStyle,
|
||||
event_type_template: str,
|
||||
event_types_template: str,
|
||||
extra_config: dict[str, Any] | None,
|
||||
) -> None:
|
||||
"""Do setup of event integration."""
|
||||
await async_setup_event_config(
|
||||
hass,
|
||||
count,
|
||||
style,
|
||||
{
|
||||
"name": TEST_OBJECT_ID,
|
||||
"event_type": event_type_template,
|
||||
"event_types": event_types_template,
|
||||
},
|
||||
extra_config,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def setup_single_attribute_state_event(
|
||||
hass: HomeAssistant,
|
||||
count: int,
|
||||
style: ConfigurationStyle,
|
||||
event_type_template: str,
|
||||
event_types_template: str,
|
||||
attribute: str,
|
||||
attribute_template: str,
|
||||
) -> None:
|
||||
"""Do setup of event integration testing a single attribute."""
|
||||
extra = {attribute: attribute_template} if attribute and attribute_template else {}
|
||||
config = {
|
||||
"name": TEST_OBJECT_ID,
|
||||
"event_type": event_type_template,
|
||||
"event_types": event_types_template,
|
||||
}
|
||||
if style == ConfigurationStyle.MODERN:
|
||||
await async_setup_modern_format(hass, count, config, extra)
|
||||
elif style == ConfigurationStyle.TRIGGER:
|
||||
await async_setup_trigger_format(hass, count, config, extra)
|
||||
|
||||
|
||||
async def test_legacy_platform_config(hass: HomeAssistant) -> None:
|
||||
"""Test a legacy platform does not create event entities."""
|
||||
with assert_setup_component(1, event.DOMAIN):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
event.DOMAIN,
|
||||
{"event": {"platform": "template", "events": {TEST_OBJECT_ID: {}}}},
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
await hass.async_block_till_done()
|
||||
assert hass.states.async_all("event") == []
|
||||
|
||||
|
||||
@pytest.mark.freeze_time(TEST_FROZEN_INPUT)
|
||||
async def test_setup_config_entry(
|
||||
hass: HomeAssistant,
|
||||
snapshot: SnapshotAssertion,
|
||||
) -> None:
|
||||
"""Test the config flow."""
|
||||
|
||||
hass.states.async_set(
|
||||
TEST_SENSOR,
|
||||
"single",
|
||||
{},
|
||||
)
|
||||
|
||||
template_config_entry = MockConfigEntry(
|
||||
data={},
|
||||
domain=template.DOMAIN,
|
||||
options={
|
||||
"name": TEST_OBJECT_ID,
|
||||
"event_type": TEST_EVENT_TYPE_TEMPLATE,
|
||||
"event_types": TEST_EVENT_TYPES_TEMPLATE,
|
||||
"template_type": event.DOMAIN,
|
||||
},
|
||||
title="My template",
|
||||
)
|
||||
template_config_entry.add_to_hass(hass)
|
||||
|
||||
assert await hass.config_entries.async_setup(template_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state is not None
|
||||
assert state == snapshot
|
||||
|
||||
|
||||
@pytest.mark.freeze_time(TEST_FROZEN_INPUT)
|
||||
async def test_device_id(
|
||||
hass: HomeAssistant,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
entity_registry: er.EntityRegistry,
|
||||
) -> None:
|
||||
"""Test for device for Template."""
|
||||
|
||||
device_config_entry = MockConfigEntry()
|
||||
device_config_entry.add_to_hass(hass)
|
||||
device_entry = device_registry.async_get_or_create(
|
||||
config_entry_id=device_config_entry.entry_id,
|
||||
identifiers={("test", "identifier_test")},
|
||||
connections={("mac", "30:31:32:33:34:35")},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
assert device_entry is not None
|
||||
assert device_entry.id is not None
|
||||
|
||||
template_config_entry = MockConfigEntry(
|
||||
data={},
|
||||
domain=template.DOMAIN,
|
||||
options={
|
||||
"name": "My template",
|
||||
"event_type": TEST_EVENT_TYPE_TEMPLATE,
|
||||
"event_types": TEST_EVENT_TYPES_TEMPLATE,
|
||||
"template_type": "event",
|
||||
"device_id": device_entry.id,
|
||||
},
|
||||
title="My template",
|
||||
)
|
||||
template_config_entry.add_to_hass(hass)
|
||||
|
||||
assert await hass.config_entries.async_setup(template_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
template_entity = entity_registry.async_get("event.my_template")
|
||||
assert template_entity is not None
|
||||
assert template_entity.device_id == device_entry.id
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("count", "event_types_template", "extra_config"),
|
||||
[(1, TEST_EVENT_TYPES_TEMPLATE, None)],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("style", "expected_state"),
|
||||
[
|
||||
(ConfigurationStyle.MODERN, STATE_UNKNOWN),
|
||||
(ConfigurationStyle.TRIGGER, STATE_UNKNOWN),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("event_type_template", ["{{states.test['big.fat...']}}"])
|
||||
@pytest.mark.usefixtures("setup_event")
|
||||
async def test_event_type_syntax_error(
|
||||
hass: HomeAssistant,
|
||||
expected_state: str,
|
||||
) -> None:
|
||||
"""Test template event_type with render error."""
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state == expected_state
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("count", "event_type_template", "event_types_template", "extra_config"),
|
||||
[(1, "{{ states('sensor.event') }}", TEST_EVENT_TYPES_TEMPLATE, None)],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"style",
|
||||
[ConfigurationStyle.MODERN, ConfigurationStyle.TRIGGER],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("event", "expected"),
|
||||
[
|
||||
("single", "single"),
|
||||
("double", "double"),
|
||||
("hold", "hold"),
|
||||
],
|
||||
)
|
||||
@pytest.mark.usefixtures("setup_event")
|
||||
async def test_event_type_template(
|
||||
hass: HomeAssistant,
|
||||
event: str,
|
||||
expected: str,
|
||||
) -> None:
|
||||
"""Test template event_type."""
|
||||
hass.states.async_set(TEST_SENSOR, event)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.attributes["event_type"] == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("count", "event_type_template", "event_types_template", "extra_config"),
|
||||
[(1, "{{ states('sensor.event') }}", TEST_EVENT_TYPES_TEMPLATE, None)],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"style",
|
||||
[ConfigurationStyle.MODERN, ConfigurationStyle.TRIGGER],
|
||||
)
|
||||
@pytest.mark.usefixtures("setup_event")
|
||||
@pytest.mark.freeze_time(TEST_FROZEN_INPUT)
|
||||
async def test_event_type_template_updates(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test template event_type updates."""
|
||||
hass.states.async_set(TEST_SENSOR, "single")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state == TEST_FROZEN_STATE
|
||||
assert state.attributes["event_type"] == "single"
|
||||
|
||||
hass.states.async_set(TEST_SENSOR, "double")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state == TEST_FROZEN_STATE
|
||||
assert state.attributes["event_type"] == "double"
|
||||
|
||||
hass.states.async_set(TEST_SENSOR, "hold")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state == TEST_FROZEN_STATE
|
||||
assert state.attributes["event_type"] == "hold"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("count", "event_types_template", "extra_config"),
|
||||
[(1, TEST_EVENT_TYPES_TEMPLATE, None)],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"style",
|
||||
[ConfigurationStyle.MODERN, ConfigurationStyle.TRIGGER],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"event_type_template",
|
||||
[
|
||||
"{{ None }}",
|
||||
"{{ 7 }}",
|
||||
"{{ 'unknown' }}",
|
||||
"{{ 'tripple_double' }}",
|
||||
],
|
||||
)
|
||||
@pytest.mark.usefixtures("setup_event")
|
||||
async def test_event_type_invalid(
|
||||
hass: HomeAssistant,
|
||||
) -> None:
|
||||
"""Test template event_type."""
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state == STATE_UNKNOWN
|
||||
assert state.attributes["event_type"] is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("count", "event_type_template", "event_types_template"),
|
||||
[(1, "{{ states('sensor.event') }}", TEST_EVENT_TYPES_TEMPLATE)],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"style",
|
||||
[ConfigurationStyle.MODERN, ConfigurationStyle.TRIGGER],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("attribute", "attribute_template", "key", "expected"),
|
||||
[
|
||||
(
|
||||
"picture",
|
||||
"{% if is_state('sensor.event', 'double') %}something{% endif %}",
|
||||
ATTR_ENTITY_PICTURE,
|
||||
"something",
|
||||
),
|
||||
(
|
||||
"icon",
|
||||
"{% if is_state('sensor.event', 'double') %}mdi:something{% endif %}",
|
||||
ATTR_ICON,
|
||||
"mdi:something",
|
||||
),
|
||||
],
|
||||
)
|
||||
@pytest.mark.usefixtures("setup_single_attribute_state_event")
|
||||
async def test_entity_picture_and_icon_templates(
|
||||
hass: HomeAssistant, key: str, expected: str
|
||||
) -> None:
|
||||
"""Test picture and icon template."""
|
||||
state = hass.states.async_set(TEST_SENSOR, "single")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.attributes.get(key) in ("", None)
|
||||
|
||||
state = hass.states.async_set(TEST_SENSOR, "double")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
|
||||
assert state.attributes[key] == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("count", "event_type_template", "extra_config"),
|
||||
[(1, "{{ None }}", None)],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"style",
|
||||
[ConfigurationStyle.MODERN, ConfigurationStyle.TRIGGER],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
("event_types_template", "expected"),
|
||||
[
|
||||
(
|
||||
"{{ ['Strobe color', 'Police', 'Christmas', 'RGB', 'Random Loop'] }}",
|
||||
["Strobe color", "Police", "Christmas", "RGB", "Random Loop"],
|
||||
),
|
||||
(
|
||||
"{{ ['Police', 'RGB', 'Random Loop'] }}",
|
||||
["Police", "RGB", "Random Loop"],
|
||||
),
|
||||
("{{ [] }}", []),
|
||||
("{{ '[]' }}", []),
|
||||
("{{ 124 }}", []),
|
||||
("{{ '124' }}", []),
|
||||
("{{ none }}", []),
|
||||
("", []),
|
||||
],
|
||||
)
|
||||
@pytest.mark.usefixtures("setup_event")
|
||||
async def test_event_types_template(hass: HomeAssistant, expected: str) -> None:
|
||||
"""Test template event_types."""
|
||||
hass.states.async_set(TEST_SENSOR, "anything")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.attributes["event_types"] == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("count", "event_type_template", "event_types_template", "extra_config"),
|
||||
[
|
||||
(
|
||||
1,
|
||||
"{{ states('sensor.event') }}",
|
||||
"{{ state_attr('sensor.event', 'options') or ['unknown'] }}",
|
||||
None,
|
||||
)
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"style",
|
||||
[ConfigurationStyle.MODERN, ConfigurationStyle.TRIGGER],
|
||||
)
|
||||
@pytest.mark.usefixtures("setup_event")
|
||||
@pytest.mark.freeze_time(TEST_FROZEN_INPUT)
|
||||
async def test_event_types_template_updates(hass: HomeAssistant) -> None:
|
||||
"""Test template event_type update with entity."""
|
||||
hass.states.async_set(
|
||||
TEST_SENSOR, "single", {"options": ["single", "double", "hold"]}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state == TEST_FROZEN_STATE
|
||||
assert state.attributes["event_type"] == "single"
|
||||
assert state.attributes["event_types"] == ["single", "double", "hold"]
|
||||
|
||||
hass.states.async_set(TEST_SENSOR, "double", {"options": ["double", "hold"]})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state == TEST_FROZEN_STATE
|
||||
assert state.attributes["event_type"] == "double"
|
||||
assert state.attributes["event_types"] == ["double", "hold"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
(
|
||||
"count",
|
||||
"event_type_template",
|
||||
"event_types_template",
|
||||
"attribute",
|
||||
"attribute_template",
|
||||
),
|
||||
[
|
||||
(
|
||||
1,
|
||||
"{{ states('sensor.event') }}",
|
||||
TEST_EVENT_TYPES_TEMPLATE,
|
||||
"availability",
|
||||
"{{ states('sensor.event') in ['single', 'double', 'hold'] }}",
|
||||
)
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"style",
|
||||
[ConfigurationStyle.MODERN, ConfigurationStyle.TRIGGER],
|
||||
)
|
||||
@pytest.mark.usefixtures("setup_single_attribute_state_event")
|
||||
async def test_available_template_with_entities(hass: HomeAssistant) -> None:
|
||||
"""Test availability templates with values from other entities."""
|
||||
hass.states.async_set(TEST_SENSOR, "single")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state != STATE_UNAVAILABLE
|
||||
assert state.attributes["event_type"] == "single"
|
||||
|
||||
hass.states.async_set(TEST_SENSOR, "triple")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state == STATE_UNAVAILABLE
|
||||
assert "event_type" not in state.attributes
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("count", "domain"), [(1, "template")])
|
||||
@pytest.mark.parametrize(
|
||||
"config",
|
||||
[
|
||||
{
|
||||
"template": {
|
||||
"trigger": {"platform": "event", "event_type": "test_event"},
|
||||
"event": {
|
||||
"name": TEST_OBJECT_ID,
|
||||
"event_type": "{{ trigger.event.data.action }}",
|
||||
"event_types": TEST_EVENT_TYPES_TEMPLATE,
|
||||
"picture": "{{ '/local/dogs.png' }}",
|
||||
"icon": "{{ 'mdi:pirate' }}",
|
||||
"attributes": {
|
||||
"plus_one": "{{ trigger.event.data.beer + 1 }}",
|
||||
"plus_two": "{{ trigger.event.data.beer + 2 }}",
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
],
|
||||
)
|
||||
async def test_trigger_entity_restore_state(
|
||||
hass: HomeAssistant,
|
||||
count: int,
|
||||
domain: str,
|
||||
config: dict,
|
||||
) -> None:
|
||||
"""Test restoring trigger event entities."""
|
||||
restored_attributes = {
|
||||
"entity_picture": "/local/cats.png",
|
||||
"event_type": "hold",
|
||||
"icon": "mdi:ship",
|
||||
"plus_one": 55,
|
||||
}
|
||||
fake_state = State(
|
||||
TEST_ENTITY_ID,
|
||||
"2021-01-01T23:59:59.123+00:00",
|
||||
restored_attributes,
|
||||
)
|
||||
fake_extra_data = {
|
||||
"last_event_type": "hold",
|
||||
"last_event_attributes": restored_attributes,
|
||||
}
|
||||
mock_restore_cache_with_extra_data(hass, ((fake_state, fake_extra_data),))
|
||||
with assert_setup_component(count, domain):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
domain,
|
||||
config,
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
test_state = "2021-01-01T23:59:59.123+00:00"
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state == test_state
|
||||
for attr, value in restored_attributes.items():
|
||||
assert state.attributes[attr] == value
|
||||
assert "plus_two" not in state.attributes
|
||||
|
||||
hass.bus.async_fire("test_event", {"action": "double", "beer": 2})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state != test_state
|
||||
assert state.attributes["icon"] == "mdi:pirate"
|
||||
assert state.attributes["entity_picture"] == "/local/dogs.png"
|
||||
assert state.attributes["event_type"] == "double"
|
||||
assert state.attributes["event_types"] == ["single", "double", "hold"]
|
||||
assert state.attributes["plus_one"] == 3
|
||||
assert state.attributes["plus_two"] == 4
|
||||
|
||||
|
||||
@pytest.mark.parametrize(("count", "domain"), [(1, "template")])
|
||||
@pytest.mark.parametrize(
|
||||
"config",
|
||||
[
|
||||
{
|
||||
"template": {
|
||||
"event": {
|
||||
"name": TEST_OBJECT_ID,
|
||||
"event_type": "{{ states('sensor.event') }}",
|
||||
"event_types": TEST_EVENT_TYPES_TEMPLATE,
|
||||
},
|
||||
},
|
||||
},
|
||||
],
|
||||
)
|
||||
async def test_event_entity_restore_state(
|
||||
hass: HomeAssistant,
|
||||
count: int,
|
||||
domain: str,
|
||||
config: dict,
|
||||
) -> None:
|
||||
"""Test restoring trigger event entities."""
|
||||
fake_state = State(
|
||||
TEST_ENTITY_ID,
|
||||
"2021-01-01T23:59:59.123+00:00",
|
||||
{},
|
||||
)
|
||||
fake_extra_data = {
|
||||
"last_event_type": "hold",
|
||||
"last_event_attributes": {},
|
||||
}
|
||||
mock_restore_cache_with_extra_data(hass, ((fake_state, fake_extra_data),))
|
||||
with assert_setup_component(count, domain):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
domain,
|
||||
config,
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
test_state = "2021-01-01T23:59:59.123+00:00"
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state == test_state
|
||||
|
||||
hass.states.async_set(TEST_SENSOR, "double")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get(TEST_ENTITY_ID)
|
||||
assert state.state != test_state
|
||||
assert state.attributes["event_type"] == "double"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
(
|
||||
"count",
|
||||
"event_type_template",
|
||||
"event_types_template",
|
||||
"attribute",
|
||||
"attribute_template",
|
||||
),
|
||||
[
|
||||
(
|
||||
1,
|
||||
TEST_EVENT_TYPE_TEMPLATE,
|
||||
TEST_EVENT_TYPES_TEMPLATE,
|
||||
"availability",
|
||||
"{{ x - 12 }}",
|
||||
)
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize(
|
||||
"style",
|
||||
[ConfigurationStyle.MODERN, ConfigurationStyle.TRIGGER],
|
||||
)
|
||||
@pytest.mark.usefixtures("setup_single_attribute_state_event")
|
||||
async def test_invalid_availability_template_keeps_component_available(
|
||||
hass: HomeAssistant,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
caplog_setup_text,
|
||||
) -> None:
|
||||
"""Test that an invalid availability keeps the device available."""
|
||||
hass.states.async_set(TEST_SENSOR, "anything")
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get(TEST_ENTITY_ID).state != STATE_UNAVAILABLE
|
||||
|
||||
error = "UndefinedError: 'x' is undefined"
|
||||
assert error in caplog_setup_text or error in caplog.text
|
||||
|
||||
|
||||
@pytest.mark.parametrize("count", [1])
|
||||
@pytest.mark.parametrize(
|
||||
("events", "style"),
|
||||
[
|
||||
(
|
||||
[
|
||||
{
|
||||
"name": "test_template_event_01",
|
||||
**TEST_UNIQUE_ID_CONFIG,
|
||||
},
|
||||
{
|
||||
"name": "test_template_event_02",
|
||||
**TEST_UNIQUE_ID_CONFIG,
|
||||
},
|
||||
],
|
||||
ConfigurationStyle.MODERN,
|
||||
),
|
||||
(
|
||||
[
|
||||
{
|
||||
"name": "test_template_event_01",
|
||||
**TEST_UNIQUE_ID_CONFIG,
|
||||
},
|
||||
{
|
||||
"name": "test_template_event_02",
|
||||
**TEST_UNIQUE_ID_CONFIG,
|
||||
},
|
||||
],
|
||||
ConfigurationStyle.TRIGGER,
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_unique_id(
|
||||
hass: HomeAssistant, count: int, events: list[dict], style: ConfigurationStyle
|
||||
) -> None:
|
||||
"""Test unique_id option only creates one event per id."""
|
||||
config = {"event": events}
|
||||
if style == ConfigurationStyle.TRIGGER:
|
||||
config = {**config, **TEST_STATE_TRIGGER}
|
||||
with assert_setup_component(count, template.DOMAIN):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
template.DOMAIN,
|
||||
{"template": config},
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_all("event")) == 1
|
||||
|
||||
|
||||
async def test_nested_unique_id(
|
||||
hass: HomeAssistant, entity_registry: er.EntityRegistry
|
||||
) -> None:
|
||||
"""Test unique_id option creates one event per nested id."""
|
||||
|
||||
with assert_setup_component(1, template.DOMAIN):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
template.DOMAIN,
|
||||
{
|
||||
"template": {
|
||||
"unique_id": "x",
|
||||
"event": [
|
||||
{
|
||||
"name": "test_a",
|
||||
**TEST_EVENT_CONFIG,
|
||||
"unique_id": "a",
|
||||
},
|
||||
{
|
||||
"name": "test_b",
|
||||
**TEST_EVENT_CONFIG,
|
||||
"unique_id": "b",
|
||||
},
|
||||
],
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
await hass.async_block_till_done()
|
||||
await hass.async_start()
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert len(hass.states.async_all("event")) == 2
|
||||
|
||||
entry = entity_registry.async_get("event.test_a")
|
||||
assert entry
|
||||
assert entry.unique_id == "x-a"
|
||||
|
||||
entry = entity_registry.async_get("event.test_b")
|
||||
assert entry
|
||||
assert entry.unique_id == "x-b"
|
||||
|
||||
|
||||
@pytest.mark.freeze_time(TEST_FROZEN_INPUT)
|
||||
async def test_flow_preview(
|
||||
hass: HomeAssistant,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
) -> None:
|
||||
"""Test the config flow preview."""
|
||||
|
||||
state = await async_get_flow_preview_state(
|
||||
hass,
|
||||
hass_ws_client,
|
||||
event.DOMAIN,
|
||||
{"name": "My template", **TEST_EVENT_CONFIG},
|
||||
)
|
||||
|
||||
assert state["state"] == TEST_FROZEN_STATE
|
||||
assert state["attributes"]["event_type"] == "single"
|
||||
assert state["attributes"]["event_types"] == ["single", "double", "hold"]
|
@@ -364,6 +364,18 @@ async def async_yaml_patch_helper(hass: HomeAssistant, filename: str) -> None:
|
||||
"value_template": "{{ true }}",
|
||||
},
|
||||
),
|
||||
(
|
||||
{
|
||||
"template_type": "event",
|
||||
"name": "My template",
|
||||
"event_type": "{{ 'single' }}",
|
||||
"event_types": "{{ ['single', 'double'] }}",
|
||||
},
|
||||
{
|
||||
"event_type": "{{ 'single' }}",
|
||||
"event_types": "{{ ['single', 'double'] }}",
|
||||
},
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_change_device(
|
||||
|
Reference in New Issue
Block a user