Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-09 07:01:41 +01:00
Code Activity

Use probatio directly in KNX config store (#176855)

Browse Source 
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Matthias Alphart
2026-08-24 21:17:54 +02:00
committed by GitHub
co-authored by Claude Opus 5
parent 09afc39ac4
commit dcbd43bfbc
7 changed files with 428 additions and 340 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/knx/storage/entity_store_schema.py
+274 -228
View File
File diff suppressed because it is too large Load Diff
homeassistant/components/knx/storage/entity_store_validation.py
+21 -24
View File
@@ -1,18 +1,21 @@
"""KNX entity store validation."""
from typing import Literal, TypedDict
from collections.abc import Callable
from typing import Any, Literal, TypedDict
import voluptuous as vol
from homeassistant.helpers.typing import VolSchemaType
import probatio
from .entity_store_schema import ENTITY_STORE_DATA_SCHEMA
class _ErrorDescription(TypedDict):
path: list[str] | None
path: list[str]
message: str
code: str | None
translation_key: str | None
placeholders: dict[str, Any]
context: dict[str, Any]
secret: bool
class EntityStoreValidationError(TypedDict):
@@ -30,37 +33,31 @@ class EntityStoreValidationSuccess(TypedDict):
entity_id: str | None
def parse_invalid(exc: vol.Invalid) -> _ErrorDescription:
"""Parse a vol.Invalid exception."""
return _ErrorDescription(
path=[str(path) for path in exc.path], # exc.path: str | vol.Required
message=exc.msg,
code=type(exc).__name__,
)
def parse_invalid(exc: probatio.Invalid) -> _ErrorDescription:
"""Parse a probatio.Invalid exception."""
description = exc.as_dict()
# path items are str or probatio.Marker; the frontend matches them against config keys
description["path"] = [str(path) for path in description["path"]]
return description # type: ignore[return-value]
def validate_config_store_data(schema: VolSchemaType, entity_data: dict) -> dict:
def validate_config_store_data(
schema: Callable[[dict], dict], entity_data: dict
) -> dict:
"""Validate data for config store.
Return validated data or raise EntityStoreValidationException.
"""
try:
# return so defaults are applied
return schema(entity_data) # type: ignore[no-any-return]
except vol.MultipleInvalid as exc:
return schema(entity_data)
except probatio.Invalid as exc:
errors = exc.errors if isinstance(exc, probatio.MultipleInvalid) else [exc]
raise EntityStoreValidationException(
validation_error={
"success": False,
"error_base": str(exc),
"errors": [parse_invalid(invalid) for invalid in exc.errors],
}
) from exc
except vol.Invalid as exc:
raise EntityStoreValidationException(
validation_error={
"success": False,
"error_base": str(exc),
"errors": [parse_invalid(exc)],
"errors": [parse_invalid(invalid) for invalid in errors],
}
) from exc
homeassistant/components/knx/storage/expose_controller.py
+18 -16
View File
@@ -2,7 +2,7 @@
from typing import Any, NotRequired, TypedDict
import voluptuous as vol
import probatio
from xknx import XKNX
from xknx.dpt import DPTBase
from xknx.telegram.address import parse_device_group_address
@@ -52,7 +52,7 @@ def validate_expose_template_no_coerce(value: str) -> str:
"""Validate an expose template without coercing to Template."""
temp = cv.template(value) # validate template
if temp.is_static:
raise vol.Invalid(
raise probatio.Invalid(
"Static templates are not supported."
" Template should start with '{{'"
" and end with '}}'"
@@ -60,34 +60,36 @@ def validate_expose_template_no_coerce(value: str) -> str:
return value # return original string for storage and later template creation
EXPOSE_OPTION_SCHEMA = vol.Schema(
EXPOSE_OPTION_SCHEMA = probatio.Schema(
{
vol.Required("ga"): GASelector(
probatio.Required("ga"): GASelector(
state=False,
passive=False,
write_required=True,
dpt=["numeric", "enum", "complex", "string"],
),
vol.Optional("attribute"): str,
vol.Optional("default"): object,
vol.Optional("cooldown"): cv.positive_float, # frontend renders to duration
vol.Optional("periodic_send"): cv.positive_float,
vol.Optional("respond_to_read"): bool,
vol.Optional("value_template"): validate_expose_template_no_coerce,
probatio.Optional("attribute"): str,
probatio.Optional("default"): object,
probatio.Optional(
"cooldown"
): cv.positive_float, # frontend renders to duration
probatio.Optional("periodic_send"): cv.positive_float,
probatio.Optional("respond_to_read"): bool,
probatio.Optional("value_template"): validate_expose_template_no_coerce,
}
)
EXPOSE_CONFIG_SCHEMA = vol.Schema(
EXPOSE_CONFIG_SCHEMA = probatio.Schema(
{
vol.Required("entity_id"): selector.EntitySelector(),
vol.Required("data"): vol.Schema(
probatio.Required("entity_id"): selector.EntitySelector(),
probatio.Required("data"): probatio.Schema(
{
vol.Required("options"): [EXPOSE_OPTION_SCHEMA],
vol.Optional("notes"): str,
probatio.Required("options"): [EXPOSE_OPTION_SCHEMA],
probatio.Optional("notes"): str,
}
),
},
extra=vol.REMOVE_EXTRA,
extra=probatio.REMOVE_EXTRA,
)
homeassistant/components/knx/storage/knx_selector.py
+66 -55
View File
@@ -4,7 +4,7 @@ from collections.abc import Iterable
from enum import Enum
from typing import Any, override
import voluptuous as vol
import probatio
from homeassistant.const import CONF_PAYLOAD
@@ -15,10 +15,10 @@ from .const import CONF_DPT, CONF_GA_PASSIVE, CONF_GA_STATE, CONF_GA_WRITE
from .util import dpt_string_to_dict
class AllSerializeFirst(vol.All):
class AllSerializeFirst(probatio.All):
"""Use the first validated value for serialization.
This is a version of vol.All with custom error handling to
This is a version of probatio.All with custom error handling to
show proper invalid markers for sub-schema items in the UI.
"""
@@ -26,7 +26,7 @@ class AllSerializeFirst(vol.All):
class KNXSelectorBase:
"""Base class for KNX selectors supporting optional nested schemas."""
schema: vol.Schema | vol.Any | vol.All | GroupSelectSchema
schema: probatio.Schema | probatio.Any | probatio.All | GroupSelectSchema
selector_type: str
# mark if self.schema should be serialized to `schema` key
serialize_subschema: bool = False
@@ -49,7 +49,7 @@ class KNXSectionFlat(KNXSelectorBase):
"""Generate a schema-neutral section with title and description."""
selector_type = "knx_section_flat"
schema = vol.Schema(None)
schema = probatio.Schema(None)
def __init__(
self,
@@ -75,12 +75,12 @@ class KNXSection(KNXSelectorBase):
def __init__(
self,
schema: dict[str | vol.Marker, vol.Schemable],
schema: dict[str | probatio.Marker, probatio.Schemable],
collapsible: bool = True,
) -> None:
"""Initialize the section."""
self.collapsible = collapsible
self.schema = vol.Schema(schema)
self.schema = probatio.Schema(schema)
@override
def serialize(self) -> dict[str, Any]:
@@ -97,10 +97,10 @@ class GroupSelectOption(KNXSelectorBase):
selector_type = "knx_group_select_option"
serialize_subschema: bool = True
def __init__(self, schema: vol.Schemable, translation_key: str) -> None:
def __init__(self, schema: probatio.Schemable, translation_key: str) -> None:
"""Initialize the group select option schema."""
self.translation_key = translation_key
self.schema = vol.Schema(schema)
self.schema = probatio.Schema(schema)
@override
def serialize(self) -> dict[str, Any]:
@@ -111,35 +111,44 @@ class GroupSelectOption(KNXSelectorBase):
}
class GroupSelectSchema:
"""Use the first validated value, like ``vol.Any``.
def _has_extra_keys_error(exc: probatio.Invalid) -> bool:
"""Check if any of the errors is about extra keys."""
errors = exc.errors if isinstance(exc, probatio.MultipleInvalid) else [exc]
return any(isinstance(error, probatio.ExtraKeysInvalid) for error in errors)
A standalone validator rather than a ``vol.Any`` subclass, so it does not
class GroupSelectSchema:
"""Use the first validated value, like ``probatio.Any``.
A standalone validator rather than a ``probatio.Any`` subclass, so it does not
reach into validation-engine internals. On total failure it raises the most
useful branch error (the first that is not an unknown-key error, else the
first) so the UI marks a real problem instead of an extra key.
"""
def __init__(self, *options: vol.Schemable, msg: str | None = None) -> None:
def __init__(self, *options: probatio.Schemable, msg: str | None = None) -> None:
"""Store the options to try in order."""
self.validators = options
self.msg = msg
self._compiled = [vol.Schema(option) for option in options]
self._compiled = [probatio.Schema(option) for option in options]
def __call__(self, data: Any) -> Any:
"""Return the first option that validates, else raise the best error."""
errors: list[vol.Invalid] = []
errors: list[probatio.Invalid] = []
for option in self._compiled:
try:
return option(data)
except vol.Invalid as err:
except probatio.Invalid as err:
errors.append(err)
if errors:
# an option is only reported when it matches the given keys;
# `code` of a MultipleInvalid is just its first errors code, so
# every error of an option is checked for being about extra keys
raise next(
(err for err in errors if err.code != "extra_keys_not_allowed"),
(err for err in errors if not _has_extra_keys_error(err)),
errors[0],
)
raise vol.AnyInvalid(self.msg or "no valid value found")
raise probatio.AnyInvalid(self.msg or "no valid value found")
class GroupSelect(KNXSelectorBase):
@@ -228,73 +237,73 @@ class GASelector(KNXSelectorBase):
"options": options,
}
def build_schema(self) -> vol.Schema:
def build_schema(self) -> probatio.Schema:
"""Create the schema based on configuration."""
schema: dict[vol.Marker, Any] = {} # will be modified in-place
schema: dict[probatio.Marker, Any] = {} # will be modified in-place
self._add_group_addresses(schema)
self._add_passive(schema)
self._add_dpt(schema)
return vol.Schema(
vol.All(
return probatio.Schema(
probatio.All(
schema,
vol.Schema( # one group address shall be included
vol.Any(
{vol.Required(CONF_GA_WRITE): vol.IsTrue()},
{vol.Required(CONF_GA_STATE): vol.IsTrue()},
{vol.Required(CONF_GA_PASSIVE): vol.IsTrue()},
probatio.Schema( # one group address shall be included
probatio.Any(
{probatio.Required(CONF_GA_WRITE): probatio.IsTrue()},
{probatio.Required(CONF_GA_STATE): probatio.IsTrue()},
{probatio.Required(CONF_GA_PASSIVE): probatio.IsTrue()},
msg="At least one group address must be set",
),
extra=vol.ALLOW_EXTRA,
extra=probatio.ALLOW_EXTRA,
),
)
)
def _add_group_addresses(self, schema: dict[vol.Marker, Any]) -> None:
def _add_group_addresses(self, schema: dict[probatio.Marker, Any]) -> None:
"""Add basic group address items to the schema."""
def add_ga_item(key: str, allowed: bool, required: bool) -> None:
"""Add a group address item validator to the schema."""
if not allowed:
schema[vol.Remove(key)] = object
schema[probatio.Remove(key)] = object
return
if required:
schema[vol.Required(key)] = ga_validator
schema[probatio.Required(key)] = ga_validator
else:
schema[vol.Optional(key, default=None)] = maybe_ga_validator
schema[probatio.Optional(key, default=None)] = maybe_ga_validator
add_ga_item(CONF_GA_WRITE, self.write, self.write_required)
add_ga_item(CONF_GA_STATE, self.state, self.state_required)
def _add_passive(self, schema: dict[vol.Marker, Any]) -> None:
def _add_passive(self, schema: dict[probatio.Marker, Any]) -> None:
"""Add passive group addresses validator to the schema."""
if self.passive:
schema[vol.Optional(CONF_GA_PASSIVE, default=list)] = vol.Any(
schema[probatio.Optional(CONF_GA_PASSIVE, default=list)] = probatio.Any(
[ga_validator],
vol.All( # Coerce `None` to an empty list if passive is allowed
vol.IsFalse(), vol.SetTo(list)
probatio.All( # Coerce `None` to an empty list if passive is allowed
probatio.IsFalse(), probatio.SetTo(list)
),
)
else:
schema[vol.Remove(CONF_GA_PASSIVE)] = object
schema[probatio.Remove(CONF_GA_PASSIVE)] = object
def _add_dpt(self, schema: dict[vol.Marker, Any]) -> None:
def _add_dpt(self, schema: dict[probatio.Marker, Any]) -> None:
"""Add DPT validator to the schema."""
if self.dpt is not None:
if isinstance(self.dpt, list):
marker = vol.Required if self.dpt_required else vol.Optional
schema[marker(CONF_DPT)] = vol.In(get_supported_dpts())
marker = probatio.Required if self.dpt_required else probatio.Optional
schema[marker(CONF_DPT)] = probatio.In(get_supported_dpts())
else:
schema[vol.Required(CONF_DPT)] = vol.In(
schema[probatio.Required(CONF_DPT)] = probatio.In(
{item.value for item in self.dpt}
)
else:
schema[vol.Remove(CONF_DPT)] = object
schema[probatio.Remove(CONF_DPT)] = object
class SyncStateSelector(KNXSelectorBase):
"""Selector for knx sync state validation."""
schema = vol.Schema(sync_state_validator)
schema = probatio.Schema(sync_state_validator)
selector_type = "knx_sync_state"
def __init__(self, allow_false: bool = False) -> None:
@@ -313,7 +322,7 @@ class SyncStateSelector(KNXSelectorBase):
def __call__(self, data: Any) -> Any:
"""Validate the passed data."""
if not self.allow_false and not data:
raise vol.Invalid(f"Sync state cannot be {data}")
raise probatio.Invalid(f"Sync state cannot be {data}")
return self.schema(data)
@@ -323,13 +332,15 @@ class KnxPayloadSelector(KNXSelectorBase):
Raw payloads are stored as hex strings.
"""
schema = vol.Any(
schema = probatio.Any(
{
vol.Required(CONF_VALUE): object,
probatio.Required(CONF_VALUE): object,
},
{
vol.Required(CONF_PAYLOAD): str,
vol.Required(CONF_PAYLOAD_LENGTH): vol.All(int, vol.Range(min=0, max=14)),
probatio.Required(CONF_PAYLOAD): str,
probatio.Required(CONF_PAYLOAD_LENGTH): probatio.All(
int, probatio.Range(min=0, max=14)
),
},
)
selector_type = "knx_payload"
@@ -356,21 +367,21 @@ class KnxPayloadSelector(KNXSelectorBase):
try:
int_payload = int(payload, 16)
except ValueError as ex:
raise vol.Invalid(f"Invalid payload format: {payload}") from ex
raise probatio.Invalid(f"Invalid payload format: {payload}") from ex
validated[CONF_PAYLOAD] = hex(int_payload) # prepends "0x" if not present
if int_payload < 0:
raise vol.Invalid(f"Payload cannot be negative: {payload}")
raise probatio.Invalid(f"Payload cannot be negative: {payload}")
if payload_length == 0:
# DPT 1,2,3 is marked length 0, has 6 bit size
if int_payload > 63:
raise vol.Invalid(
raise probatio.Invalid(
f"Payload exceeds DPT 1,2,3 limit of 0x3f (63): {payload}"
)
else:
max_payload = (1 << (payload_length * 8)) - 1
if int_payload > max_payload:
raise vol.Invalid(
raise probatio.Invalid(
f"Payload {payload} exceeds possible maximum for "
f"length {payload_length}: {hex(max_payload)}"
)
@@ -392,7 +403,7 @@ class KnxSelectOptionsSelector(KNXSelectorBase):
"""Initialize the options selector."""
self.ga_path = ga_path
self._payload_selector = KnxPayloadSelector(ga_path=ga_path)
self.schema = vol.Schema([self._validate_option])
self.schema = probatio.Schema([self._validate_option])
@override
def serialize(self) -> dict[str, Any]:
@@ -409,10 +420,10 @@ class KnxSelectOptionsSelector(KNXSelectorBase):
sub-validator.
"""
if not isinstance(data, dict):
raise vol.Invalid("Each option must be a dictionary")
raise probatio.Invalid("Each option must be a dictionary")
option = data.get(SelectConf.OPTION)
if not isinstance(option, str) or not option:
raise vol.Invalid("Option name is required", path=[SelectConf.OPTION])
raise probatio.Invalid("Option name is required", path=[SelectConf.OPTION])
payload = {
key: value for key, value in data.items() if key != SelectConf.OPTION
}
homeassistant/components/knx/storage/time_server.py
+5 -5
View File
@@ -2,7 +2,7 @@
from typing import Any, TypedDict
import voluptuous as vol
import probatio
from xknx import XKNX
from ..expose import KnxExposeTime, create_time_server_exposures
@@ -18,15 +18,15 @@ class KNXTimeServerStoreModel(TypedDict, total=False):
datetime: dict[str, Any] | None
TIME_SERVER_CONFIG_SCHEMA = vol.Schema(
TIME_SERVER_CONFIG_SCHEMA = probatio.Schema(
{
vol.Optional("time"): GASelector(
probatio.Optional("time"): GASelector(
state=False, passive=False, valid_dpt="10.001"
),
vol.Optional("date"): GASelector(
probatio.Optional("date"): GASelector(
state=False, passive=False, valid_dpt="11.001"
),
vol.Optional("datetime"): GASelector(
probatio.Optional("datetime"): GASelector(
state=False, passive=False, valid_dpt="19.001"
),
}
tests/components/knx/test_config_store.py
+32 -3
View File
@@ -419,7 +419,7 @@ async def test_validate_entity(
assert res["result"]["success"] is False
assert res["result"]["errors"][0]["path"] == ["data", "knx", "ga_switch", "write"]
assert res["result"]["errors"][0]["message"] == "required key not provided"
assert res["result"]["errors"][0]["code"] == "RequiredFieldInvalid"
assert res["result"]["errors"][0]["code"] == "required"
assert res["result"]["error_base"].startswith("required key not provided")
# invalid group_select data
@@ -451,9 +451,38 @@ async def test_validate_entity(
"ga_blue_brightness",
]
assert res["result"]["errors"][0]["message"] == "required key not provided"
assert res["result"]["errors"][0]["code"] == "RequiredFieldInvalid"
assert res["result"]["errors"][0]["code"] == "required"
assert res["result"]["error_base"].startswith("required key not provided")
# partially configured group_select option
await client.send_json_auto_id(
{
"type": "knx/validate_entity",
"platform": Platform.LIGHT,
"data": {
"entity": {"name": "test_name"},
"knx": {
"color": {
"ga_hue": {"write": "1/2/3"},
# ga_saturation is missing - which is required
}
},
},
}
)
res = await client.receive_json()
assert res["success"], res
assert res["result"]["success"] is False
# the error of the option the user started configuring shall be reported,
# not a "required key" error of one of the other options
assert res["result"]["errors"][0]["path"] == [
"data",
"knx",
"color",
"ga_saturation",
]
assert res["result"]["errors"][0]["code"] == "required"
########
# EXPOSE
@@ -481,7 +510,7 @@ async def test_update_expose_error(
assert res["result"]["success"] is False
assert res["result"]["errors"][0]["path"] == ["data", "options", "0", "ga", "write"]
assert res["result"]["errors"][0]["message"] == "required key not provided"
assert res["result"]["errors"][0]["code"] == "RequiredFieldInvalid"
assert res["result"]["errors"][0]["code"] == "required"
async def test_validate_expose(
tests/components/knx/test_knx_selectors.py
+12 -9
View File
@@ -2,9 +2,8 @@
from typing import Any
from probatio import to_field_list
import probatio
import pytest
import voluptuous as vol
from homeassistant.components.knx.const import ColorTempModes
from homeassistant.components.knx.storage.knx_selector import (
@@ -177,7 +176,7 @@ def test_ga_selector_invalid(
) -> None:
"""Test GASelector."""
selector = GASelector(**selector_config)
with pytest.raises(vol.Invalid, match=error_str):
with pytest.raises(probatio.Invalid, match=error_str):
selector(data)
@@ -186,10 +185,10 @@ def test_sync_state_selector() -> None:
selector = SyncStateSelector()
assert selector("expire 50") == "expire 50"
with pytest.raises(vol.Invalid):
with pytest.raises(probatio.Invalid):
selector("invalid")
with pytest.raises(vol.Invalid, match="Sync state cannot be False"):
with pytest.raises(probatio.Invalid, match="Sync state cannot be False"):
selector(False)
false_allowed = SyncStateSelector(allow_false=True)
@@ -265,7 +264,9 @@ def test_ga_selector_serialization(
("schema", "serialized"),
[
(
AllSerializeFirst(vol.Schema({"key": int}), vol.Schema({"ignored": str})),
AllSerializeFirst(
probatio.Schema({"key": int}), probatio.Schema({"ignored": str})
),
[{"name": "key", "required": False, "type": "integer"}],
),
(
@@ -324,10 +325,10 @@ def test_ga_selector_serialization(
},
),
( # in a dict schema `name` and `required` keys are added
vol.Schema(
probatio.Schema(
{
"section_test": KNXSectionFlat(),
vol.Optional("key"): selector.BooleanSelector(),
probatio.Optional("key"): selector.BooleanSelector(),
}
),
[
@@ -350,4 +351,6 @@ def test_ga_selector_serialization(
)
def test_serialization(schema: Any, serialized: dict[str, Any]) -> None:
"""Test serialization of the selector."""
assert to_field_list(schema, custom_serializer=knx_serializer) == serialized
assert (
probatio.to_field_list(schema, custom_serializer=knx_serializer) == serialized
)