mirror of
https://github.com/home-assistant/core.git
synced 2025-04-23 08:47:57 +00:00
Blueprints for template entities (#126971)
* Template domain blueprints * Default blueprint for templates * Some linting * Template entity updates * Load and use blueprints in config * Added missing mapping methods for templates * Linting * Added tests * Wrong schema type * Hassfest errors * More linting issues * Refactor based on desired schema In the [architecture discussion](https://github.com/home-assistant/architecture/discussions/1027), the template blueprint instance did not specify the platform (e.g. `binary_sensor`), but the initial implementation assumed that schema. * Create default template blueprints on first run * Moved TemplateConfig definition This is to avoid circular references * Corrected methods to find templates based on blueprints * Corrected missing entity config information * Added tests * Don't use hass.data Address comments https://github.com/home-assistant/core/pull/126971/#discussion_r1780097187 * Prevent creating blueprints during testing * Combine 2 ifs Address comment https://github.com/home-assistant/core/pull/126971/#discussion_r1780160870 * Improve test coverage * Prevent template component from dirtying test env * Remove useless hard-coded validation * Improve code coverage to 100% * Address review comments * Moved helpers in helpers.py As per comment https://github.com/home-assistant/core/pull/126971#discussion_r1786539889 * Fix blueprint source URL --------- Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
parent
7e6c106869
commit
d9b077154e
@ -8,6 +8,7 @@ from . import websocket_api
|
||||
from .const import CONF_USE_BLUEPRINT, DOMAIN # noqa: F401
|
||||
from .errors import ( # noqa: F401
|
||||
BlueprintException,
|
||||
BlueprintInUse,
|
||||
BlueprintWithNameException,
|
||||
FailedToLoad,
|
||||
InvalidBlueprint,
|
||||
@ -15,7 +16,11 @@ from .errors import ( # noqa: F401
|
||||
MissingInput,
|
||||
)
|
||||
from .models import Blueprint, BlueprintInputs, DomainBlueprints # noqa: F401
|
||||
from .schemas import BLUEPRINT_SCHEMA, is_blueprint_instance_config # noqa: F401
|
||||
from .schemas import ( # noqa: F401
|
||||
BLUEPRINT_INSTANCE_FIELDS,
|
||||
BLUEPRINT_SCHEMA,
|
||||
is_blueprint_instance_config,
|
||||
)
|
||||
|
||||
CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN)
|
||||
|
||||
|
@ -29,6 +29,7 @@ from homeassistant.util.hass_dict import HassKey
|
||||
|
||||
from .const import CONF_MAX, CONF_MIN, CONF_STEP, CONF_TRIGGER, DOMAIN, PLATFORMS
|
||||
from .coordinator import TriggerUpdateCoordinator
|
||||
from .helpers import async_get_blueprints
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
DATA_COORDINATORS: HassKey[list[TriggerUpdateCoordinator]] = HassKey(DOMAIN)
|
||||
@ -36,6 +37,17 @@ DATA_COORDINATORS: HassKey[list[TriggerUpdateCoordinator]] = HassKey(DOMAIN)
|
||||
|
||||
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
"""Set up the template integration."""
|
||||
|
||||
# Register template as valid domain for Blueprint
|
||||
blueprints = async_get_blueprints(hass)
|
||||
|
||||
# Add some default blueprints to blueprints/template, does nothing
|
||||
# if blueprints/template already exists but still has to create
|
||||
# an executor job to check if the folder exists so we run it in a
|
||||
# separate task to avoid waiting for it to finish setting up
|
||||
# since a tracked task will be waited at the end of startup
|
||||
hass.async_create_task(blueprints.async_populate(), eager_start=True)
|
||||
|
||||
if DOMAIN in config:
|
||||
await _process_config(hass, config)
|
||||
|
||||
@ -136,7 +148,14 @@ async def _process_config(hass: HomeAssistant, hass_config: ConfigType) -> None:
|
||||
DOMAIN,
|
||||
{
|
||||
"unique_id": conf_section.get(CONF_UNIQUE_ID),
|
||||
"entities": conf_section[platform_domain],
|
||||
"entities": [
|
||||
{
|
||||
**entity_conf,
|
||||
"raw_blueprint_inputs": conf_section.raw_blueprint_inputs,
|
||||
"raw_configs": conf_section.raw_config,
|
||||
}
|
||||
for entity_conf in conf_section[platform_domain]
|
||||
],
|
||||
},
|
||||
hass_config,
|
||||
),
|
||||
|
@ -0,0 +1,27 @@
|
||||
blueprint:
|
||||
name: Invert a binary sensor
|
||||
description: Creates a binary_sensor which holds the inverted value of a reference binary_sensor
|
||||
domain: template
|
||||
source_url: https://github.com/home-assistant/core/blob/dev/homeassistant/components/template/blueprints/inverted_binary_sensor.yaml
|
||||
input:
|
||||
reference_entity:
|
||||
name: Binary sensor to be inverted
|
||||
description: The binary_sensor which needs to have its value inverted
|
||||
selector:
|
||||
entity:
|
||||
domain: binary_sensor
|
||||
variables:
|
||||
reference_entity: !input reference_entity
|
||||
binary_sensor:
|
||||
state: >
|
||||
{% if states(reference_entity) == 'on' %}
|
||||
off
|
||||
{% elif states(reference_entity) == 'off' %}
|
||||
on
|
||||
{% else %}
|
||||
{{ states(reference_entity) }}
|
||||
{% endif %}
|
||||
# delay_on: not_used in this example
|
||||
# delay_off: not_used in this example
|
||||
# auto_off: not_used in this example
|
||||
availability: "{{ states(reference_entity) not in ('unknown', 'unavailable') }}"
|
@ -1,10 +1,15 @@
|
||||
"""Template config validator."""
|
||||
|
||||
from contextlib import suppress
|
||||
import logging
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
|
||||
from homeassistant.components.blueprint import (
|
||||
BLUEPRINT_INSTANCE_FIELDS,
|
||||
is_blueprint_instance_config,
|
||||
)
|
||||
from homeassistant.components.button import DOMAIN as BUTTON_DOMAIN
|
||||
from homeassistant.components.image import DOMAIN as IMAGE_DOMAIN
|
||||
from homeassistant.components.number import DOMAIN as NUMBER_DOMAIN
|
||||
@ -12,7 +17,13 @@ from homeassistant.components.select import DOMAIN as SELECT_DOMAIN
|
||||
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
|
||||
from homeassistant.components.weather import DOMAIN as WEATHER_DOMAIN
|
||||
from homeassistant.config import async_log_schema_error, config_without_domain
|
||||
from homeassistant.const import CONF_BINARY_SENSORS, CONF_SENSORS, CONF_UNIQUE_ID
|
||||
from homeassistant.const import (
|
||||
CONF_BINARY_SENSORS,
|
||||
CONF_NAME,
|
||||
CONF_SENSORS,
|
||||
CONF_UNIQUE_ID,
|
||||
CONF_VARIABLES,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.condition import async_validate_conditions_config
|
||||
@ -29,7 +40,15 @@ from . import (
|
||||
sensor as sensor_platform,
|
||||
weather as weather_platform,
|
||||
)
|
||||
from .const import CONF_ACTION, CONF_CONDITION, CONF_TRIGGER, DOMAIN
|
||||
from .const import (
|
||||
CONF_ACTION,
|
||||
CONF_CONDITION,
|
||||
CONF_TRIGGER,
|
||||
DOMAIN,
|
||||
PLATFORMS,
|
||||
TemplateConfig,
|
||||
)
|
||||
from .helpers import async_get_blueprints
|
||||
|
||||
PACKAGE_MERGE_HINT = "list"
|
||||
|
||||
@ -39,6 +58,7 @@ CONFIG_SECTION_SCHEMA = vol.Schema(
|
||||
vol.Optional(CONF_TRIGGER): cv.TRIGGER_SCHEMA,
|
||||
vol.Optional(CONF_CONDITION): cv.CONDITIONS_SCHEMA,
|
||||
vol.Optional(CONF_ACTION): cv.SCRIPT_SCHEMA,
|
||||
vol.Optional(CONF_VARIABLES): cv.SCRIPT_VARIABLES_SCHEMA,
|
||||
vol.Optional(NUMBER_DOMAIN): vol.All(
|
||||
cv.ensure_list, [number_platform.NUMBER_SCHEMA]
|
||||
),
|
||||
@ -66,9 +86,73 @@ CONFIG_SECTION_SCHEMA = vol.Schema(
|
||||
vol.Optional(WEATHER_DOMAIN): vol.All(
|
||||
cv.ensure_list, [weather_platform.WEATHER_SCHEMA]
|
||||
),
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
TEMPLATE_BLUEPRINT_INSTANCE_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Optional(CONF_NAME): cv.string,
|
||||
vol.Optional(CONF_UNIQUE_ID): cv.string,
|
||||
}
|
||||
).extend(BLUEPRINT_INSTANCE_FIELDS.schema)
|
||||
|
||||
|
||||
async def _async_resolve_blueprints(
|
||||
hass: HomeAssistant,
|
||||
config: ConfigType,
|
||||
) -> TemplateConfig:
|
||||
"""If a config item requires a blueprint, resolve that item to an actual config."""
|
||||
raw_config = None
|
||||
raw_blueprint_inputs = None
|
||||
|
||||
with suppress(ValueError): # Invalid config
|
||||
raw_config = dict(config)
|
||||
|
||||
if is_blueprint_instance_config(config):
|
||||
config = TEMPLATE_BLUEPRINT_INSTANCE_SCHEMA(config)
|
||||
blueprints = async_get_blueprints(hass)
|
||||
|
||||
blueprint_inputs = await blueprints.async_inputs_from_config(config)
|
||||
raw_blueprint_inputs = blueprint_inputs.config_with_inputs
|
||||
|
||||
config = blueprint_inputs.async_substitute()
|
||||
|
||||
platforms = [platform for platform in PLATFORMS if platform in config]
|
||||
if len(platforms) > 1:
|
||||
raise vol.Invalid("more than one platform defined per blueprint")
|
||||
if len(platforms) == 1:
|
||||
platform = platforms.pop()
|
||||
for prop in (CONF_NAME, CONF_UNIQUE_ID, CONF_VARIABLES):
|
||||
if prop in config:
|
||||
config[platform][prop] = config.pop(prop)
|
||||
raw_config = dict(config)
|
||||
|
||||
template_config = TemplateConfig(CONFIG_SECTION_SCHEMA(config))
|
||||
template_config.raw_blueprint_inputs = raw_blueprint_inputs
|
||||
template_config.raw_config = raw_config
|
||||
|
||||
return template_config
|
||||
|
||||
|
||||
async def async_validate_config_section(
|
||||
hass: HomeAssistant, config: ConfigType
|
||||
) -> TemplateConfig:
|
||||
"""Validate an entire config section for the template integration."""
|
||||
|
||||
validated_config = await _async_resolve_blueprints(hass, config)
|
||||
|
||||
if CONF_TRIGGER in validated_config:
|
||||
validated_config[CONF_TRIGGER] = await async_validate_trigger_config(
|
||||
hass, validated_config[CONF_TRIGGER]
|
||||
)
|
||||
|
||||
if CONF_CONDITION in validated_config:
|
||||
validated_config[CONF_CONDITION] = await async_validate_conditions_config(
|
||||
hass, validated_config[CONF_CONDITION]
|
||||
)
|
||||
|
||||
return validated_config
|
||||
|
||||
|
||||
async def async_validate_config(hass: HomeAssistant, config: ConfigType) -> ConfigType:
|
||||
"""Validate config."""
|
||||
@ -79,17 +163,9 @@ async def async_validate_config(hass: HomeAssistant, config: ConfigType) -> Conf
|
||||
|
||||
for cfg in cv.ensure_list(config[DOMAIN]):
|
||||
try:
|
||||
cfg = CONFIG_SECTION_SCHEMA(cfg)
|
||||
|
||||
if CONF_TRIGGER in cfg:
|
||||
cfg[CONF_TRIGGER] = await async_validate_trigger_config(
|
||||
hass, cfg[CONF_TRIGGER]
|
||||
)
|
||||
|
||||
if CONF_CONDITION in cfg:
|
||||
cfg[CONF_CONDITION] = await async_validate_conditions_config(
|
||||
hass, cfg[CONF_CONDITION]
|
||||
)
|
||||
template_config: TemplateConfig = await async_validate_config_section(
|
||||
hass, cfg
|
||||
)
|
||||
except vol.Invalid as err:
|
||||
async_log_schema_error(err, DOMAIN, cfg, hass)
|
||||
async_notify_setup_error(hass, DOMAIN)
|
||||
@ -109,7 +185,7 @@ async def async_validate_config(hass: HomeAssistant, config: ConfigType) -> Conf
|
||||
binary_sensor_platform.rewrite_legacy_to_modern_conf,
|
||||
),
|
||||
):
|
||||
if old_key not in cfg:
|
||||
if old_key not in template_config:
|
||||
continue
|
||||
|
||||
if not legacy_warn_printed:
|
||||
@ -121,11 +197,13 @@ async def async_validate_config(hass: HomeAssistant, config: ConfigType) -> Conf
|
||||
"https://www.home-assistant.io/integrations/template#configuration-for-trigger-based-template-sensors"
|
||||
)
|
||||
|
||||
definitions = list(cfg[new_key]) if new_key in cfg else []
|
||||
definitions.extend(transform(hass, cfg[old_key]))
|
||||
cfg = {**cfg, new_key: definitions}
|
||||
definitions = (
|
||||
list(template_config[new_key]) if new_key in template_config else []
|
||||
)
|
||||
definitions.extend(transform(hass, template_config[old_key]))
|
||||
template_config = TemplateConfig({**template_config, new_key: definitions})
|
||||
|
||||
config_sections.append(cfg)
|
||||
config_sections.append(template_config)
|
||||
|
||||
# Create a copy of the configuration with all config for current
|
||||
# component removed and add validated config back in.
|
||||
|
@ -1,6 +1,8 @@
|
||||
"""Constants for the Template Platform Components."""
|
||||
|
||||
from homeassistant.components.blueprint import BLUEPRINT_SCHEMA
|
||||
from homeassistant.const import Platform
|
||||
from homeassistant.helpers.typing import ConfigType
|
||||
|
||||
CONF_ACTION = "action"
|
||||
CONF_ATTRIBUTE_TEMPLATES = "attribute_templates"
|
||||
@ -38,3 +40,12 @@ PLATFORMS = [
|
||||
Platform.VACUUM,
|
||||
Platform.WEATHER,
|
||||
]
|
||||
|
||||
TEMPLATE_BLUEPRINT_SCHEMA = BLUEPRINT_SCHEMA
|
||||
|
||||
|
||||
class TemplateConfig(dict):
|
||||
"""Dummy class to allow adding attributes."""
|
||||
|
||||
raw_config: ConfigType | None = None
|
||||
raw_blueprint_inputs: ConfigType | None = None
|
||||
|
63
homeassistant/components/template/helpers.py
Normal file
63
homeassistant/components/template/helpers.py
Normal file
@ -0,0 +1,63 @@
|
||||
"""Helpers for template integration."""
|
||||
|
||||
import logging
|
||||
|
||||
from homeassistant.components import blueprint
|
||||
from homeassistant.const import SERVICE_RELOAD
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers.entity_platform import async_get_platforms
|
||||
from homeassistant.helpers.singleton import singleton
|
||||
|
||||
from .const import DOMAIN, TEMPLATE_BLUEPRINT_SCHEMA
|
||||
from .template_entity import TemplateEntity
|
||||
|
||||
DATA_BLUEPRINTS = "template_blueprints"
|
||||
|
||||
LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@callback
|
||||
def templates_with_blueprint(hass: HomeAssistant, blueprint_path: str) -> list[str]:
|
||||
"""Return all template entity ids that reference the blueprint."""
|
||||
return [
|
||||
entity_id
|
||||
for platform in async_get_platforms(hass, DOMAIN)
|
||||
for entity_id, template_entity in platform.entities.items()
|
||||
if isinstance(template_entity, TemplateEntity)
|
||||
and template_entity.referenced_blueprint == blueprint_path
|
||||
]
|
||||
|
||||
|
||||
@callback
|
||||
def blueprint_in_template(hass: HomeAssistant, entity_id: str) -> str | None:
|
||||
"""Return the blueprint the template entity is based on or None."""
|
||||
for platform in async_get_platforms(hass, DOMAIN):
|
||||
if isinstance(
|
||||
(template_entity := platform.entities.get(entity_id)), TemplateEntity
|
||||
):
|
||||
return template_entity.referenced_blueprint
|
||||
return None
|
||||
|
||||
|
||||
def _blueprint_in_use(hass: HomeAssistant, blueprint_path: str) -> bool:
|
||||
"""Return True if any template references the blueprint."""
|
||||
return len(templates_with_blueprint(hass, blueprint_path)) > 0
|
||||
|
||||
|
||||
async def _reload_blueprint_templates(hass: HomeAssistant, blueprint_path: str) -> None:
|
||||
"""Reload all templates that rely on a specific blueprint."""
|
||||
await hass.services.async_call(DOMAIN, SERVICE_RELOAD)
|
||||
|
||||
|
||||
@singleton(DATA_BLUEPRINTS)
|
||||
@callback
|
||||
def async_get_blueprints(hass: HomeAssistant) -> blueprint.DomainBlueprints:
|
||||
"""Get template blueprints."""
|
||||
return blueprint.DomainBlueprints(
|
||||
hass,
|
||||
DOMAIN,
|
||||
LOGGER,
|
||||
_blueprint_in_use,
|
||||
_reload_blueprint_templates,
|
||||
TEMPLATE_BLUEPRINT_SCHEMA,
|
||||
)
|
@ -4,6 +4,7 @@
|
||||
"after_dependencies": ["group"],
|
||||
"codeowners": ["@PhracturedBlue", "@tetienne", "@home-assistant/core"],
|
||||
"config_flow": true,
|
||||
"dependencies": ["blueprint"],
|
||||
"documentation": "https://www.home-assistant.io/integrations/template",
|
||||
"integration_type": "helper",
|
||||
"iot_class": "local_push",
|
||||
|
@ -6,17 +6,20 @@ from collections.abc import Callable, Mapping
|
||||
import contextlib
|
||||
import itertools
|
||||
import logging
|
||||
from typing import Any
|
||||
from typing import Any, cast
|
||||
|
||||
from propcache import under_cached_property
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.blueprint import CONF_USE_BLUEPRINT
|
||||
from homeassistant.const import (
|
||||
CONF_ENTITY_PICTURE_TEMPLATE,
|
||||
CONF_FRIENDLY_NAME,
|
||||
CONF_ICON,
|
||||
CONF_ICON_TEMPLATE,
|
||||
CONF_NAME,
|
||||
CONF_PATH,
|
||||
CONF_VARIABLES,
|
||||
STATE_UNKNOWN,
|
||||
)
|
||||
from homeassistant.core import (
|
||||
@ -77,6 +80,7 @@ TEMPLATE_ENTITY_COMMON_SCHEMA = vol.Schema(
|
||||
{
|
||||
vol.Optional(CONF_ATTRIBUTES): vol.Schema({cv.string: cv.template}),
|
||||
vol.Optional(CONF_AVAILABILITY): cv.template,
|
||||
vol.Optional(CONF_VARIABLES): cv.SCRIPT_VARIABLES_SCHEMA,
|
||||
}
|
||||
).extend(TEMPLATE_ENTITY_BASE_SCHEMA.schema)
|
||||
|
||||
@ -287,12 +291,16 @@ class TemplateEntity(Entity): # pylint: disable=hass-enforce-class-module
|
||||
self._icon_template = icon_template
|
||||
self._entity_picture_template = entity_picture_template
|
||||
self._friendly_name_template = None
|
||||
self._run_variables = {}
|
||||
self._blueprint_inputs = None
|
||||
else:
|
||||
self._attribute_templates = config.get(CONF_ATTRIBUTES)
|
||||
self._availability_template = config.get(CONF_AVAILABILITY)
|
||||
self._icon_template = config.get(CONF_ICON)
|
||||
self._entity_picture_template = config.get(CONF_PICTURE)
|
||||
self._friendly_name_template = config.get(CONF_NAME)
|
||||
self._run_variables = config.get(CONF_VARIABLES, {})
|
||||
self._blueprint_inputs = config.get("raw_blueprint_inputs")
|
||||
|
||||
class DummyState(State):
|
||||
"""None-state for template entities not yet added to the state machine."""
|
||||
@ -331,6 +339,18 @@ class TemplateEntity(Entity): # pylint: disable=hass-enforce-class-module
|
||||
variables=variables, parse_result=False
|
||||
)
|
||||
|
||||
@callback
|
||||
def _render_variables(self) -> dict:
|
||||
if isinstance(self._run_variables, dict):
|
||||
return self._run_variables
|
||||
|
||||
return self._run_variables.async_render(
|
||||
self.hass,
|
||||
{
|
||||
"this": TemplateStateFromEntityId(self.hass, self.entity_id),
|
||||
},
|
||||
)
|
||||
|
||||
@callback
|
||||
def _update_available(self, result: str | TemplateError) -> None:
|
||||
if isinstance(result, TemplateError):
|
||||
@ -360,6 +380,13 @@ class TemplateEntity(Entity): # pylint: disable=hass-enforce-class-module
|
||||
attribute_key, attribute_template, None, _update_attribute
|
||||
)
|
||||
|
||||
@property
|
||||
def referenced_blueprint(self) -> str | None:
|
||||
"""Return referenced blueprint or None."""
|
||||
if self._blueprint_inputs is None:
|
||||
return None
|
||||
return cast(str, self._blueprint_inputs[CONF_USE_BLUEPRINT][CONF_PATH])
|
||||
|
||||
def add_template_attribute(
|
||||
self,
|
||||
attribute: str,
|
||||
@ -459,7 +486,10 @@ class TemplateEntity(Entity): # pylint: disable=hass-enforce-class-module
|
||||
template_var_tups: list[TrackTemplate] = []
|
||||
has_availability_template = False
|
||||
|
||||
variables = {"this": TemplateStateFromEntityId(self.hass, self.entity_id)}
|
||||
variables = {
|
||||
"this": TemplateStateFromEntityId(self.hass, self.entity_id),
|
||||
**self._render_variables(),
|
||||
}
|
||||
|
||||
for template, attributes in self._template_attrs.items():
|
||||
template_var_tup = TrackTemplate(template, variables)
|
||||
@ -563,6 +593,7 @@ class TemplateEntity(Entity): # pylint: disable=hass-enforce-class-module
|
||||
await script.async_run(
|
||||
run_variables={
|
||||
"this": TemplateStateFromEntityId(self.hass, self.entity_id),
|
||||
**self._render_variables(),
|
||||
**run_variables,
|
||||
},
|
||||
context=context,
|
||||
|
@ -37,6 +37,11 @@ import homeassistant.util.dt as dt_util
|
||||
from tests.common import assert_setup_component, get_fixture_path
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
|
||||
def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
|
||||
"""Stub copying the blueprints to the config folder."""
|
||||
|
||||
|
||||
@pytest.fixture(name="values")
|
||||
def values_fixture() -> list[State]:
|
||||
"""Fixture for a list of test States."""
|
||||
|
@ -36,3 +36,8 @@ async def start_ha(
|
||||
async def caplog_setup_text(caplog: pytest.LogCaptureFixture) -> str:
|
||||
"""Return setup log of integration."""
|
||||
return caplog.text
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, name="stub_blueprint_populate")
|
||||
def stub_blueprint_populate_autouse(stub_blueprint_populate: None) -> None:
|
||||
"""Stub copying the blueprints to the config folder."""
|
||||
|
242
tests/components/template/test_blueprint.py
Normal file
242
tests/components/template/test_blueprint.py
Normal file
@ -0,0 +1,242 @@
|
||||
"""Test blueprints."""
|
||||
|
||||
from collections.abc import Iterator
|
||||
import contextlib
|
||||
from os import PathLike
|
||||
import pathlib
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components import template
|
||||
from homeassistant.components.blueprint import (
|
||||
BLUEPRINT_SCHEMA,
|
||||
Blueprint,
|
||||
BlueprintInUse,
|
||||
DomainBlueprints,
|
||||
)
|
||||
from homeassistant.components.template import DOMAIN, SERVICE_RELOAD
|
||||
from homeassistant.core import HomeAssistant, callback
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.setup import async_setup_component
|
||||
from homeassistant.util import yaml
|
||||
|
||||
from tests.common import async_mock_service
|
||||
|
||||
BUILTIN_BLUEPRINT_FOLDER = pathlib.Path(template.__file__).parent / "blueprints"
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def patch_blueprint(
|
||||
blueprint_path: str, data_path: str | PathLike[str]
|
||||
) -> Iterator[None]:
|
||||
"""Patch blueprint loading from a different source."""
|
||||
orig_load = DomainBlueprints._load_blueprint
|
||||
|
||||
@callback
|
||||
def mock_load_blueprint(self, path):
|
||||
if path != blueprint_path:
|
||||
pytest.fail(f"Unexpected blueprint {path}")
|
||||
return orig_load(self, path)
|
||||
|
||||
return Blueprint(
|
||||
yaml.load_yaml(data_path),
|
||||
expected_domain=self.domain,
|
||||
path=path,
|
||||
schema=BLUEPRINT_SCHEMA,
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.blueprint.models.DomainBlueprints._load_blueprint",
|
||||
mock_load_blueprint,
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def patch_invalid_blueprint() -> Iterator[None]:
|
||||
"""Patch blueprint returning an invalid one."""
|
||||
|
||||
@callback
|
||||
def mock_load_blueprint(self, path):
|
||||
return Blueprint(
|
||||
{
|
||||
"blueprint": {
|
||||
"domain": "template",
|
||||
"name": "Invalid template blueprint",
|
||||
},
|
||||
"binary_sensor": {},
|
||||
"sensor": {},
|
||||
},
|
||||
expected_domain=self.domain,
|
||||
path=path,
|
||||
schema=BLUEPRINT_SCHEMA,
|
||||
)
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.blueprint.models.DomainBlueprints._load_blueprint",
|
||||
mock_load_blueprint,
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
async def test_inverted_binary_sensor(
|
||||
hass: HomeAssistant, device_registry: dr.DeviceRegistry
|
||||
) -> None:
|
||||
"""Test inverted binary sensor blueprint."""
|
||||
hass.states.async_set("binary_sensor.foo", "on", {"friendly_name": "Foo"})
|
||||
hass.states.async_set("binary_sensor.bar", "off", {"friendly_name": "Bar"})
|
||||
|
||||
with patch_blueprint(
|
||||
"inverted_binary_sensor.yaml",
|
||||
BUILTIN_BLUEPRINT_FOLDER / "inverted_binary_sensor.yaml",
|
||||
):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
"template",
|
||||
{
|
||||
"template": [
|
||||
{
|
||||
"use_blueprint": {
|
||||
"path": "inverted_binary_sensor.yaml",
|
||||
"input": {"reference_entity": "binary_sensor.foo"},
|
||||
},
|
||||
"name": "Inverted foo",
|
||||
},
|
||||
{
|
||||
"use_blueprint": {
|
||||
"path": "inverted_binary_sensor.yaml",
|
||||
"input": {"reference_entity": "binary_sensor.bar"},
|
||||
},
|
||||
"name": "Inverted bar",
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
hass.states.async_set("binary_sensor.foo", "off", {"friendly_name": "Foo"})
|
||||
hass.states.async_set("binary_sensor.bar", "on", {"friendly_name": "Bar"})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get("binary_sensor.foo").state == "off"
|
||||
assert hass.states.get("binary_sensor.bar").state == "on"
|
||||
|
||||
inverted_foo = hass.states.get("binary_sensor.inverted_foo")
|
||||
assert inverted_foo
|
||||
assert inverted_foo.state == "on"
|
||||
|
||||
inverted_bar = hass.states.get("binary_sensor.inverted_bar")
|
||||
assert inverted_bar
|
||||
assert inverted_bar.state == "off"
|
||||
|
||||
foo_template = template.helpers.blueprint_in_template(hass, "binary_sensor.foo")
|
||||
inverted_foo_template = template.helpers.blueprint_in_template(
|
||||
hass, "binary_sensor.inverted_foo"
|
||||
)
|
||||
assert foo_template is None
|
||||
assert inverted_foo_template == "inverted_binary_sensor.yaml"
|
||||
|
||||
inverted_binary_sensor_blueprint_entity_ids = (
|
||||
template.helpers.templates_with_blueprint(hass, "inverted_binary_sensor.yaml")
|
||||
)
|
||||
assert len(inverted_binary_sensor_blueprint_entity_ids) == 2
|
||||
|
||||
assert len(template.helpers.templates_with_blueprint(hass, "dummy.yaml")) == 0
|
||||
|
||||
with pytest.raises(BlueprintInUse):
|
||||
await template.async_get_blueprints(hass).async_remove_blueprint(
|
||||
"inverted_binary_sensor.yaml"
|
||||
)
|
||||
|
||||
|
||||
async def test_domain_blueprint(hass: HomeAssistant) -> None:
|
||||
"""Test DomainBlueprint services."""
|
||||
reload_handler_calls = async_mock_service(hass, DOMAIN, SERVICE_RELOAD)
|
||||
mock_create_file = MagicMock()
|
||||
mock_create_file.return_value = True
|
||||
|
||||
with patch(
|
||||
"homeassistant.components.blueprint.models.DomainBlueprints._create_file",
|
||||
mock_create_file,
|
||||
):
|
||||
await template.async_get_blueprints(hass).async_add_blueprint(
|
||||
Blueprint(
|
||||
{
|
||||
"blueprint": {
|
||||
"domain": DOMAIN,
|
||||
"name": "Test",
|
||||
},
|
||||
},
|
||||
expected_domain="template",
|
||||
path="xxx",
|
||||
schema=BLUEPRINT_SCHEMA,
|
||||
),
|
||||
"xxx",
|
||||
True,
|
||||
)
|
||||
assert len(reload_handler_calls) == 1
|
||||
|
||||
|
||||
async def test_invalid_blueprint(
|
||||
hass: HomeAssistant, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""Test an invalid blueprint definition."""
|
||||
|
||||
with patch_invalid_blueprint():
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
"template",
|
||||
{
|
||||
"template": [
|
||||
{
|
||||
"use_blueprint": {
|
||||
"path": "invalid.yaml",
|
||||
},
|
||||
"name": "Invalid blueprint instance",
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
assert "more than one platform defined per blueprint" in caplog.text
|
||||
assert await template.async_get_blueprints(hass).async_get_blueprints() == {}
|
||||
|
||||
|
||||
async def test_no_blueprint(hass: HomeAssistant) -> None:
|
||||
"""Test templates without blueprints."""
|
||||
with patch_blueprint(
|
||||
"inverted_binary_sensor.yaml",
|
||||
BUILTIN_BLUEPRINT_FOLDER / "inverted_binary_sensor.yaml",
|
||||
):
|
||||
assert await async_setup_component(
|
||||
hass,
|
||||
"template",
|
||||
{
|
||||
"template": [
|
||||
{"binary_sensor": {"name": "test entity", "state": "off"}},
|
||||
{
|
||||
"use_blueprint": {
|
||||
"path": "inverted_binary_sensor.yaml",
|
||||
"input": {"reference_entity": "binary_sensor.foo"},
|
||||
},
|
||||
"name": "inverted entity",
|
||||
},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
hass.states.async_set("binary_sensor.foo", "off", {"friendly_name": "Foo"})
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert (
|
||||
len(
|
||||
template.helpers.templates_with_blueprint(
|
||||
hass, "inverted_binary_sensor.yaml"
|
||||
)
|
||||
)
|
||||
== 1
|
||||
)
|
||||
assert (
|
||||
template.helpers.blueprint_in_template(hass, "binary_sensor.test_entity")
|
||||
is None
|
||||
)
|
Loading…
x
Reference in New Issue
Block a user