Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix validation for zwave_js device trigger and condition (#54974)

pull/55044/head
Raman Gupta 2021-08-22 20:43:59 -04:00 committed by GitHub
parent 305475a635
commit 5f5c8ade41
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 260 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
homeassistant/components/zwave_js/device_condition.py
View File

@ -4,9 +4,12 @@ from __future__ import annotations
from typing import cast
import voluptuous as vol
from zwave_js_server.const import CommandClass, ConfigurationValueType
from zwave_js_server.const import CommandClass
from zwave_js_server.model.value import ConfigurationValue
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.const import CONF_CONDITION, CONF_DEVICE_ID, CONF_DOMAIN, CONF_TYPE
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
@ -22,7 +25,14 @@ from .const import (
ATTR_PROPERTY_KEY,
ATTR_VALUE,
)
from .helpers import async_get_node_from_device_id, get_zwave_value_from_config
from .helpers import (
async_get_node_from_device_id,
async_is_device_config_entry_not_loaded,
check_type_schema_map,
get_value_state_schema,
get_zwave_value_from_config,
remove_keys_with_empty_values,
)
CONF_SUBTYPE = "subtype"
CONF_VALUE_ID = "value_id"
@ -67,10 +77,21 @@ VALUE_CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend(
}
)
CONDITION_SCHEMA = vol.Any(
NODE_STATUS_CONDITION_SCHEMA,
CONFIG_PARAMETER_CONDITION_SCHEMA,
VALUE_CONDITION_SCHEMA,
TYPE_SCHEMA_MAP = {
NODE_STATUS_TYPE: NODE_STATUS_CONDITION_SCHEMA,
CONFIG_PARAMETER_TYPE: CONFIG_PARAMETER_CONDITION_SCHEMA,
VALUE_TYPE: VALUE_CONDITION_SCHEMA,
}
CONDITION_TYPE_SCHEMA = vol.Schema(
{vol.Required(CONF_TYPE): vol.In(TYPE_SCHEMA_MAP)}, extra=vol.ALLOW_EXTRA
)
CONDITION_SCHEMA = vol.All(
remove_keys_with_empty_values,
CONDITION_TYPE_SCHEMA,
check_type_schema_map(TYPE_SCHEMA_MAP),
)
@ -79,9 +100,18 @@ async def async_validate_condition_config(
) -> ConfigType:
"""Validate config."""
config = CONDITION_SCHEMA(config)
# We return early if the config entry for this device is not ready because we can't
# validate the value without knowing the state of the device
if async_is_device_config_entry_not_loaded(hass, config[CONF_DEVICE_ID]):
return config
if config[CONF_TYPE] == VALUE_TYPE:
node = async_get_node_from_device_id(hass, config[CONF_DEVICE_ID])
get_zwave_value_from_config(node, config)
try:
node = async_get_node_from_device_id(hass, config[CONF_DEVICE_ID])
get_zwave_value_from_config(node, config)
except vol.Invalid as err:
raise InvalidDeviceAutomationConfig(err.msg) from err
return config
@ -174,20 +204,8 @@ async def async_get_condition_capabilities(
# Add additional fields to the automation trigger UI
if config[CONF_TYPE] == CONFIG_PARAMETER_TYPE:
value_id = config[CONF_VALUE_ID]
config_value = cast(ConfigurationValue, node.values[value_id])
min_ = config_value.metadata.min
max_ = config_value.metadata.max
if config_value.configuration_value_type in (
ConfigurationValueType.RANGE,
ConfigurationValueType.MANUAL_ENTRY,
):
value_schema = vol.Range(min=min_, max=max_)
elif config_value.configuration_value_type == ConfigurationValueType.ENUMERATED:
value_schema = vol.In(
{int(k): v for k, v in config_value.metadata.states.items()}
)
else:
value_schema = get_value_state_schema(node.values[value_id])
if not value_schema:
return {}
return {"extra_fields": vol.Schema({vol.Required(ATTR_VALUE): value_schema})}

147
homeassistant/components/zwave_js/device_trigger.py
View File

@ -4,10 +4,13 @@ from __future__ import annotations
from typing import Any
import voluptuous as vol
from zwave_js_server.const import CommandClass, ConfigurationValueType
from zwave_js_server.const import CommandClass
from homeassistant.components.automation import AutomationActionType
from homeassistant.components.device_automation import DEVICE_TRIGGER_BASE_SCHEMA
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.components.homeassistant.triggers import event, state
from homeassistant.const import (
CONF_DEVICE_ID,
@ -46,12 +49,20 @@ from .const import (
from .helpers import (
async_get_node_from_device_id,
async_get_node_status_sensor_entity_id,
async_is_device_config_entry_not_loaded,
check_type_schema_map,
copy_available_params,
get_value_state_schema,
get_zwave_value_from_config,
remove_keys_with_empty_values,
)
from .triggers.value_updated import (
ATTR_FROM,
ATTR_TO,
PLATFORM_TYPE as VALUE_UPDATED_PLATFORM_TYPE,
)
from .triggers.value_updated import ATTR_FROM, ATTR_TO
CONF_SUBTYPE = "subtype"
CONF_VALUE_ID = "value_id"
# Trigger types
ENTRY_CONTROL_NOTIFICATION = "event.notification.entry_control"
@ -59,8 +70,8 @@ NOTIFICATION_NOTIFICATION = "event.notification.notification"
BASIC_VALUE_NOTIFICATION = "event.value_notification.basic"
CENTRAL_SCENE_VALUE_NOTIFICATION = "event.value_notification.central_scene"
SCENE_ACTIVATION_VALUE_NOTIFICATION = "event.value_notification.scene_activation"
CONFIG_PARAMETER_VALUE_UPDATED = f"{DOMAIN}.value_updated.config_parameter"
VALUE_VALUE_UPDATED = f"{DOMAIN}.value_updated.value"
CONFIG_PARAMETER_VALUE_UPDATED = f"{VALUE_UPDATED_PLATFORM_TYPE}.config_parameter"
VALUE_VALUE_UPDATED = f"{VALUE_UPDATED_PLATFORM_TYPE}.value"
NODE_STATUS = "state.node_status"
VALUE_SCHEMA = vol.Any(
@ -71,6 +82,7 @@ VALUE_SCHEMA = vol.Any(
cv.string,
)
NOTIFICATION_EVENT_CC_MAPPINGS = (
(ENTRY_CONTROL_NOTIFICATION, CommandClass.ENTRY_CONTROL),
(NOTIFICATION_NOTIFICATION, CommandClass.NOTIFICATION),
@ -104,7 +116,7 @@ ENTRY_CONTROL_NOTIFICATION_SCHEMA = BASE_EVENT_SCHEMA.extend(
BASE_VALUE_NOTIFICATION_EVENT_SCHEMA = BASE_EVENT_SCHEMA.extend(
{
vol.Required(ATTR_PROPERTY): vol.Any(int, str),
vol.Required(ATTR_PROPERTY_KEY): vol.Any(None, int, str),
vol.Optional(ATTR_PROPERTY_KEY): vol.Any(int, str),
vol.Required(ATTR_ENDPOINT): vol.Coerce(int),
vol.Optional(ATTR_VALUE): vol.Coerce(int),
vol.Required(CONF_SUBTYPE): cv.string,
@ -174,17 +186,61 @@ VALUE_VALUE_UPDATED_SCHEMA = BASE_VALUE_UPDATED_SCHEMA.extend(
}
)
TRIGGER_SCHEMA = vol.Any(
ENTRY_CONTROL_NOTIFICATION_SCHEMA,
NOTIFICATION_NOTIFICATION_SCHEMA,
BASIC_VALUE_NOTIFICATION_SCHEMA,
CENTRAL_SCENE_VALUE_NOTIFICATION_SCHEMA,
SCENE_ACTIVATION_VALUE_NOTIFICATION_SCHEMA,
CONFIG_PARAMETER_VALUE_UPDATED_SCHEMA,
VALUE_VALUE_UPDATED_SCHEMA,
NODE_STATUS_SCHEMA,
TYPE_SCHEMA_MAP = {
ENTRY_CONTROL_NOTIFICATION: ENTRY_CONTROL_NOTIFICATION_SCHEMA,
NOTIFICATION_NOTIFICATION: NOTIFICATION_NOTIFICATION_SCHEMA,
BASIC_VALUE_NOTIFICATION: BASIC_VALUE_NOTIFICATION_SCHEMA,
CENTRAL_SCENE_VALUE_NOTIFICATION: CENTRAL_SCENE_VALUE_NOTIFICATION_SCHEMA,
SCENE_ACTIVATION_VALUE_NOTIFICATION: SCENE_ACTIVATION_VALUE_NOTIFICATION_SCHEMA,
CONFIG_PARAMETER_VALUE_UPDATED: CONFIG_PARAMETER_VALUE_UPDATED_SCHEMA,
VALUE_VALUE_UPDATED: VALUE_VALUE_UPDATED_SCHEMA,
NODE_STATUS: NODE_STATUS_SCHEMA,
}
TRIGGER_TYPE_SCHEMA = vol.Schema(
{vol.Required(CONF_TYPE): vol.In(TYPE_SCHEMA_MAP)}, extra=vol.ALLOW_EXTRA
)
TRIGGER_SCHEMA = vol.All(
remove_keys_with_empty_values,
TRIGGER_TYPE_SCHEMA,
check_type_schema_map(TYPE_SCHEMA_MAP),
)
async def async_validate_trigger_config(
hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
config = TRIGGER_SCHEMA(config)
# We return early if the config entry for this device is not ready because we can't
# validate the value without knowing the state of the device
if async_is_device_config_entry_not_loaded(hass, config[CONF_DEVICE_ID]):
return config
trigger_type = config[CONF_TYPE]
if get_trigger_platform_from_type(trigger_type) == VALUE_UPDATED_PLATFORM_TYPE:
try:
node = async_get_node_from_device_id(hass, config[CONF_DEVICE_ID])
get_zwave_value_from_config(node, config)
except vol.Invalid as err:
raise InvalidDeviceAutomationConfig(err.msg) from err
return config
def get_trigger_platform_from_type(trigger_type: str) -> str:
"""Get trigger platform from Z-Wave JS trigger type."""
trigger_split = trigger_type.split(".")
# Our convention for trigger types is to have the trigger type at the beginning
# delimited by a `.`. For zwave_js triggers, there is a `.` in the name
trigger_platform = trigger_split[0]
if trigger_platform == DOMAIN:
return ".".join(trigger_split[:2])
return trigger_platform
async def async_get_triggers(
hass: HomeAssistant, device_id: str
@ -298,15 +354,6 @@ async def async_get_triggers(
return triggers
def copy_available_params(
input_dict: dict, output_dict: dict, params: list[str]
) -> None:
"""Copy available params from input into output."""
for param in params:
if (val := input_dict.get(param)) not in ("", None):
output_dict[param] = val
async def async_attach_trigger(
hass: HomeAssistant,
config: ConfigType,
@ -315,12 +362,7 @@ async def async_attach_trigger(
) -> CALLBACK_TYPE:
"""Attach a trigger."""
trigger_type = config[CONF_TYPE]
trigger_split = trigger_type.split(".")
# Our convention for trigger types is to have the trigger type at the beginning
# delimited by a `.`. For zwave_js triggers, there is a `.` in the name
trigger_platform = trigger_split[0]
if trigger_platform == DOMAIN:
trigger_platform = ".".join(trigger_split[:2])
trigger_platform = get_trigger_platform_from_type(trigger_type)
# Take input data from automation trigger UI and add it to the trigger we are
# attaching to
@ -379,14 +421,7 @@ async def async_attach_trigger(
hass, state_config, action, automation_info, platform_type="device"
)
if trigger_platform == f"{DOMAIN}.value_updated":
# Try to get the value to make sure the value ID is valid
try:
node = async_get_node_from_device_id(hass, config[CONF_DEVICE_ID])
get_zwave_value_from_config(node, config)
except (ValueError, vol.Invalid) as err:
raise HomeAssistantError("Invalid value specified") from err
if trigger_platform == VALUE_UPDATED_PLATFORM_TYPE:
zwave_js_config = {
state.CONF_PLATFORM: trigger_platform,
CONF_DEVICE_ID: config[CONF_DEVICE_ID],
@ -420,9 +455,7 @@ async def async_get_trigger_capabilities(
trigger_type = config[CONF_TYPE]
node = async_get_node_from_device_id(hass, config[CONF_DEVICE_ID])
value = (
get_zwave_value_from_config(node, config) if ATTR_PROPERTY in config else None
)
# Add additional fields to the automation trigger UI
if trigger_type == NOTIFICATION_NOTIFICATION:
return {
@ -462,33 +495,23 @@ async def async_get_trigger_capabilities(
CENTRAL_SCENE_VALUE_NOTIFICATION,
SCENE_ACTIVATION_VALUE_NOTIFICATION,
):
if value.metadata.states:
value_schema = vol.In({int(k): v for k, v in value.metadata.states.items()})
else:
value_schema = vol.All(
vol.Coerce(int),
vol.Range(min=value.metadata.min, max=value.metadata.max),
)
value_schema = get_value_state_schema(get_zwave_value_from_config(node, config))
# We should never get here, but just in case we should add a guard
if not value_schema:
return {}
return {"extra_fields": vol.Schema({vol.Optional(ATTR_VALUE): value_schema})}
if trigger_type == CONFIG_PARAMETER_VALUE_UPDATED:
# We can be more deliberate about the config parameter schema here because
# there are a limited number of types
if value.configuration_value_type == ConfigurationValueType.UNDEFINED:
value_schema = get_value_state_schema(get_zwave_value_from_config(node, config))
if not value_schema:
return {}
if value.configuration_value_type == ConfigurationValueType.ENUMERATED:
value_schema = vol.In({int(k): v for k, v in value.metadata.states.items()})
else:
value_schema = vol.All(
vol.Coerce(int),
vol.Range(min=value.metadata.min, max=value.metadata.max),
)
return {
"extra_fields": vol.Schema(
{
vol.Optional(state.CONF_FROM): value_schema,
vol.Optional(state.CONF_TO): value_schema,
vol.Optional(ATTR_FROM): value_schema,
vol.Optional(ATTR_TO): value_schema,
}
)
}
@ -509,8 +532,8 @@ async def async_get_trigger_capabilities(
vol.Required(ATTR_PROPERTY): cv.string,
vol.Optional(ATTR_PROPERTY_KEY): cv.string,
vol.Optional(ATTR_ENDPOINT): cv.string,
vol.Optional(state.CONF_FROM): cv.string,
vol.Optional(state.CONF_TO): cv.string,
vol.Optional(ATTR_FROM): cv.string,
vol.Optional(ATTR_TO): cv.string,
}
)
}

79
homeassistant/components/zwave_js/helpers.py
View File

@ -1,16 +1,21 @@
"""Helper functions for Z-Wave JS integration."""
from __future__ import annotations
from typing import Any, cast
from typing import Any, Callable, cast
import voluptuous as vol
from zwave_js_server.client import Client as ZwaveClient
from zwave_js_server.const import ConfigurationValueType
from zwave_js_server.model.node import Node as ZwaveNode
from zwave_js_server.model.value import Value as ZwaveValue, get_value_id
from zwave_js_server.model.value import (
ConfigurationValue,
Value as ZwaveValue,
get_value_id,
)
from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import __version__ as HA_VERSION
from homeassistant.config_entries import ConfigEntry, ConfigEntryState
from homeassistant.const import CONF_TYPE, __version__ as HA_VERSION
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
@ -242,3 +247,69 @@ def async_get_node_status_sensor_entity_id(
)
return entity_id
def remove_keys_with_empty_values(config: ConfigType) -> ConfigType:
"""Remove keys from config where the value is an empty string or None."""
return {key: value for key, value in config.items() if value not in ("", None)}
def check_type_schema_map(schema_map: dict[str, vol.Schema]) -> Callable:
"""Check type specific schema against config."""
def _check_type_schema(config: ConfigType) -> ConfigType:
"""Check type specific schema against config."""
return cast(ConfigType, schema_map[str(config[CONF_TYPE])](config))
return _check_type_schema
def copy_available_params(
input_dict: dict[str, Any], output_dict: dict[str, Any], params: list[str]
) -> None:
"""Copy available params from input into output."""
output_dict.update(
{param: input_dict[param] for param in params if param in input_dict}
)
@callback
def async_is_device_config_entry_not_loaded(
hass: HomeAssistant, device_id: str
) -> bool:
"""Return whether device's config entries are not loaded."""
dev_reg = dr.async_get(hass)
device = dev_reg.async_get(device_id)
assert device
return any(
(entry := hass.config_entries.async_get_entry(entry_id))
and entry.state != ConfigEntryState.LOADED
for entry_id in device.config_entries
)
def get_value_state_schema(
value: ZwaveValue,
) -> vol.Schema | None:
"""Return device automation schema for a config entry."""
if isinstance(value, ConfigurationValue):
min_ = value.metadata.min
max_ = value.metadata.max
if value.configuration_value_type in (
ConfigurationValueType.RANGE,
ConfigurationValueType.MANUAL_ENTRY,
):
return vol.All(vol.Coerce(int), vol.Range(min=min_, max=max_))
if value.configuration_value_type == ConfigurationValueType.ENUMERATED:
return vol.In({int(k): v for k, v in value.metadata.states.items()})
return None
if value.metadata.states:
return vol.In({int(k): v for k, v in value.metadata.states.items()})
return vol.All(
vol.Coerce(int),
vol.Range(min=value.metadata.min, max=value.metadata.max),
)

28
tests/components/zwave_js/test_device_condition.py
View File

@ -10,6 +10,9 @@ from zwave_js_server.const import CommandClass
from zwave_js_server.event import Event
from homeassistant.components import automation
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.components.zwave_js import DOMAIN, device_condition
from homeassistant.components.zwave_js.helpers import get_zwave_value_from_config
from homeassistant.exceptions import HomeAssistantError
@ -519,6 +522,7 @@ async def test_get_condition_capabilities_config_parameter(
{
"name": "value",
"required": True,
"type": "integer",
"valueMin": 0,
"valueMax": 124,
}
@ -565,6 +569,30 @@ async def test_failure_scenarios(hass, client, hank_binary_switch, integration):
== {}
)
INVALID_CONFIG = {
"condition": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "value",
"command_class": CommandClass.DOOR_LOCK.value,
"property": 9999,
"property_key": 9999,
"endpoint": 9999,
"value": 9999,
}
# Test that invalid config raises exception
with pytest.raises(InvalidDeviceAutomationConfig):
await device_condition.async_validate_condition_config(hass, INVALID_CONFIG)
# Unload entry so we can verify that validation will pass on an invalid config
# since we return early
await hass.config_entries.async_unload(integration.entry_id)
assert (
await device_condition.async_validate_condition_config(hass, INVALID_CONFIG)
== INVALID_CONFIG
)
async def test_get_value_from_config_failure(
hass, client, hank_binary_switch, integration

58
tests/components/zwave_js/test_device_trigger.py
View File

@ -8,11 +8,10 @@ from zwave_js_server.event import Event
from zwave_js_server.model.node import Node
from homeassistant.components import automation
from homeassistant.components.zwave_js import DOMAIN, device_trigger
from homeassistant.components.zwave_js.device_trigger import (
async_attach_trigger,
async_get_trigger_capabilities,
from homeassistant.components.device_automation.exceptions import (
InvalidDeviceAutomationConfig,
)
from homeassistant.components.zwave_js import DOMAIN, device_trigger
from homeassistant.components.zwave_js.helpers import (
async_get_node_status_sensor_entity_id,
)
@ -1281,12 +1280,12 @@ async def test_get_trigger_capabilities_value_updated_config_parameter_enumerate
async def test_failure_scenarios(hass, client, hank_binary_switch, integration):
"""Test failure scenarios."""
with pytest.raises(HomeAssistantError):
await async_attach_trigger(
await device_trigger.async_attach_trigger(
hass, {"type": "failed.test", "device_id": "invalid_device_id"}, None, {}
)
with pytest.raises(HomeAssistantError):
await async_attach_trigger(
await device_trigger.async_attach_trigger(
hass,
{"type": "event.failed_type", "device_id": "invalid_device_id"},
None,
@ -1297,12 +1296,12 @@ async def test_failure_scenarios(hass, client, hank_binary_switch, integration):
device = async_entries_for_config_entry(dev_reg, integration.entry_id)[0]
with pytest.raises(HomeAssistantError):
await async_attach_trigger(
await device_trigger.async_attach_trigger(
hass, {"type": "failed.test", "device_id": device.id}, None, {}
)
with pytest.raises(HomeAssistantError):
await async_attach_trigger(
await device_trigger.async_attach_trigger(
hass,
{"type": "event.failed_type", "device_id": device.id},
None,
@ -1310,29 +1309,13 @@ async def test_failure_scenarios(hass, client, hank_binary_switch, integration):
)
with pytest.raises(HomeAssistantError):
await async_attach_trigger(
await device_trigger.async_attach_trigger(
hass,
{"type": "state.failed_type", "device_id": device.id},
None,
{},
)
with pytest.raises(HomeAssistantError):
await async_attach_trigger(
hass,
{
"device_id": device.id,
"type": "zwave_js.value_updated.value",
"command_class": CommandClass.DOOR_LOCK.value,
"property": -1234,
"property_key": None,
"endpoint": None,
"from": "open",
},
None,
{},
)
with patch(
"homeassistant.components.zwave_js.device_trigger.async_get_node_from_device_id",
return_value=None,
@ -1341,7 +1324,7 @@ async def test_failure_scenarios(hass, client, hank_binary_switch, integration):
return_value=None,
):
assert (
await async_get_trigger_capabilities(
await device_trigger.async_get_trigger_capabilities(
hass, {"type": "failed.test", "device_id": "invalid_device_id"}
)
== {}
@ -1349,3 +1332,26 @@ async def test_failure_scenarios(hass, client, hank_binary_switch, integration):
with pytest.raises(HomeAssistantError):
async_get_node_status_sensor_entity_id(hass, "invalid_device_id")
INVALID_CONFIG = {
"platform": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "zwave_js.value_updated.value",
"command_class": CommandClass.DOOR_LOCK.value,
"property": 9999,
"property_key": 9999,
"endpoint": 9999,
}
# Test that invalid config raises exception
with pytest.raises(InvalidDeviceAutomationConfig):
await device_trigger.async_validate_trigger_config(hass, INVALID_CONFIG)
# Unload entry so we can verify that validation will pass on an invalid config
# since we return early
await hass.config_entries.async_unload(integration.entry_id)
assert (
await device_trigger.async_validate_trigger_config(hass, INVALID_CONFIG)
== INVALID_CONFIG
)