evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bypass dynamic validation for zwave_js custom triggers (#72471)

Browse source
This commit is contained in:
Raman Gupta 2022-05-25 13:20:40 -04:00 • committed by GitHub
parent 9b40de18cd
commit c8c4bf6c37
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 144 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

11
homeassistant/components/zwave_js/triggers/event.py
View file

@ -30,12 +30,13 @@ from homeassistant.components.zwave_js.helpers import (
get_device_id,
get_home_and_node_id_from_device_entry,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_DEVICE_ID, ATTR_ENTITY_ID, CONF_PLATFORM
from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.typing import ConfigType
from .helpers import async_bypass_dynamic_config_validation
# Platform type should be <DOMAIN>.<SUBMODULE_NAME>
PLATFORM_TYPE = f"{DOMAIN}.{__name__.rsplit('.', maxsplit=1)[-1]}"
@ -115,6 +116,9 @@ 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]:
@ -126,12 +130,9 @@ async def async_validate_trigger_config(
return config
entry_id = config[ATTR_CONFIG_ENTRY_ID]
if (entry := hass.config_entries.async_get_entry(entry_id)) is None:
if hass.config_entries.async_get_entry(entry_id) is None:
raise vol.Invalid(f"Config entry '{entry_id}' not found")
if entry.state is not ConfigEntryState.LOADED:
raise vol.Invalid(f"Config entry '{entry_id}' not loaded")
return config

35
homeassistant/components/zwave_js/triggers/helpers.py Normal file
View file

@ -0,0 +1,35 @@
"""Helpers for Z-Wave JS custom triggers."""
from homeassistant.components.zwave_js.const import ATTR_CONFIG_ENTRY_ID, DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_DEVICE_ID, ATTR_ENTITY_ID
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.typing import ConfigType
@callback
def async_bypass_dynamic_config_validation(
hass: HomeAssistant, config: ConfigType
) -> bool:
"""Return whether target zwave_js config entry is not loaded."""
# If the config entry is not loaded for a zwave_js device, entity, or the
# config entry ID provided, we can't perform dynamic validation
dev_reg = dr.async_get(hass)
ent_reg = er.async_get(hass)
trigger_devices = config.get(ATTR_DEVICE_ID, [])
trigger_entities = config.get(ATTR_ENTITY_ID, [])
return any(
entry.state != ConfigEntryState.LOADED
and (
entry.entry_id == config.get(ATTR_CONFIG_ENTRY_ID)
or any(
device.id in trigger_devices
for device in dr.async_entries_for_config_entry(dev_reg, entry.entry_id)
)
or (
entity.entity_id in trigger_entities
for entity in er.async_entries_for_config_entry(ent_reg, entry.entry_id)
)
)
for entry in hass.config_entries.async_entries(DOMAIN)
)

6
homeassistant/components/zwave_js/triggers/value_updated.py
View file

@ -12,6 +12,7 @@ from homeassistant.components.automation import (
AutomationActionType,
AutomationTriggerInfo,
)
from homeassistant.components.zwave_js.config_validation import VALUE_SCHEMA
from homeassistant.components.zwave_js.const import (
ATTR_COMMAND_CLASS,
ATTR_COMMAND_CLASS_NAME,
@ -37,7 +38,7 @@ from homeassistant.core import CALLBACK_TYPE, HassJob, HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.typing import ConfigType
from ..config_validation import VALUE_SCHEMA
from .helpers import async_bypass_dynamic_config_validation
# Platform type should be <DOMAIN>.<SUBMODULE_NAME>
PLATFORM_TYPE = f"{DOMAIN}.{__name__.rsplit('.', maxsplit=1)[-1]}"
@ -75,6 +76,9 @@ async def async_validate_trigger_config(
"""Validate config."""
config = TRIGGER_SCHEMA(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]:
raise vol.Invalid(

127
tests/components/zwave_js/test_trigger.py
View file

@ -10,6 +10,9 @@ from zwave_js_server.model.node import Node
from homeassistant.components import automation
from homeassistant.components.zwave_js import DOMAIN
from homeassistant.components.zwave_js.trigger import async_validate_trigger_config
from homeassistant.components.zwave_js.triggers.helpers import (
async_bypass_dynamic_config_validation,
)
from homeassistant.const import SERVICE_RELOAD
from homeassistant.helpers.device_registry import (
async_entries_for_config_entry,
@ -671,35 +674,6 @@ async def test_zwave_js_event_invalid_config_entry_id(
caplog.clear()
async def test_zwave_js_event_unloaded_config_entry(hass, client, integration, caplog):
"""Test zwave_js.event automation trigger fails when config entry is unloaded."""
trigger_type = f"{DOMAIN}.event"
await hass.config_entries.async_unload(integration.entry_id)
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {
"platform": trigger_type,
"config_entry_id": integration.entry_id,
"event_source": "controller",
"event": "inclusion started",
},
"action": {
"event": "node_no_event_data_filter",
},
}
]
},
)
assert f"Config entry '{integration.entry_id}' not loaded" in caplog.text
async def test_async_validate_trigger_config(hass):
"""Test async_validate_trigger_config."""
mock_platform = AsyncMock()
@ -735,3 +709,98 @@ async def test_invalid_trigger_configs(hass):
"property": "latchStatus",
},
)
async def test_zwave_js_trigger_config_entry_unloaded(
hass, client, lock_schlage_be469, integration
):
"""Test zwave_js triggers bypass dynamic validation when needed."""
dev_reg = async_get_dev_reg(hass)
device = async_entries_for_config_entry(dev_reg, integration.entry_id)[0]
# Test bypass check is False
assert not async_bypass_dynamic_config_validation(
hass,
{
"platform": f"{DOMAIN}.value_updated",
"entity_id": SCHLAGE_BE469_LOCK_ENTITY,
"command_class": CommandClass.DOOR_LOCK.value,
"property": "latchStatus",
},
)
await hass.config_entries.async_unload(integration.entry_id)
# Test full validation for both events
assert await async_validate_trigger_config(
hass,
{
"platform": f"{DOMAIN}.value_updated",
"entity_id": SCHLAGE_BE469_LOCK_ENTITY,
"command_class": CommandClass.DOOR_LOCK.value,
"property": "latchStatus",
},
)
assert await async_validate_trigger_config(
hass,
{
"platform": f"{DOMAIN}.event",
"entity_id": SCHLAGE_BE469_LOCK_ENTITY,
"event_source": "node",
"event": "interview stage completed",
},
)
# Test bypass check
assert async_bypass_dynamic_config_validation(
hass,
{
"platform": f"{DOMAIN}.value_updated",
"entity_id": SCHLAGE_BE469_LOCK_ENTITY,
"command_class": CommandClass.DOOR_LOCK.value,
"property": "latchStatus",
},
)
assert async_bypass_dynamic_config_validation(
hass,
{
"platform": f"{DOMAIN}.value_updated",
"device_id": device.id,
"command_class": CommandClass.DOOR_LOCK.value,
"property": "latchStatus",
"from": "ajar",
},
)
assert async_bypass_dynamic_config_validation(
hass,
{
"platform": f"{DOMAIN}.event",
"entity_id": SCHLAGE_BE469_LOCK_ENTITY,
"event_source": "node",
"event": "interview stage completed",
},
)
assert async_bypass_dynamic_config_validation(
hass,
{
"platform": f"{DOMAIN}.event",
"device_id": device.id,
"event_source": "node",
"event": "interview stage completed",
"event_data": {"stageName": "ProtocolInfo"},
},
)
assert async_bypass_dynamic_config_validation(
hass,
{
"platform": f"{DOMAIN}.event",
"config_entry_id": integration.entry_id,
"event_source": "controller",
"event": "nvm convert progress",
},
)