Compare commits

...
Sign in to create a new pull request.

7 commits

Author SHA1 Message Date
Erik
4c0fcbb824 Don't fail on camera 2024-11-08 13:26:04 +01:00
Erik
d8716b6b72 Don't fail on unifiprotect 2024-11-08 13:25:23 +01:00
Erik
e88092dc52 Extend checks 2024-11-08 13:25:23 +01:00
Erik
6e4b5f31d4 Extend checks 2024-11-08 13:25:23 +01:00
Erik
df6763328f Remove explicit templating of knx service data 2024-11-08 13:25:23 +01:00
Erik
51e409c1c7 Check all services 2024-11-08 13:25:23 +01:00
Erik
8c9bf7e41c Fail on templated services 2024-11-08 13:23:12 +01:00
3 changed files with 60 additions and 1 deletions

View file

@ -13,7 +13,7 @@ from xknx.telegram import Telegram
from xknx.telegram.address import parse_device_group_address
from xknx.telegram.apci import GroupValueRead, GroupValueResponse, GroupValueWrite
from homeassistant.const import CONF_TYPE, SERVICE_RELOAD
from homeassistant.const import CONF_TYPE, CONF_VALUE_TEMPLATE, SERVICE_RELOAD
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
import homeassistant.helpers.config_validation as cv
@ -143,6 +143,7 @@ SERVICE_KNX_EXPOSURE_REGISTER_SCHEMA = vol.Any(
ExposeSchema.EXPOSE_SENSOR_SCHEMA.extend(
{
vol.Optional(SERVICE_KNX_ATTR_REMOVE, default=False): cv.boolean,
vol.Optional(CONF_VALUE_TEMPLATE): cv.string,
}
),
vol.Schema(

View file

@ -2610,6 +2610,10 @@ class ServiceRegistry:
This method must be run in the event loop.
"""
# pylint: disable-next=import-outside-toplevel
from .helpers import config_validation as cv
cv.raise_on_templated_service(domain, service, schema)
domain = domain.lower()
service = service.lower()
service_obj = Service(

View file

@ -1380,6 +1380,60 @@ def _make_entity_service_schema(schema: dict, extra: int) -> VolSchemaType:
BASE_ENTITY_SCHEMA = _make_entity_service_schema({}, vol.PREVENT_EXTRA)
def raise_on_templated_service(
domain: str, _service: str, schema: VolDictType | VolSchemaType | None
) -> None:
"""Raise if service schema explicitly allows templates."""
return _raise_on_templated_service(domain, _service, schema, [])
def _raise_on_templated_service(
domain: str,
_service: str,
schema: Any,
_path: list[str | int],
) -> None:
"""Raise if service schema explicitly allows templates."""
if not schema:
return
if isinstance(schema, dict):
for key, val in schema.items():
_raise_on_templated_service(domain, _service, val, [*_path, key])
return
if isinstance(schema, list):
for pos, val in enumerate(schema):
_raise_on_templated_service(domain, _service, val, [*_path, pos])
return
if isinstance(schema, vol.All):
for pos, val in enumerate(schema.validators):
_raise_on_templated_service(domain, _service, val, [*_path, "All", pos])
if isinstance(schema, vol.Any):
for pos, val in enumerate(schema.validators):
_raise_on_templated_service(domain, _service, val, [*_path, "Any", pos])
if isinstance(schema, (vol.Schema)):
_raise_on_templated_service(domain, _service, schema.schema, _path)
if _path[-5:] == ["All", 0, "entity_id", "Any", 1]:
return
if _path[-7:] == ["All", 0, "entity_id", "Any", 2, "All", 1]:
return
if _path[-10:] == ["All", 0, "device_id", "Any", 1, "All", 1, 0, "Any", 0]:
return
if _path[-10:] == ["All", 0, "area_id", "Any", 1, "All", 1, 0, "Any", 0]:
return
if _path[-10:] == ["All", 0, "floor_id", "Any", 1, "All", 1, 0, "Any", 0]:
return
if _path[-10:] == ["All", 0, "label_id", "Any", 1, "All", 1, 0, "Any", 0]:
return
if domain == "camera" and _service in ("record", "snapshot"):
return
if domain == "unifiprotect" and _service == "set_chime_paired_doorbells":
return
if schema in (dynamic_template, template, template_complex):
raise ValueError(
f"Template in service data is not allowed! {domain}.{_service}:{_path}"
)
def make_entity_service_schema(
schema: dict | None, *, extra: int = vol.PREVENT_EXTRA
) -> VolSchemaType: