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

Fix zwave_js custom trigger validation bug (#72656) 

* Fix zwave_js custom trigger validation bug

* update comments

* Switch to ValueError

* Switch to ValueError
pull/72824/head
Raman Gupta 2022-05-29 12:30:00 -04:00 committed by Paulus Schoutsen
parent c62692dff1
commit f039aac31c
3 changed files with 241 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
homeassistant/components/zwave_js/triggers/event.py
View File

@ -8,7 +8,7 @@ import voluptuous as vol
from zwave_js_server.client import Client
from zwave_js_server.model.controller import CONTROLLER_EVENT_MODEL_MAP
from zwave_js_server.model.driver import DRIVER_EVENT_MODEL_MAP
from zwave_js_server.model.node import NODE_EVENT_MODEL_MAP, Node
from zwave_js_server.model.node import NODE_EVENT_MODEL_MAP
from homeassistant.components.automation import (
AutomationActionType,
@ -20,7 +20,6 @@ from homeassistant.components.zwave_js.const import (
ATTR_EVENT_DATA,
ATTR_EVENT_SOURCE,
ATTR_NODE_ID,
ATTR_NODES,
ATTR_PARTIAL_DICT_MATCH,
DATA_CLIENT,
DOMAIN,
@ -116,23 +115,21 @@ async def async_validate_trigger_config(
"""Validate config."""
config = TRIGGER_SCHEMA(config)
if async_bypass_dynamic_config_validation(hass, config):
return config
if config[ATTR_EVENT_SOURCE] == "node":
config[ATTR_NODES] = async_get_nodes_from_targets(hass, config)
if not config[ATTR_NODES]:
raise vol.Invalid(
f"No nodes found for given {ATTR_DEVICE_ID}s or {ATTR_ENTITY_ID}s."
)
if ATTR_CONFIG_ENTRY_ID not in config:
return config
if ATTR_CONFIG_ENTRY_ID in config:
entry_id = config[ATTR_CONFIG_ENTRY_ID]
if hass.config_entries.async_get_entry(entry_id) is None:
raise vol.Invalid(f"Config entry '{entry_id}' not found")
if async_bypass_dynamic_config_validation(hass, config):
return config
if config[ATTR_EVENT_SOURCE] == "node" and not async_get_nodes_from_targets(
hass, config
):
raise vol.Invalid(
f"No nodes found for given {ATTR_DEVICE_ID}s or {ATTR_ENTITY_ID}s."
)
return config
@ -145,7 +142,12 @@ async def async_attach_trigger(
platform_type: str = PLATFORM_TYPE,
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
nodes: set[Node] = config.get(ATTR_NODES, {})
dev_reg = dr.async_get(hass)
nodes = async_get_nodes_from_targets(hass, config, dev_reg=dev_reg)
if config[ATTR_EVENT_SOURCE] == "node" and not nodes:
raise ValueError(
f"No nodes found for given {ATTR_DEVICE_ID}s or {ATTR_ENTITY_ID}s."
)
event_source = config[ATTR_EVENT_SOURCE]
event_name = config[ATTR_EVENT]
@ -200,8 +202,6 @@ async def async_attach_trigger(
hass.async_run_hass_job(job, {"trigger": payload})
dev_reg = dr.async_get(hass)
if not nodes:
entry_id = config[ATTR_CONFIG_ENTRY_ID]
client: Client = hass.data[DOMAIN][entry_id][DATA_CLIENT]

12
homeassistant/components/zwave_js/triggers/value_updated.py
View File

@ -5,7 +5,6 @@ import functools
import voluptuous as vol
from zwave_js_server.const import CommandClass
from zwave_js_server.model.node import Node
from zwave_js_server.model.value import Value, get_value_id
from homeassistant.components.automation import (
@ -20,7 +19,6 @@ from homeassistant.components.zwave_js.const import (
ATTR_CURRENT_VALUE_RAW,
ATTR_ENDPOINT,
ATTR_NODE_ID,
ATTR_NODES,
ATTR_PREVIOUS_VALUE,
ATTR_PREVIOUS_VALUE_RAW,
ATTR_PROPERTY,
@ -79,8 +77,7 @@ async def async_validate_trigger_config(
if async_bypass_dynamic_config_validation(hass, config):
return config
config[ATTR_NODES] = async_get_nodes_from_targets(hass, config)
if not config[ATTR_NODES]:
if not async_get_nodes_from_targets(hass, config):
raise vol.Invalid(
f"No nodes found for given {ATTR_DEVICE_ID}s or {ATTR_ENTITY_ID}s."
)
@ -96,7 +93,11 @@ async def async_attach_trigger(
platform_type: str = PLATFORM_TYPE,
) -> CALLBACK_TYPE:
"""Listen for state changes based on configuration."""
nodes: set[Node] = config[ATTR_NODES]
dev_reg = dr.async_get(hass)
if not (nodes := async_get_nodes_from_targets(hass, config, dev_reg=dev_reg)):
raise ValueError(
f"No nodes found for given {ATTR_DEVICE_ID}s or {ATTR_ENTITY_ID}s."
)
from_value = config[ATTR_FROM]
to_value = config[ATTR_TO]
@ -163,7 +164,6 @@ async def async_attach_trigger(
hass.async_run_hass_job(job, {"trigger": payload})
dev_reg = dr.async_get(hass)
for node in nodes:
driver = node.client.driver
assert driver is not None # The node comes from the driver.

217
tests/components/zwave_js/test_trigger.py
View File

@ -269,6 +269,122 @@ async def test_zwave_js_value_updated(hass, client, lock_schlage_be469, integrat
await hass.services.async_call(automation.DOMAIN, SERVICE_RELOAD, blocking=True)
async def test_zwave_js_value_updated_bypass_dynamic_validation(
hass, client, lock_schlage_be469, integration
):
"""Test zwave_js.value_updated trigger when bypassing dynamic validation."""
trigger_type = f"{DOMAIN}.value_updated"
node: Node = lock_schlage_be469
no_value_filter = async_capture_events(hass, "no_value_filter")
with patch(
"homeassistant.components.zwave_js.triggers.value_updated.async_bypass_dynamic_config_validation",
return_value=True,
):
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
# no value filter
{
"trigger": {
"platform": trigger_type,
"entity_id": SCHLAGE_BE469_LOCK_ENTITY,
"command_class": CommandClass.DOOR_LOCK.value,
"property": "latchStatus",
},
"action": {
"event": "no_value_filter",
},
},
]
},
)
# Test that no value filter is triggered
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": node.node_id,
"args": {
"commandClassName": "Door Lock",
"commandClass": 98,
"endpoint": 0,
"property": "latchStatus",
"newValue": "boo",
"prevValue": "hiss",
"propertyName": "latchStatus",
},
},
)
node.receive_event(event)
await hass.async_block_till_done()
assert len(no_value_filter) == 1
async def test_zwave_js_value_updated_bypass_dynamic_validation_no_nodes(
hass, client, lock_schlage_be469, integration
):
"""Test value_updated trigger when bypassing dynamic validation with no nodes."""
trigger_type = f"{DOMAIN}.value_updated"
node: Node = lock_schlage_be469
no_value_filter = async_capture_events(hass, "no_value_filter")
with patch(
"homeassistant.components.zwave_js.triggers.value_updated.async_bypass_dynamic_config_validation",
return_value=True,
):
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
# no value filter
{
"trigger": {
"platform": trigger_type,
"entity_id": "sensor.test",
"command_class": CommandClass.DOOR_LOCK.value,
"property": "latchStatus",
},
"action": {
"event": "no_value_filter",
},
},
]
},
)
# Test that no value filter is NOT triggered because automation failed setup
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": node.node_id,
"args": {
"commandClassName": "Door Lock",
"commandClass": 98,
"endpoint": 0,
"property": "latchStatus",
"newValue": "boo",
"prevValue": "hiss",
"propertyName": "latchStatus",
},
},
)
node.receive_event(event)
await hass.async_block_till_done()
assert len(no_value_filter) == 0
async def test_zwave_js_event(hass, client, lock_schlage_be469, integration):
"""Test for zwave_js.event automation trigger."""
trigger_type = f"{DOMAIN}.event"
@ -644,6 +760,107 @@ async def test_zwave_js_event(hass, client, lock_schlage_be469, integration):
await hass.services.async_call(automation.DOMAIN, SERVICE_RELOAD, blocking=True)
async def test_zwave_js_event_bypass_dynamic_validation(
hass, client, lock_schlage_be469, integration
):
"""Test zwave_js.event trigger when bypassing dynamic config validation."""
trigger_type = f"{DOMAIN}.event"
node: Node = lock_schlage_be469
node_no_event_data_filter = async_capture_events(hass, "node_no_event_data_filter")
with patch(
"homeassistant.components.zwave_js.triggers.event.async_bypass_dynamic_config_validation",
return_value=True,
):
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
# node filter: no event data
{
"trigger": {
"platform": trigger_type,
"entity_id": SCHLAGE_BE469_LOCK_ENTITY,
"event_source": "node",
"event": "interview stage completed",
},
"action": {
"event": "node_no_event_data_filter",
},
},
]
},
)
# Test that `node no event data filter` is triggered and `node event data filter` is not
event = Event(
type="interview stage completed",
data={
"source": "node",
"event": "interview stage completed",
"stageName": "NodeInfo",
"nodeId": node.node_id,
},
)
node.receive_event(event)
await hass.async_block_till_done()
assert len(node_no_event_data_filter) == 1
async def test_zwave_js_event_bypass_dynamic_validation_no_nodes(
hass, client, lock_schlage_be469, integration
):
"""Test event trigger when bypassing dynamic validation with no nodes."""
trigger_type = f"{DOMAIN}.event"
node: Node = lock_schlage_be469
node_no_event_data_filter = async_capture_events(hass, "node_no_event_data_filter")
with patch(
"homeassistant.components.zwave_js.triggers.event.async_bypass_dynamic_config_validation",
return_value=True,
):
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
# node filter: no event data
{
"trigger": {
"platform": trigger_type,
"entity_id": "sensor.fake",
"event_source": "node",
"event": "interview stage completed",
},
"action": {
"event": "node_no_event_data_filter",
},
},
]
},
)
# Test that `node no event data filter` is NOT triggered because automation failed
# setup
event = Event(
type="interview stage completed",
data={
"source": "node",
"event": "interview stage completed",
"stageName": "NodeInfo",
"nodeId": node.node_id,
},
)
node.receive_event(event)
await hass.async_block_till_done()
assert len(node_no_event_data_filter) == 0
async def test_zwave_js_event_invalid_config_entry_id(
hass, client, integration, caplog
):