From dcbd43bfbcb606a7d32035308b1be0d1a77d6592 Mon Sep 17 00:00:00 2001 From: Matthias Alphart Date: Mon, 24 Aug 2026 21:17:54 +0200 Subject: [PATCH] Use probatio directly in KNX config store (#176855) Co-authored-by: Claude Opus 5 --- .../knx/storage/entity_store_schema.py | 502 ++++++++++-------- .../knx/storage/entity_store_validation.py | 45 +- .../knx/storage/expose_controller.py | 34 +- .../components/knx/storage/knx_selector.py | 121 +++-- .../components/knx/storage/time_server.py | 10 +- tests/components/knx/test_config_store.py | 35 +- tests/components/knx/test_knx_selectors.py | 21 +- 7 files changed, 428 insertions(+), 340 deletions(-) diff --git a/homeassistant/components/knx/storage/entity_store_schema.py b/homeassistant/components/knx/storage/entity_store_schema.py index 1d26bd0480e8..23abcc612e9b 100644 --- a/homeassistant/components/knx/storage/entity_store_schema.py +++ b/homeassistant/components/knx/storage/entity_store_schema.py @@ -3,7 +3,7 @@ from collections.abc import Hashable from enum import StrEnum, unique -import voluptuous as vol +import probatio from xknx.dpt import DPTBase, DPTBinary, DPTNumeric from xknx.exceptions import ConversionError @@ -31,9 +31,9 @@ from homeassistant.const import ( CONF_UNIT_OF_MEASUREMENT, Platform, ) -from homeassistant.helpers import config_validation as cv, selector +from homeassistant.helpers import selector from homeassistant.helpers.entity import ENTITY_CATEGORIES_SCHEMA -from homeassistant.helpers.typing import VolDictType, VolSchemaType +from homeassistant.helpers.typing import VolDictType from ..const import ( CONF_CONTEXT_TIMEOUT, @@ -142,51 +142,51 @@ from .knx_selector import ( SyncStateSelector, ) -BASE_ENTITY_SCHEMA = vol.All( +BASE_ENTITY_SCHEMA = probatio.All( { - vol.Optional(CONF_NAME, default=None): vol.Maybe(str), - vol.Optional(CONF_DEVICE_INFO, default=None): vol.Maybe(str), - vol.Optional(CONF_ENTITY_CATEGORY, default=None): vol.Any( - ENTITY_CATEGORIES_SCHEMA, vol.SetTo(None) + probatio.Optional(CONF_NAME, default=None): probatio.Maybe(str), + probatio.Optional(CONF_DEVICE_INFO, default=None): probatio.Maybe(str), + probatio.Optional(CONF_ENTITY_CATEGORY, default=None): probatio.Any( + ENTITY_CATEGORIES_SCHEMA, probatio.SetTo(None) ), }, - vol.Any( - vol.Schema( + probatio.Any( + probatio.Schema( { - vol.Required(CONF_NAME): vol.All(str, vol.IsTrue()), + probatio.Required(CONF_NAME): probatio.All(str, probatio.IsTrue()), }, - extra=vol.ALLOW_EXTRA, + extra=probatio.ALLOW_EXTRA, ), - vol.Schema( + probatio.Schema( { - vol.Required(CONF_DEVICE_INFO): str, + probatio.Required(CONF_DEVICE_INFO): str, }, - extra=vol.ALLOW_EXTRA, + extra=probatio.ALLOW_EXTRA, ), msg="One of `Device` or `Name` is required", ), ) -BINARY_SENSOR_KNX_SCHEMA = vol.Schema( +BINARY_SENSOR_KNX_SCHEMA = probatio.Schema( { - vol.Required(CONF_GA_SENSOR): GASelector( + probatio.Required(CONF_GA_SENSOR): GASelector( write=False, state_required=True, valid_dpt="1" ), - vol.Optional(CONF_INVERT): selector.BooleanSelector(), + probatio.Optional(CONF_INVERT): selector.BooleanSelector(), "section_advanced_options": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_IGNORE_INTERNAL_STATE): selector.BooleanSelector(), - vol.Optional(CONF_CONTEXT_TIMEOUT): selector.NumberSelector( + probatio.Optional(CONF_IGNORE_INTERNAL_STATE): selector.BooleanSelector(), + probatio.Optional(CONF_CONTEXT_TIMEOUT): selector.NumberSelector( selector.NumberSelectorConfig( min=0, max=10, step=0.1, unit_of_measurement="s" ) ), - vol.Optional(CONF_RESET_AFTER): selector.NumberSelector( + probatio.Optional(CONF_RESET_AFTER): selector.NumberSelector( selector.NumberSelectorConfig( min=0, max=600, step=0.1, unit_of_measurement="s" ) ), - vol.Required(CONF_SYNC_STATE, default=True): SyncStateSelector( + probatio.Required(CONF_SYNC_STATE, default=True): SyncStateSelector( allow_false=True ), }, @@ -205,7 +205,7 @@ def _button_data_sub_validator(config: dict) -> dict: try: transcoder.to_knx(config[CONF_DATA][CONF_VALUE]) except ConversionError as ex: - raise vol.Invalid( + raise probatio.Invalid( f"Value invalid for DPT {transcoder.dpt_number_str()}", path=([CONF_DATA]), ) from ex @@ -214,7 +214,7 @@ def _button_data_sub_validator(config: dict) -> dict: if length != transcoder.payload_length or ( length != 0 and transcoder.payload_type is DPTBinary ): - raise vol.Invalid( + raise probatio.Invalid( f"Payload length invalid for DPT {transcoder.dpt_number_str()}", path=([CONF_DATA]), ) @@ -222,78 +222,78 @@ def _button_data_sub_validator(config: dict) -> dict: # without DPT only raw allowed -> payload + payload_length (checked by KnxPayloadSelector) if CONF_PAYLOAD_LENGTH in config[CONF_DATA]: return config - raise vol.Invalid("Invalid configuration for button entity") + raise probatio.Invalid("Invalid configuration for button entity") BUTTON_KNX_SCHEMA = AllSerializeFirst( - vol.Schema( + probatio.Schema( { - vol.Required(CONF_GA_SEND): GASelector( + probatio.Required(CONF_GA_SEND): GASelector( state=False, write_required=True, passive=False, dpt=["numeric", "enum", "complex", "string"], dpt_required=False, # for raw payload support ), - vol.Required(CONF_DATA): KnxPayloadSelector(ga_path=CONF_GA_SEND), + probatio.Required(CONF_DATA): KnxPayloadSelector(ga_path=CONF_GA_SEND), }, ), _button_data_sub_validator, ) COVER_KNX_SCHEMA = AllSerializeFirst( - vol.Schema( + probatio.Schema( { - vol.Optional(CONF_GA_UP_DOWN): GASelector(state=False, valid_dpt="1"), - vol.Optional(CoverConf.INVERT_UPDOWN): selector.BooleanSelector(), - vol.Optional(CONF_GA_STOP): GASelector(state=False, valid_dpt="1"), - vol.Optional(CONF_GA_STEP): GASelector(state=False, valid_dpt="1"), + probatio.Optional(CONF_GA_UP_DOWN): GASelector(state=False, valid_dpt="1"), + probatio.Optional(CoverConf.INVERT_UPDOWN): selector.BooleanSelector(), + probatio.Optional(CONF_GA_STOP): GASelector(state=False, valid_dpt="1"), + probatio.Optional(CONF_GA_STEP): GASelector(state=False, valid_dpt="1"), "section_position_control": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_POSITION_SET): GASelector( + probatio.Optional(CONF_GA_POSITION_SET): GASelector( state=False, valid_dpt="5.001" ), - vol.Optional(CONF_GA_POSITION_STATE): GASelector( + probatio.Optional(CONF_GA_POSITION_STATE): GASelector( write=False, valid_dpt="5.001" ), - vol.Optional(CoverConf.INVERT_POSITION): selector.BooleanSelector(), + probatio.Optional(CoverConf.INVERT_POSITION): selector.BooleanSelector(), "section_tilt_control": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_ANGLE): GASelector(valid_dpt="5.001"), - vol.Optional(CoverConf.INVERT_ANGLE): selector.BooleanSelector(), + probatio.Optional(CONF_GA_ANGLE): GASelector(valid_dpt="5.001"), + probatio.Optional(CoverConf.INVERT_ANGLE): selector.BooleanSelector(), "section_travel_time": KNXSectionFlat(), - vol.Required( + probatio.Required( CoverConf.TRAVELLING_TIME_UP, default=25 ): selector.NumberSelector( selector.NumberSelectorConfig( min=0, max=1000, step=0.1, unit_of_measurement="s" ) ), - vol.Required( + probatio.Required( CoverConf.TRAVELLING_TIME_DOWN, default=25 ): selector.NumberSelector( selector.NumberSelectorConfig( min=0, max=1000, step=0.1, unit_of_measurement="s" ) ), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), }, - extra=vol.REMOVE_EXTRA, + extra=probatio.REMOVE_EXTRA, ), - vol.Any( - vol.Schema( + probatio.Any( + probatio.Schema( { - vol.Required(CONF_GA_UP_DOWN): GASelector( + probatio.Required(CONF_GA_UP_DOWN): GASelector( state=False, write_required=True ) }, - extra=vol.ALLOW_EXTRA, + extra=probatio.ALLOW_EXTRA, ), - vol.Schema( + probatio.Schema( { - vol.Required(CONF_GA_POSITION_SET): GASelector( + probatio.Required(CONF_GA_POSITION_SET): GASelector( state=False, write_required=True ) }, - extra=vol.ALLOW_EXTRA, + extra=probatio.ALLOW_EXTRA, ), msg=( "At least one of 'Open/Close control' or" @@ -302,35 +302,41 @@ COVER_KNX_SCHEMA = AllSerializeFirst( ), ) -DATE_KNX_SCHEMA = vol.Schema( +DATE_KNX_SCHEMA = probatio.Schema( { - vol.Required(CONF_GA_DATE): GASelector(write_required=True, valid_dpt="11.001"), - vol.Optional(CONF_RESPOND_TO_READ, default=False): selector.BooleanSelector(), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Required(CONF_GA_DATE): GASelector( + write_required=True, valid_dpt="11.001" + ), + probatio.Optional( + CONF_RESPOND_TO_READ, default=False + ): selector.BooleanSelector(), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), } ) -DATETIME_KNX_SCHEMA = vol.Schema( +DATETIME_KNX_SCHEMA = probatio.Schema( { - vol.Required(CONF_GA_DATETIME): GASelector( + probatio.Required(CONF_GA_DATETIME): GASelector( write_required=True, valid_dpt="19.001" ), - vol.Optional(CONF_RESPOND_TO_READ, default=False): selector.BooleanSelector(), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Optional( + CONF_RESPOND_TO_READ, default=False + ): selector.BooleanSelector(), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), } ) FAN_KNX_SCHEMA = AllSerializeFirst( - vol.Schema( + probatio.Schema( { - vol.Optional(CONF_GA_SWITCH): GASelector( + probatio.Optional(CONF_GA_SWITCH): GASelector( write_required=True, valid_dpt="1" ), - vol.Optional(CONF_SPEED): GroupSelect( + probatio.Optional(CONF_SPEED): GroupSelect( GroupSelectOption( translation_key="percentage_mode", schema={ - vol.Required(CONF_GA_SPEED): GASelector( + probatio.Required(CONF_GA_SPEED): GASelector( write_required=True, valid_dpt="5.001" ), }, @@ -338,10 +344,10 @@ FAN_KNX_SCHEMA = AllSerializeFirst( GroupSelectOption( translation_key="step_mode", schema={ - vol.Required(CONF_GA_STEP): GASelector( + probatio.Required(CONF_GA_STEP): GASelector( write_required=True, valid_dpt="5.010" ), - vol.Required( + probatio.Required( FanConf.MAX_STEP, default=3 ): selector.NumberSelector( selector.NumberSelectorConfig( @@ -355,20 +361,20 @@ FAN_KNX_SCHEMA = AllSerializeFirst( ), collapsible=False, ), - vol.Optional(CONF_GA_OSCILLATION): GASelector( + probatio.Optional(CONF_GA_OSCILLATION): GASelector( write_required=True, valid_dpt="1" ), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), } ), - vol.Any( - vol.Schema( - {vol.Required(CONF_GA_SWITCH): object}, - extra=vol.ALLOW_EXTRA, + probatio.Any( + probatio.Schema( + {probatio.Required(CONF_GA_SWITCH): object}, + extra=probatio.ALLOW_EXTRA, ), - vol.Schema( - {vol.Required(CONF_SPEED): object}, - extra=vol.ALLOW_EXTRA, + probatio.Schema( + {probatio.Required(CONF_SPEED): object}, + extra=probatio.ALLOW_EXTRA, ), msg=("At least one of 'Switch' or 'Fan speed' is required."), ), @@ -390,33 +396,37 @@ _hs_color_inclusion_msg = ( LIGHT_KNX_SCHEMA = AllSerializeFirst( - vol.Schema( + probatio.Schema( { - vol.Optional(CONF_GA_SWITCH): GASelector( + probatio.Optional(CONF_GA_SWITCH): GASelector( write_required=True, valid_dpt="1" ), - vol.Optional(CONF_GA_BRIGHTNESS): GASelector( + probatio.Optional(CONF_GA_BRIGHTNESS): GASelector( write_required=True, valid_dpt="5.001" ), "section_color_temp": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_COLOR_TEMP): GASelector( + probatio.Optional(CONF_GA_COLOR_TEMP): GASelector( write_required=True, dpt=ColorTempModes ), - vol.Required(CONF_COLOR_TEMP_MIN, default=2700): selector.NumberSelector( + probatio.Required( + CONF_COLOR_TEMP_MIN, default=2700 + ): selector.NumberSelector( selector.NumberSelectorConfig( min=1, max=10000, step=1, unit_of_measurement="K" ) ), - vol.Required(CONF_COLOR_TEMP_MAX, default=6000): selector.NumberSelector( + probatio.Required( + CONF_COLOR_TEMP_MAX, default=6000 + ): selector.NumberSelector( selector.NumberSelectorConfig( min=1, max=10000, step=1, unit_of_measurement="K" ) ), - vol.Optional(CONF_COLOR): GroupSelect( + probatio.Optional(CONF_COLOR): GroupSelect( GroupSelectOption( translation_key="single_address", schema={ - vol.Optional(CONF_GA_COLOR): GASelector( + probatio.Optional(CONF_GA_COLOR): GASelector( write_required=True, dpt=LightColorMode ) }, @@ -424,28 +434,28 @@ LIGHT_KNX_SCHEMA = AllSerializeFirst( GroupSelectOption( translation_key="individual_addresses", schema={ - vol.Optional(CONF_GA_RED_SWITCH): GASelector( + probatio.Optional(CONF_GA_RED_SWITCH): GASelector( write_required=False, valid_dpt="1" ), - vol.Required(CONF_GA_RED_BRIGHTNESS): GASelector( + probatio.Required(CONF_GA_RED_BRIGHTNESS): GASelector( write_required=True, valid_dpt="5.001" ), - vol.Optional(CONF_GA_GREEN_SWITCH): GASelector( + probatio.Optional(CONF_GA_GREEN_SWITCH): GASelector( write_required=False, valid_dpt="1" ), - vol.Required(CONF_GA_GREEN_BRIGHTNESS): GASelector( + probatio.Required(CONF_GA_GREEN_BRIGHTNESS): GASelector( write_required=True, valid_dpt="5.001" ), - vol.Optional(CONF_GA_BLUE_SWITCH): GASelector( + probatio.Optional(CONF_GA_BLUE_SWITCH): GASelector( write_required=False, valid_dpt="1" ), - vol.Required(CONF_GA_BLUE_BRIGHTNESS): GASelector( + probatio.Required(CONF_GA_BLUE_BRIGHTNESS): GASelector( write_required=True, valid_dpt="5.001" ), - vol.Optional(CONF_GA_WHITE_SWITCH): GASelector( + probatio.Optional(CONF_GA_WHITE_SWITCH): GASelector( write_required=False, valid_dpt="1" ), - vol.Optional(CONF_GA_WHITE_BRIGHTNESS): GASelector( + probatio.Optional(CONF_GA_WHITE_BRIGHTNESS): GASelector( write_required=True, valid_dpt="5.001" ), }, @@ -453,59 +463,65 @@ LIGHT_KNX_SCHEMA = AllSerializeFirst( GroupSelectOption( translation_key="hsv_addresses", schema={ - vol.Required(CONF_GA_HUE): GASelector( + probatio.Required(CONF_GA_HUE): GASelector( write_required=True, valid_dpt="5.003" ), - vol.Required(CONF_GA_SATURATION): GASelector( + probatio.Required(CONF_GA_SATURATION): GASelector( write_required=True, valid_dpt="5.001" ), }, ), ), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), } ), - vol.Any( - vol.Schema( - {vol.Required(CONF_GA_SWITCH): object}, - extra=vol.ALLOW_EXTRA, + probatio.Any( + probatio.Schema( + {probatio.Required(CONF_GA_SWITCH): object}, + extra=probatio.ALLOW_EXTRA, ), - vol.Schema( # brightness addresses are required in INDIVIDUAL_COLOR_SCHEMA - {vol.Required(CONF_COLOR): {vol.Required(CONF_GA_RED_BRIGHTNESS): object}}, - extra=vol.ALLOW_EXTRA, + probatio.Schema( # brightness addresses are required in INDIVIDUAL_COLOR_SCHEMA + { + probatio.Required(CONF_COLOR): { + probatio.Required(CONF_GA_RED_BRIGHTNESS): object + } + }, + extra=probatio.ALLOW_EXTRA, ), msg="either 'address' or 'individual_colors' is required", ), - vol.Any( - vol.Schema( # 'brightness' is non-optional for hs-color + probatio.Any( + probatio.Schema( # 'brightness' is non-optional for hs-color { - vol.Required(CONF_GA_BRIGHTNESS, msg=_hs_color_inclusion_msg): object, - vol.Required(CONF_COLOR): { - vol.Required(CONF_GA_HUE, msg=_hs_color_inclusion_msg): object, - vol.Required( + probatio.Required( + CONF_GA_BRIGHTNESS, msg=_hs_color_inclusion_msg + ): object, + probatio.Required(CONF_COLOR): { + probatio.Required(CONF_GA_HUE, msg=_hs_color_inclusion_msg): object, + probatio.Required( CONF_GA_SATURATION, msg=_hs_color_inclusion_msg ): object, }, }, - extra=vol.ALLOW_EXTRA, + extra=probatio.ALLOW_EXTRA, ), - vol.Schema( # hs-colors not used + probatio.Schema( # hs-colors not used { - vol.Optional(CONF_COLOR): { - vol.Optional(CONF_GA_HUE): None, - vol.Optional(CONF_GA_SATURATION): None, + probatio.Optional(CONF_COLOR): { + probatio.Optional(CONF_GA_HUE): None, + probatio.Optional(CONF_GA_SATURATION): None, }, }, - extra=vol.ALLOW_EXTRA, + extra=probatio.ALLOW_EXTRA, ), msg=_hs_color_inclusion_msg, ), ) -NOTIFY_KNX_SCHEMA = vol.Schema( +NOTIFY_KNX_SCHEMA = probatio.Schema( { - vol.Required(CONF_GA_SEND): GASelector( + probatio.Required(CONF_GA_SEND): GASelector( state=False, passive=False, write_required=True, dpt=["string"] ), } @@ -521,29 +537,31 @@ def _number_limit_sub_validator(config: dict) -> dict: NUMBER_KNX_SCHEMA = AllSerializeFirst( - vol.Schema( + probatio.Schema( { - vol.Required(CONF_GA_SENSOR): GASelector( + probatio.Required(CONF_GA_SENSOR): GASelector( write_required=True, dpt=["numeric"] ), - vol.Optional( + probatio.Optional( CONF_RESPOND_TO_READ, default=False ): selector.BooleanSelector(), "section_advanced_options": KNXSectionFlat(collapsible=True), - vol.Required(CONF_MODE, default=NumberMode.AUTO): selector.SelectSelector( + probatio.Required( + CONF_MODE, default=NumberMode.AUTO + ): selector.SelectSelector( selector.SelectSelectorConfig( options=list(NumberMode), translation_key="component.knx.config_panel.entities.create.number.knx.mode", ), ), - vol.Optional(NumberConf.MIN): selector.NumberSelector(), - vol.Optional(NumberConf.MAX): selector.NumberSelector(), - vol.Optional(NumberConf.STEP): selector.NumberSelector( + probatio.Optional(NumberConf.MIN): selector.NumberSelector(), + probatio.Optional(NumberConf.MAX): selector.NumberSelector(), + probatio.Optional(NumberConf.STEP): selector.NumberSelector( selector.NumberSelectorConfig( min=0, step="any", mode=selector.NumberSelectorMode.BOX ) ), - vol.Optional(CONF_UNIT_OF_MEASUREMENT): selector.SelectSelector( + probatio.Optional(CONF_UNIT_OF_MEASUREMENT): selector.SelectSelector( selector.SelectSelectorConfig( options=sorted( { @@ -557,7 +575,7 @@ NUMBER_KNX_SCHEMA = AllSerializeFirst( custom_value=True, ), ), - vol.Optional(CONF_DEVICE_CLASS): selector.SelectSelector( + probatio.Optional(CONF_DEVICE_CLASS): selector.SelectSelector( selector.SelectSelectorConfig( options=[cls.value for cls in NumberDeviceClass], # should align with sensor @@ -565,27 +583,27 @@ NUMBER_KNX_SCHEMA = AllSerializeFirst( sort=True, ) ), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), }, ), _number_limit_sub_validator, ) -SCENE_KNX_SCHEMA = vol.Schema( +SCENE_KNX_SCHEMA = probatio.Schema( { - vol.Required(CONF_GA_SCENE): GASelector( + probatio.Required(CONF_GA_SCENE): GASelector( state=False, passive=False, write_required=True, valid_dpt=["17.001", "18.001"], ), - vol.Required(SceneConf.SCENE_NUMBER): AllSerializeFirst( + probatio.Required(SceneConf.SCENE_NUMBER): AllSerializeFirst( selector.NumberSelector( selector.NumberSelectorConfig( min=1, max=64, step=1, mode=selector.NumberSelectorMode.BOX ) ), - vol.Coerce(int), + probatio.Coerce(int), ), }, ) @@ -608,7 +626,7 @@ def _select_options_sub_validator(config: dict) -> dict: if SelectConf.GA_ENUM in source: dpt = source[SelectConf.GA_ENUM].get(CONF_DPT) if dpt is None or get_supported_dpts()[dpt]["dpt_class"] != "enum": - raise vol.Invalid( + raise probatio.Invalid( "An enum data point type is required", path=[SelectConf.OPTIONS_SOURCE, SelectConf.GA_ENUM], ) @@ -617,7 +635,7 @@ def _select_options_sub_validator(config: dict) -> dict: error_path: list[Hashable] = [SelectConf.OPTIONS_SOURCE, SelectConf.CUSTOM_OPTIONS] options = source[SelectConf.CUSTOM_OPTIONS] if not options: - raise vol.Invalid("At least one option is required", path=error_path) + raise probatio.Invalid("At least one option is required", path=error_path) dpt = source[SelectConf.GA_CUSTOM].get(CONF_DPT) transcoder = DPTBase.parse_transcoder(dpt) if dpt is not None else None @@ -628,12 +646,14 @@ def _select_options_sub_validator(config: dict) -> dict: for option in options: name = option[SelectConf.OPTION] if name in options_seen: - raise vol.Invalid(f"Duplicate option not allowed: {name}", path=error_path) + raise probatio.Invalid( + f"Duplicate option not allowed: {name}", path=error_path + ) options_seen.add(name) if CONF_VALUE in option: if transcoder is None: - raise vol.Invalid( + raise probatio.Invalid( f"A data point type is required for typed option '{name}'", path=error_path, ) @@ -643,7 +663,7 @@ def _select_options_sub_validator(config: dict) -> dict: byteorder="big", ) except ConversionError as ex: - raise vol.Invalid( + raise probatio.Invalid( f"Value invalid for option '{name}' with DPT " f"{transcoder.dpt_number_str()}", path=error_path, @@ -658,7 +678,7 @@ def _select_options_sub_validator(config: dict) -> dict: if transcoder is not None else "the other options" ) - raise vol.Invalid( + raise probatio.Invalid( f"Payload length {option_length} of option '{name}' doesn't " f"match payload length {payload_length} of {expected}", path=error_path, @@ -666,7 +686,7 @@ def _select_options_sub_validator(config: dict) -> dict: payload = int(option[CONF_PAYLOAD], 16) if payload in payloads_seen: - raise vol.Invalid( + raise probatio.Invalid( f"Duplicate payload not allowed for option '{name}'", path=error_path ) payloads_seen.add(payload) @@ -674,13 +694,13 @@ def _select_options_sub_validator(config: dict) -> dict: SELECT_KNX_SCHEMA = AllSerializeFirst( - vol.Schema( + probatio.Schema( { - vol.Required(SelectConf.OPTIONS_SOURCE): GroupSelect( + probatio.Required(SelectConf.OPTIONS_SOURCE): GroupSelect( GroupSelectOption( translation_key="from_dpt", schema={ - vol.Required(SelectConf.GA_ENUM): GASelector( + probatio.Required(SelectConf.GA_ENUM): GASelector( write_required=True, dpt=["enum"] ), }, @@ -688,55 +708,67 @@ SELECT_KNX_SCHEMA = AllSerializeFirst( GroupSelectOption( translation_key="custom", schema={ - vol.Required(SelectConf.GA_CUSTOM): GASelector( + probatio.Required(SelectConf.GA_CUSTOM): GASelector( write_required=True, dpt=["numeric", "enum", "complex", "string"], dpt_required=False, ), - vol.Required( + probatio.Required( SelectConf.CUSTOM_OPTIONS ): KnxSelectOptionsSelector(ga_path=SelectConf.GA_CUSTOM), }, ), collapsible=False, ), - vol.Optional( + probatio.Optional( CONF_RESPOND_TO_READ, default=False ): selector.BooleanSelector(), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), } ), _select_options_sub_validator, ) -SWITCH_KNX_SCHEMA = vol.Schema( +SWITCH_KNX_SCHEMA = probatio.Schema( { - vol.Required(CONF_GA_SWITCH): GASelector(write_required=True, valid_dpt="1"), - vol.Optional(CONF_INVERT, default=False): selector.BooleanSelector(), - vol.Optional(CONF_RESPOND_TO_READ, default=False): selector.BooleanSelector(), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Required(CONF_GA_SWITCH): GASelector( + write_required=True, valid_dpt="1" + ), + probatio.Optional(CONF_INVERT, default=False): selector.BooleanSelector(), + probatio.Optional( + CONF_RESPOND_TO_READ, default=False + ): selector.BooleanSelector(), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), }, ) -TEXT_KNX_SCHEMA = vol.Schema( +TEXT_KNX_SCHEMA = probatio.Schema( { - vol.Required(CONF_GA_TEXT): GASelector(write_required=True, dpt=["string"]), - vol.Required(CONF_MODE, default=TextMode.TEXT): selector.SelectSelector( + probatio.Required(CONF_GA_TEXT): GASelector( + write_required=True, dpt=["string"] + ), + probatio.Required(CONF_MODE, default=TextMode.TEXT): selector.SelectSelector( selector.SelectSelectorConfig( options=list(TextMode), translation_key="component.knx.config_panel.entities.create.text.knx.mode", ), ), - vol.Optional(CONF_RESPOND_TO_READ, default=False): selector.BooleanSelector(), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Optional( + CONF_RESPOND_TO_READ, default=False + ): selector.BooleanSelector(), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), }, ) -TIME_KNX_SCHEMA = vol.Schema( +TIME_KNX_SCHEMA = probatio.Schema( { - vol.Required(CONF_GA_TIME): GASelector(write_required=True, valid_dpt="10.001"), - vol.Optional(CONF_RESPOND_TO_READ, default=False): selector.BooleanSelector(), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Required(CONF_GA_TIME): GASelector( + write_required=True, valid_dpt="10.001" + ), + probatio.Optional( + CONF_RESPOND_TO_READ, default=False + ): selector.BooleanSelector(), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), } ) @@ -757,36 +789,36 @@ class ConfClimateFanSpeedMode(StrEnum): STEPS = "5.010" -CLIMATE_KNX_SCHEMA = vol.Schema( +CLIMATE_KNX_SCHEMA = probatio.Schema( { - vol.Required(CONF_GA_TEMPERATURE_CURRENT): GASelector( + probatio.Required(CONF_GA_TEMPERATURE_CURRENT): GASelector( write=False, state_required=True, valid_dpt="9.001" ), - vol.Optional(CONF_GA_HUMIDITY_CURRENT): GASelector( + probatio.Optional(CONF_GA_HUMIDITY_CURRENT): GASelector( write=False, valid_dpt="9.007" ), - vol.Required(CONF_TARGET_TEMPERATURE): GroupSelect( + probatio.Required(CONF_TARGET_TEMPERATURE): GroupSelect( GroupSelectOption( translation_key="group_direct_temp", schema={ - vol.Required(CONF_GA_TEMPERATURE_TARGET): GASelector( + probatio.Required(CONF_GA_TEMPERATURE_TARGET): GASelector( write_required=True, valid_dpt="9.001" ), - vol.Required( + probatio.Required( ClimateConf.MIN_TEMP, default=7 ): selector.NumberSelector( selector.NumberSelectorConfig( min=-20, max=80, step=1, unit_of_measurement="°C" ) ), - vol.Required( + probatio.Required( ClimateConf.MAX_TEMP, default=28 ): selector.NumberSelector( selector.NumberSelectorConfig( min=0, max=100, step=1, unit_of_measurement="°C" ) ), - vol.Required( + probatio.Required( ClimateConf.TEMPERATURE_STEP, default=0.1 ): selector.NumberSelector( selector.NumberSelectorConfig( @@ -798,29 +830,29 @@ CLIMATE_KNX_SCHEMA = vol.Schema( GroupSelectOption( translation_key="group_setpoint_shift", schema={ - vol.Required(CONF_GA_TEMPERATURE_TARGET): GASelector( + probatio.Required(CONF_GA_TEMPERATURE_TARGET): GASelector( write=False, state_required=True, valid_dpt="9.001" ), - vol.Required(CONF_GA_SETPOINT_SHIFT): GASelector( + probatio.Required(CONF_GA_SETPOINT_SHIFT): GASelector( write_required=True, state_required=True, dpt=ConfSetpointShiftMode, ), - vol.Required( + probatio.Required( ClimateConf.SETPOINT_SHIFT_MIN, default=-6 ): selector.NumberSelector( selector.NumberSelectorConfig( min=-32, max=0, step=1, unit_of_measurement="K" ) ), - vol.Required( + probatio.Required( ClimateConf.SETPOINT_SHIFT_MAX, default=6 ): selector.NumberSelector( selector.NumberSelectorConfig( min=0, max=32, step=1, unit_of_measurement="K" ) ), - vol.Required( + probatio.Required( ClimateConf.TEMPERATURE_STEP, default=0.1 ): selector.NumberSelector( selector.NumberSelectorConfig( @@ -832,27 +864,31 @@ CLIMATE_KNX_SCHEMA = vol.Schema( collapsible=False, ), "section_activity": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_ACTIVE): GASelector(write=False, valid_dpt="1"), - vol.Optional(CONF_GA_VALVE): GASelector(write=False, valid_dpt="5.001"), + probatio.Optional(CONF_GA_ACTIVE): GASelector(write=False, valid_dpt="1"), + probatio.Optional(CONF_GA_VALVE): GASelector(write=False, valid_dpt="5.001"), "section_operation_mode": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_OPERATION_MODE): GASelector(valid_dpt="20.102"), - vol.Optional(CONF_IGNORE_AUTO_MODE): selector.BooleanSelector(), + probatio.Optional(CONF_GA_OPERATION_MODE): GASelector(valid_dpt="20.102"), + probatio.Optional(CONF_IGNORE_AUTO_MODE): selector.BooleanSelector(), "section_operation_mode_individual": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_OP_MODE_COMFORT): GASelector(state=False, valid_dpt="1"), - vol.Optional(CONF_GA_OP_MODE_ECO): GASelector(state=False, valid_dpt="1"), - vol.Optional(CONF_GA_OP_MODE_STANDBY): GASelector(state=False, valid_dpt="1"), - vol.Optional(CONF_GA_OP_MODE_PROTECTION): GASelector( + probatio.Optional(CONF_GA_OP_MODE_COMFORT): GASelector( + state=False, valid_dpt="1" + ), + probatio.Optional(CONF_GA_OP_MODE_ECO): GASelector(state=False, valid_dpt="1"), + probatio.Optional(CONF_GA_OP_MODE_STANDBY): GASelector( + state=False, valid_dpt="1" + ), + probatio.Optional(CONF_GA_OP_MODE_PROTECTION): GASelector( state=False, valid_dpt="1" ), "section_heat_cool": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_HEAT_COOL): GASelector(valid_dpt="1.100"), + probatio.Optional(CONF_GA_HEAT_COOL): GASelector(valid_dpt="1.100"), "section_on_off": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_ON_OFF): GASelector(valid_dpt="1"), - vol.Optional(ClimateConf.ON_OFF_INVERT): selector.BooleanSelector(), + probatio.Optional(CONF_GA_ON_OFF): GASelector(valid_dpt="1"), + probatio.Optional(ClimateConf.ON_OFF_INVERT): selector.BooleanSelector(), "section_controller_mode": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_CONTROLLER_MODE): GASelector(valid_dpt="20.105"), - vol.Optional(CONF_GA_CONTROLLER_STATUS): GASelector(write=False), - vol.Required( + probatio.Optional(CONF_GA_CONTROLLER_MODE): GASelector(valid_dpt="20.105"), + probatio.Optional(CONF_GA_CONTROLLER_STATUS): GASelector(write=False), + probatio.Required( ClimateConf.DEFAULT_CONTROLLER_MODE, default=HVACMode.HEAT ): selector.SelectSelector( selector.SelectSelectorConfig( @@ -861,14 +897,14 @@ CLIMATE_KNX_SCHEMA = vol.Schema( ) ), "section_fan": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_FAN_SPEED): GASelector(dpt=ConfClimateFanSpeedMode), - vol.Required(ClimateConf.FAN_MAX_STEP, default=3): AllSerializeFirst( + probatio.Optional(CONF_GA_FAN_SPEED): GASelector(dpt=ConfClimateFanSpeedMode), + probatio.Required(ClimateConf.FAN_MAX_STEP, default=3): AllSerializeFirst( selector.NumberSelector( selector.NumberSelectorConfig(min=1, max=100, step=1) ), - vol.Coerce(int), + probatio.Coerce(int), ), - vol.Required( + probatio.Required( ClimateConf.FAN_ZERO_MODE, default=FanZeroMode.OFF ): selector.SelectSelector( selector.SelectSelectorConfig( @@ -876,9 +912,9 @@ CLIMATE_KNX_SCHEMA = vol.Schema( translation_key="component.knx.config_panel.entities.create.climate.knx.fan_zero_mode", ) ), - vol.Optional(CONF_GA_FAN_SWING): GASelector(valid_dpt="1"), - vol.Optional(CONF_GA_FAN_SWING_HORIZONTAL): GASelector(valid_dpt="1"), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Optional(CONF_GA_FAN_SWING): GASelector(valid_dpt="1"), + probatio.Optional(CONF_GA_FAN_SWING_HORIZONTAL): GASelector(valid_dpt="1"), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), }, ) @@ -891,13 +927,13 @@ def _sensor_attribute_sub_validator(config: dict) -> dict: SENSOR_KNX_SCHEMA = AllSerializeFirst( - vol.Schema( + probatio.Schema( { - vol.Required(CONF_GA_SENSOR): GASelector( + probatio.Required(CONF_GA_SENSOR): GASelector( write=False, state_required=True, dpt=["numeric", "string"] ), "section_advanced_options": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_UNIT_OF_MEASUREMENT): selector.SelectSelector( + probatio.Optional(CONF_UNIT_OF_MEASUREMENT): selector.SelectSelector( selector.SelectSelectorConfig( options=sorted( { @@ -912,7 +948,7 @@ SENSOR_KNX_SCHEMA = AllSerializeFirst( custom_value=True, ), ), - vol.Optional(CONF_DEVICE_CLASS): selector.SelectSelector( + probatio.Optional(CONF_DEVICE_CLASS): selector.SelectSelector( selector.SelectSelectorConfig( options=[ cls.value @@ -923,15 +959,15 @@ SENSOR_KNX_SCHEMA = AllSerializeFirst( sort=True, ) ), - vol.Optional(CONF_SENSOR_STATE_CLASS): selector.SelectSelector( + probatio.Optional(CONF_SENSOR_STATE_CLASS): selector.SelectSelector( selector.SelectSelectorConfig( options=list(SensorStateClass), translation_key="component.knx.selector.sensor_state_class", mode=selector.SelectSelectorMode.DROPDOWN, ) ), - vol.Optional(CONF_ALWAYS_CALLBACK): selector.BooleanSelector(), - vol.Required(CONF_SYNC_STATE, default=True): SyncStateSelector( + probatio.Optional(CONF_ALWAYS_CALLBACK): selector.BooleanSelector(), + probatio.Required(CONF_SYNC_STATE, default=True): SyncStateSelector( allow_false=True ), }, @@ -939,38 +975,46 @@ SENSOR_KNX_SCHEMA = AllSerializeFirst( _sensor_attribute_sub_validator, ) -WEATHER_KNX_SCHEMA = vol.Schema( +WEATHER_KNX_SCHEMA = probatio.Schema( { - vol.Required(CONF_GA_TEMPERATURE): GASelector( + probatio.Required(CONF_GA_TEMPERATURE): GASelector( write=False, state_required=True, valid_dpt="9.001" ), - vol.Optional(CONF_GA_HUMIDITY): GASelector(write=False, valid_dpt="9.007"), - vol.Optional(CONF_GA_AIR_PRESSURE): GASelector( + probatio.Optional(CONF_GA_HUMIDITY): GASelector(write=False, valid_dpt="9.007"), + probatio.Optional(CONF_GA_AIR_PRESSURE): GASelector( write=False, valid_dpt=["9.006", "14.058"] ), - vol.Optional(CONF_GA_WIND_SPEED): GASelector(write=False, valid_dpt="9.005"), - vol.Optional(CONF_GA_WIND_BEARING): GASelector(write=False, valid_dpt="5.003"), + probatio.Optional(CONF_GA_WIND_SPEED): GASelector( + write=False, valid_dpt="9.005" + ), + probatio.Optional(CONF_GA_WIND_BEARING): GASelector( + write=False, valid_dpt="5.003" + ), "section_brightness": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_BRIGHTNESS_EAST): GASelector( + probatio.Optional(CONF_GA_BRIGHTNESS_EAST): GASelector( write=False, valid_dpt="9.004" ), - vol.Optional(CONF_GA_BRIGHTNESS_SOUTH): GASelector( + probatio.Optional(CONF_GA_BRIGHTNESS_SOUTH): GASelector( write=False, valid_dpt="9.004" ), - vol.Optional(CONF_GA_BRIGHTNESS_WEST): GASelector( + probatio.Optional(CONF_GA_BRIGHTNESS_WEST): GASelector( write=False, valid_dpt="9.004" ), - vol.Optional(CONF_GA_BRIGHTNESS_NORTH): GASelector( + probatio.Optional(CONF_GA_BRIGHTNESS_NORTH): GASelector( write=False, valid_dpt="9.004" ), "section_day_night": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_DAY_NIGHT): GASelector(write=False, valid_dpt="1.024"), - vol.Optional(CONF_INVERT_DAY_NIGHT, default=False): selector.BooleanSelector(), + probatio.Optional(CONF_GA_DAY_NIGHT): GASelector( + write=False, valid_dpt="1.024" + ), + probatio.Optional( + CONF_INVERT_DAY_NIGHT, default=False + ): selector.BooleanSelector(), "section_alarms": KNXSectionFlat(collapsible=True), - vol.Optional(CONF_GA_RAIN_ALARM): GASelector(write=False, valid_dpt="1"), - vol.Optional(CONF_GA_FROST_ALARM): GASelector(write=False, valid_dpt="1"), - vol.Optional(CONF_GA_WIND_ALARM): GASelector(write=False, valid_dpt="1"), - vol.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), + probatio.Optional(CONF_GA_RAIN_ALARM): GASelector(write=False, valid_dpt="1"), + probatio.Optional(CONF_GA_FROST_ALARM): GASelector(write=False, valid_dpt="1"), + probatio.Optional(CONF_GA_WIND_ALARM): GASelector(write=False, valid_dpt="1"), + probatio.Optional(CONF_SYNC_STATE, default=True): SyncStateSelector(), } ) @@ -994,31 +1038,31 @@ KNX_SCHEMA_FOR_PLATFORM = { Platform.WEATHER: WEATHER_KNX_SCHEMA, } -ENTITY_STORE_DATA_SCHEMA: VolSchemaType = vol.All( - vol.Schema( +ENTITY_STORE_DATA_SCHEMA = probatio.All( + probatio.Schema( { - vol.Required(CONF_PLATFORM): vol.All( - vol.Coerce(Platform), - vol.In(SUPPORTED_PLATFORMS_UI), + probatio.Required(CONF_PLATFORM): probatio.All( + probatio.Coerce(Platform), + probatio.In(SUPPORTED_PLATFORMS_UI), ), - vol.Required(CONF_DATA): dict, + probatio.Required(CONF_DATA): dict, }, - extra=vol.ALLOW_EXTRA, + extra=probatio.ALLOW_EXTRA, ), - cv.key_value_schemas( + probatio.TaggedUnion( CONF_PLATFORM, { - platform: vol.Schema( + platform: probatio.Schema( { - vol.Required(CONF_DATA): vol.Schema( + probatio.Required(CONF_DATA): probatio.Schema( { - vol.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA, - vol.Required(DOMAIN): knx_schema, + probatio.Required(CONF_ENTITY): BASE_ENTITY_SCHEMA, + probatio.Required(DOMAIN): knx_schema, }, - extra=vol.PREVENT_EXTRA, # restrict in data key for yaml edit + extra=probatio.PREVENT_EXTRA, # restrict in data key for yaml edit ), }, - extra=vol.ALLOW_EXTRA, # eg. "type" from WS-endpoint when validating directly + extra=probatio.ALLOW_EXTRA, # eg. "type" from WS-endpoint when validating directly ) for platform, knx_schema in KNX_SCHEMA_FOR_PLATFORM.items() }, @@ -1026,11 +1070,13 @@ ENTITY_STORE_DATA_SCHEMA: VolSchemaType = vol.All( ) CREATE_ENTITY_BASE_SCHEMA: VolDictType = { - vol.Required(CONF_PLATFORM): str, - vol.Required(CONF_DATA): dict, # validated by ENTITY_STORE_DATA_SCHEMA for platform + probatio.Required(CONF_PLATFORM): str, + probatio.Required( + CONF_DATA + ): dict, # validated by ENTITY_STORE_DATA_SCHEMA for platform } UPDATE_ENTITY_BASE_SCHEMA = { - vol.Required(CONF_ENTITY_ID): str, + probatio.Required(CONF_ENTITY_ID): str, **CREATE_ENTITY_BASE_SCHEMA, } diff --git a/homeassistant/components/knx/storage/entity_store_validation.py b/homeassistant/components/knx/storage/entity_store_validation.py index 34b89bcf5442..57dea02ca0ec 100644 --- a/homeassistant/components/knx/storage/entity_store_validation.py +++ b/homeassistant/components/knx/storage/entity_store_validation.py @@ -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 diff --git a/homeassistant/components/knx/storage/expose_controller.py b/homeassistant/components/knx/storage/expose_controller.py index 96206d5c2355..24f69afc53b2 100644 --- a/homeassistant/components/knx/storage/expose_controller.py +++ b/homeassistant/components/knx/storage/expose_controller.py @@ -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, ) diff --git a/homeassistant/components/knx/storage/knx_selector.py b/homeassistant/components/knx/storage/knx_selector.py index 5bc877d2cde7..49d6012d84d8 100644 --- a/homeassistant/components/knx/storage/knx_selector.py +++ b/homeassistant/components/knx/storage/knx_selector.py @@ -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 } diff --git a/homeassistant/components/knx/storage/time_server.py b/homeassistant/components/knx/storage/time_server.py index ba1e0c323bf1..1ef5c7b81590 100644 --- a/homeassistant/components/knx/storage/time_server.py +++ b/homeassistant/components/knx/storage/time_server.py @@ -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" ), } diff --git a/tests/components/knx/test_config_store.py b/tests/components/knx/test_config_store.py index 6e9841436679..7edf7cc2f6ea 100644 --- a/tests/components/knx/test_config_store.py +++ b/tests/components/knx/test_config_store.py @@ -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( diff --git a/tests/components/knx/test_knx_selectors.py b/tests/components/knx/test_knx_selectors.py index d25db3dd5a48..7b7fdb39a52a 100644 --- a/tests/components/knx/test_knx_selectors.py +++ b/tests/components/knx/test_knx_selectors.py @@ -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 + )