jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 17:27:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Initial support for zwave_js device conditions (#52003)

Browse Source
This commit is contained in:
Raman Gupta 2021-07-12 00:22:33 -04:00 committed by GitHub
parent 2ddaf746e6
commit e652ef51a1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 848 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
homeassistant/components/zwave_js/const.py
View File

@ -45,6 +45,9 @@ ATTR_EVENT_DATA = "event_data"
ATTR_DATA_TYPE = "data_type"
ATTR_WAIT_FOR_RESULT = "wait_for_result"
ATTR_NODE = "node"
ATTR_ZWAVE_VALUE = "zwave_value"
# service constants
ATTR_NODES = "nodes"

217
homeassistant/components/zwave_js/device_condition.py Normal file
View File

@ -0,0 +1,217 @@
"""Provide the device conditions for Z-Wave JS."""
from __future__ import annotations
from typing import cast
import voluptuous as vol
from zwave_js_server.const import CommandClass, ConfigurationValueType
from zwave_js_server.model.value import ConfigurationValue
from homeassistant.const import CONF_CONDITION, CONF_DEVICE_ID, CONF_DOMAIN, CONF_TYPE
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import condition, config_validation as cv
from homeassistant.helpers.config_validation import DEVICE_CONDITION_BASE_SCHEMA
from homeassistant.helpers.typing import ConfigType, TemplateVarsType
from . import DOMAIN
from .const import (
ATTR_COMMAND_CLASS,
ATTR_ENDPOINT,
ATTR_PROPERTY,
ATTR_PROPERTY_KEY,
ATTR_VALUE,
)
from .helpers import async_get_node_from_device_id, get_zwave_value_from_config
CONF_SUBTYPE = "subtype"
CONF_VALUE_ID = "value_id"
CONF_STATUS = "status"
NODE_STATUS_TYPE = "node_status"
NODE_STATUS_TYPES = ["asleep", "awake", "dead", "alive"]
CONFIG_PARAMETER_TYPE = "config_parameter"
VALUE_TYPE = "value"
CONDITION_TYPES = {NODE_STATUS_TYPE, CONFIG_PARAMETER_TYPE, VALUE_TYPE}
NODE_STATUS_CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): NODE_STATUS_TYPE,
vol.Required(CONF_STATUS): vol.In(NODE_STATUS_TYPES),
}
)
CONFIG_PARAMETER_CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): CONFIG_PARAMETER_TYPE,
vol.Required(CONF_VALUE_ID): cv.string,
vol.Required(CONF_SUBTYPE): cv.string,
vol.Optional(ATTR_VALUE): vol.Coerce(int),
}
)
VALUE_CONDITION_SCHEMA = DEVICE_CONDITION_BASE_SCHEMA.extend(
{
vol.Required(CONF_TYPE): VALUE_TYPE,
vol.Required(ATTR_COMMAND_CLASS): vol.In([cc.value for cc in CommandClass]),
vol.Required(ATTR_PROPERTY): vol.Any(vol.Coerce(int), cv.string),
vol.Optional(ATTR_PROPERTY_KEY): vol.Any(vol.Coerce(int), cv.string),
vol.Optional(ATTR_ENDPOINT): vol.Coerce(int),
vol.Required(ATTR_VALUE): vol.Any(
bool,
vol.Coerce(int),
vol.Coerce(float),
cv.boolean,
cv.string,
),
}
)
CONDITION_SCHEMA = vol.Any(
NODE_STATUS_CONDITION_SCHEMA,
CONFIG_PARAMETER_CONDITION_SCHEMA,
VALUE_CONDITION_SCHEMA,
)
async def async_validate_condition_config(
hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
config = CONDITION_SCHEMA(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)
return config
async def async_get_conditions(
hass: HomeAssistant, device_id: str
) -> list[dict[str, str]]:
"""List device conditions for Z-Wave JS devices."""
conditions = []
base_condition = {
CONF_CONDITION: "device",
CONF_DEVICE_ID: device_id,
CONF_DOMAIN: DOMAIN,
}
node = async_get_node_from_device_id(hass, device_id)
# Any value's value condition
conditions.append({**base_condition, CONF_TYPE: VALUE_TYPE})
# Node status conditions
conditions.append({**base_condition, CONF_TYPE: NODE_STATUS_TYPE})
# Config parameter conditions
conditions.extend(
[
{
**base_condition,
CONF_VALUE_ID: config_value.value_id,
CONF_TYPE: CONFIG_PARAMETER_TYPE,
CONF_SUBTYPE: f"{config_value.value_id} ({config_value.property_name})",
}
for config_value in node.get_configuration_values().values()
]
)
return conditions
@callback
def async_condition_from_config(
config: ConfigType, config_validation: bool
) -> condition.ConditionCheckerType:
"""Create a function to test a device condition."""
if config_validation:
config = CONDITION_SCHEMA(config)
condition_type = config[CONF_TYPE]
device_id = config[CONF_DEVICE_ID]
@callback
def test_node_status(hass: HomeAssistant, variables: TemplateVarsType) -> bool:
"""Test if node status is a certain state."""
node = async_get_node_from_device_id(hass, device_id)
return bool(node.status.name.lower() == config[CONF_STATUS])
if condition_type == NODE_STATUS_TYPE:
return test_node_status
@callback
def test_config_parameter(hass: HomeAssistant, variables: TemplateVarsType) -> bool:
"""Test if config parameter is a certain state."""
node = async_get_node_from_device_id(hass, device_id)
config_value = cast(ConfigurationValue, node.values[config[CONF_VALUE_ID]])
return bool(config_value.value == config[ATTR_VALUE])
if condition_type == CONFIG_PARAMETER_TYPE:
return test_config_parameter
@callback
def test_value(hass: HomeAssistant, variables: TemplateVarsType) -> bool:
"""Test if value is a certain state."""
node = async_get_node_from_device_id(hass, device_id)
value = get_zwave_value_from_config(node, config)
return bool(value.value == config[ATTR_VALUE])
if condition_type == VALUE_TYPE:
return test_value
raise HomeAssistantError(f"Unhandled condition type {condition_type}")
@callback
async def async_get_condition_capabilities(
hass: HomeAssistant, config: ConfigType
) -> dict[str, vol.Schema]:
"""List condition capabilities."""
device_id = config[CONF_DEVICE_ID]
node = async_get_node_from_device_id(hass, device_id)
# 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:
return {}
return {"extra_fields": vol.Schema({vol.Required(ATTR_VALUE): value_schema})}
if config[CONF_TYPE] == VALUE_TYPE:
return {
"extra_fields": vol.Schema(
{
vol.Required(ATTR_COMMAND_CLASS): vol.In(
{cc.value: cc.name for cc in CommandClass}
),
vol.Required(ATTR_PROPERTY): cv.string,
vol.Optional(ATTR_PROPERTY_KEY): cv.string,
vol.Optional(ATTR_ENDPOINT): cv.string,
vol.Required(ATTR_VALUE): cv.string,
}
)
}
if config[CONF_TYPE] == NODE_STATUS_TYPE:
return {
"extra_fields": vol.Schema(
{vol.Required(CONF_STATUS): vol.In(NODE_STATUS_TYPES)}
)
}
return {}

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

@ -3,9 +3,10 @@ from __future__ import annotations
from typing import Any, cast
import voluptuous as vol
from zwave_js_server.client import Client as ZwaveClient
from zwave_js_server.model.node import Node as ZwaveNode
from zwave_js_server.model.value import Value as ZwaveValue
from zwave_js_server.model.value import Value as ZwaveValue, get_value_id
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import __version__ as HA_VERSION
@ -18,8 +19,17 @@ from homeassistant.helpers.entity_registry import (
EntityRegistry,
async_get as async_get_ent_reg,
)
from homeassistant.helpers.typing import ConfigType
from .const import CONF_DATA_COLLECTION_OPTED_IN, DATA_CLIENT, DOMAIN
from .const import (
ATTR_COMMAND_CLASS,
ATTR_ENDPOINT,
ATTR_PROPERTY,
ATTR_PROPERTY_KEY,
CONF_DATA_COLLECTION_OPTED_IN,
DATA_CLIENT,
DOMAIN,
)
@callback
@ -143,3 +153,23 @@ def async_get_node_from_entity_id(
# tied to a device
assert entity_entry.device_id
return async_get_node_from_device_id(hass, entity_entry.device_id, dev_reg)
def get_zwave_value_from_config(node: ZwaveNode, config: ConfigType) -> ZwaveValue:
"""Get a Z-Wave JS Value from a config."""
endpoint = None
if config.get(ATTR_ENDPOINT):
endpoint = config[ATTR_ENDPOINT]
property_key = None
if config.get(ATTR_PROPERTY_KEY):
property_key = config[ATTR_PROPERTY_KEY]
value_id = get_value_id(
node,
config[ATTR_COMMAND_CLASS],
config[ATTR_PROPERTY],
endpoint,
property_key,
)
if value_id not in node.values:
raise vol.Invalid(f"Value {value_id} can't be found on node {node}")
return node.values[value_id]

17
homeassistant/components/zwave_js/strings.json
View File

@ -10,7 +10,9 @@
"on_supervisor": {
"title": "Select connection method",
"description": "Do you want to use the Z-Wave JS Supervisor add-on?",
"data": { "use_addon": "Use the Z-Wave JS Supervisor add-on" }
"data": {
"use_addon": "Use the Z-Wave JS Supervisor add-on"
}
},
"install_addon": {
"title": "The Z-Wave JS add-on installation has started"
@ -22,7 +24,9 @@
"network_key": "Network Key"
}
},
"start_addon": { "title": "The Z-Wave JS add-on is starting." },
"start_addon": {
"title": "The Z-Wave JS add-on is starting."
},
"hassio_confirm": {
"title": "Set up Z-Wave JS integration with the Z-Wave JS add-on"
}
@ -93,5 +97,12 @@
"install_addon": "Please wait while the Z-Wave JS add-on installation finishes. This can take several minutes.",
"start_addon": "Please wait while the Z-Wave JS add-on start completes. This may take some seconds."
}
},
"device_automation": {
"condition_type": {
"node_status": "Node status",
"config_parameter": "Config parameter {subtype} value",
"value": "Current value of a Z-Wave Value"
}
}
}
}

10
homeassistant/components/zwave_js/translations/en.json
View File

@ -101,5 +101,15 @@
}
}
},
"device_automation": {
"condition_type": {
"alive": "Node is alive",
"asleep": "Node is asleep",
"awake": "Node is awake",
"config_parameter": "Config parameter {subtype} value",
"dead": "Node is dead",
"value": "Current value of a Z-Wave Value"
}
},
"title": "Z-Wave JS"
}

572
tests/components/zwave_js/test_device_condition.py Normal file
View File

@ -0,0 +1,572 @@
"""The tests for Z-Wave JS device conditions."""
from __future__ import annotations
from unittest.mock import patch
import pytest
import voluptuous as vol
import voluptuous_serialize
from zwave_js_server.const import CommandClass
from zwave_js_server.event import Event
from homeassistant.components import automation
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
from homeassistant.helpers import config_validation as cv, device_registry
from homeassistant.setup import async_setup_component
from tests.common import async_get_device_automations, async_mock_service
@pytest.fixture
def calls(hass):
"""Track calls to a mock service."""
return async_mock_service(hass, "test", "automation")
async def test_get_conditions(hass, client, lock_schlage_be469, integration) -> None:
"""Test we get the expected onditions from a zwave_js."""
dev_reg = device_registry.async_get(hass)
device = device_registry.async_entries_for_config_entry(
dev_reg, integration.entry_id
)[0]
config_value = list(lock_schlage_be469.get_configuration_values().values())[0]
value_id = config_value.value_id
name = config_value.property_name
expected_conditions = [
{
"condition": "device",
"domain": DOMAIN,
"type": "node_status",
"device_id": device.id,
},
{
"condition": "device",
"domain": DOMAIN,
"type": "config_parameter",
"device_id": device.id,
"value_id": value_id,
"subtype": f"{value_id} ({name})",
},
{
"condition": "device",
"domain": DOMAIN,
"type": "value",
"device_id": device.id,
},
]
conditions = await async_get_device_automations(hass, "condition", device.id)
for condition in expected_conditions:
assert condition in conditions
async def test_node_status_state(
hass, client, lock_schlage_be469, integration, calls
) -> None:
"""Test for node_status conditions."""
dev_reg = device_registry.async_get(hass)
device = device_registry.async_entries_for_config_entry(
dev_reg, integration.entry_id
)[0]
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {"platform": "event", "event_type": "test_event1"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "node_status",
"status": "alive",
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "alive - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
{
"trigger": {"platform": "event", "event_type": "test_event2"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "node_status",
"status": "awake",
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "awake - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
{
"trigger": {"platform": "event", "event_type": "test_event3"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "node_status",
"status": "asleep",
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "asleep - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
{
"trigger": {"platform": "event", "event_type": "test_event4"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "node_status",
"status": "dead",
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "dead - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
]
},
)
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3")
hass.bus.async_fire("test_event4")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "alive - event - test_event1"
event = Event(
"wake up",
data={
"source": "node",
"event": "wake up",
"nodeId": lock_schlage_be469.node_id,
},
)
lock_schlage_be469.receive_event(event)
await hass.async_block_till_done()
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3")
hass.bus.async_fire("test_event4")
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "awake - event - test_event2"
event = Event(
"sleep",
data={"source": "node", "event": "sleep", "nodeId": lock_schlage_be469.node_id},
)
lock_schlage_be469.receive_event(event)
await hass.async_block_till_done()
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3")
hass.bus.async_fire("test_event4")
await hass.async_block_till_done()
assert len(calls) == 3
assert calls[2].data["some"] == "asleep - event - test_event3"
event = Event(
"dead",
data={"source": "node", "event": "dead", "nodeId": lock_schlage_be469.node_id},
)
lock_schlage_be469.receive_event(event)
await hass.async_block_till_done()
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
hass.bus.async_fire("test_event3")
hass.bus.async_fire("test_event4")
await hass.async_block_till_done()
assert len(calls) == 4
assert calls[3].data["some"] == "dead - event - test_event4"
event = Event(
"unknown",
data={
"source": "node",
"event": "unknown",
"nodeId": lock_schlage_be469.node_id,
},
)
lock_schlage_be469.receive_event(event)
await hass.async_block_till_done()
async def test_config_parameter_state(
hass, client, lock_schlage_be469, integration, calls
) -> None:
"""Test for config_parameter conditions."""
dev_reg = device_registry.async_get(hass)
device = device_registry.async_entries_for_config_entry(
dev_reg, integration.entry_id
)[0]
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {"platform": "event", "event_type": "test_event1"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "config_parameter",
"value_id": f"{lock_schlage_be469.node_id}-112-0-3",
"subtype": f"{lock_schlage_be469.node_id}-112-0-3 (Beeper)",
"value": 255,
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "Beeper - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
{
"trigger": {"platform": "event", "event_type": "test_event2"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "config_parameter",
"value_id": f"{lock_schlage_be469.node_id}-112-0-6",
"subtype": f"{lock_schlage_be469.node_id}-112-0-6 (User Slot Status)",
"value": 1,
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "User Slot Status - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
]
},
)
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "Beeper - event - test_event1"
# Flip Beeper state to not match condition
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": lock_schlage_be469.node_id,
"args": {
"commandClassName": "Configuration",
"commandClass": 112,
"endpoint": 0,
"property": 3,
"newValue": 0,
"prevValue": 255,
},
},
)
lock_schlage_be469.receive_event(event)
# Flip User Slot Status to match condition
event = Event(
type="value updated",
data={
"source": "node",
"event": "value updated",
"nodeId": lock_schlage_be469.node_id,
"args": {
"commandClassName": "Configuration",
"commandClass": 112,
"endpoint": 0,
"property": 6,
"newValue": 1,
"prevValue": 117440512,
},
},
)
lock_schlage_be469.receive_event(event)
hass.bus.async_fire("test_event1")
hass.bus.async_fire("test_event2")
await hass.async_block_till_done()
assert len(calls) == 2
assert calls[1].data["some"] == "User Slot Status - event - test_event2"
async def test_value_state(
hass, client, lock_schlage_be469, integration, calls
) -> None:
"""Test for value conditions."""
dev_reg = device_registry.async_get(hass)
device = device_registry.async_entries_for_config_entry(
dev_reg, integration.entry_id
)[0]
assert await async_setup_component(
hass,
automation.DOMAIN,
{
automation.DOMAIN: [
{
"trigger": {"platform": "event", "event_type": "test_event1"},
"condition": [
{
"condition": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "value",
"command_class": 112,
"property": 3,
"value": 255,
}
],
"action": {
"service": "test.automation",
"data_template": {
"some": "value - {{ trigger.platform }} - {{ trigger.event.event_type }}"
},
},
},
]
},
)
hass.bus.async_fire("test_event1")
await hass.async_block_till_done()
assert len(calls) == 1
assert calls[0].data["some"] == "value - event - test_event1"
async def test_get_condition_capabilities_node_status(
hass, client, lock_schlage_be469, integration
):
"""Test we don't get capabilities from a node_status condition."""
dev_reg = device_registry.async_get(hass)
device = device_registry.async_entries_for_config_entry(
dev_reg, integration.entry_id
)[0]
capabilities = await device_condition.async_get_condition_capabilities(
hass,
{
"platform": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "node_status",
},
)
assert capabilities and "extra_fields" in capabilities
assert voluptuous_serialize.convert(
capabilities["extra_fields"], custom_serializer=cv.custom_serializer
) == [
{
"name": "status",
"required": True,
"type": "select",
"options": [
("asleep", "asleep"),
("awake", "awake"),
("dead", "dead"),
("alive", "alive"),
],
}
]
async def test_get_condition_capabilities_value(
hass, client, lock_schlage_be469, integration
):
"""Test we get the expected capabilities from a value condition."""
dev_reg = device_registry.async_get(hass)
device = device_registry.async_entries_for_config_entry(
dev_reg, integration.entry_id
)[0]
capabilities = await device_condition.async_get_condition_capabilities(
hass,
{
"platform": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "value",
},
)
assert capabilities and "extra_fields" in capabilities
cc_options = [(cc.value, cc.name) for cc in CommandClass]
assert voluptuous_serialize.convert(
capabilities["extra_fields"], custom_serializer=cv.custom_serializer
) == [
{
"name": "command_class",
"required": True,
"options": cc_options,
"type": "select",
},
{"name": "property", "required": True, "type": "string"},
{"name": "property_key", "optional": True, "type": "string"},
{"name": "endpoint", "optional": True, "type": "string"},
{"name": "value", "required": True, "type": "string"},
]
async def test_get_condition_capabilities_config_parameter(
hass, client, climate_radio_thermostat_ct100_plus, integration
):
"""Test we get the expected capabilities from a config_parameter condition."""
node = climate_radio_thermostat_ct100_plus
dev_reg = device_registry.async_get(hass)
device = device_registry.async_entries_for_config_entry(
dev_reg, integration.entry_id
)[0]
# Test enumerated type param
capabilities = await device_condition.async_get_condition_capabilities(
hass,
{
"platform": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "config_parameter",
"value_id": f"{node.node_id}-112-0-1",
"subtype": f"{node.node_id}-112-0-1 (Temperature Reporting Threshold)",
},
)
assert capabilities and "extra_fields" in capabilities
assert voluptuous_serialize.convert(
capabilities["extra_fields"], custom_serializer=cv.custom_serializer
) == [
{
"name": "value",
"required": True,
"options": [
(0, "Disabled"),
(1, "0.5° F"),
(2, "1.0° F"),
(3, "1.5° F"),
(4, "2.0° F"),
],
"type": "select",
}
]
# Test range type param
capabilities = await device_condition.async_get_condition_capabilities(
hass,
{
"platform": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "config_parameter",
"value_id": f"{node.node_id}-112-0-10",
"subtype": f"{node.node_id}-112-0-10 (Temperature Reporting Filter)",
},
)
assert capabilities and "extra_fields" in capabilities
assert voluptuous_serialize.convert(
capabilities["extra_fields"], custom_serializer=cv.custom_serializer
) == [
{
"name": "value",
"required": True,
"valueMin": 0,
"valueMax": 124,
}
]
# Test undefined type param
capabilities = await device_condition.async_get_condition_capabilities(
hass,
{
"platform": "device",
"domain": DOMAIN,
"device_id": device.id,
"type": "config_parameter",
"value_id": f"{node.node_id}-112-0-2",
"subtype": f"{node.node_id}-112-0-2 (HVAC Settings)",
},
)
assert not capabilities
async def test_failure_scenarios(hass, client, hank_binary_switch, integration):
"""Test failure scenarios."""
dev_reg = device_registry.async_get(hass)
device = device_registry.async_entries_for_config_entry(
dev_reg, integration.entry_id
)[0]
with pytest.raises(HomeAssistantError):
await device_condition.async_condition_from_config(
{"type": "failed.test", "device_id": device.id}, False
)
with patch(
"homeassistant.components.zwave_js.device_condition.async_get_node_from_device_id",
return_value=None,
), patch(
"homeassistant.components.zwave_js.device_condition.get_zwave_value_from_config",
return_value=None,
):
assert (
await device_condition.async_get_condition_capabilities(
hass, {"type": "failed.test", "device_id": device.id}
)
== {}
)
async def test_get_value_from_config_failure(
hass, client, hank_binary_switch, integration
):
"""Test get_value_from_config invalid value ID."""
with pytest.raises(vol.Invalid):
get_zwave_value_from_config(
hank_binary_switch,
{
"command_class": CommandClass.SCENE_ACTIVATION.value,
"property": "sceneId",
"property_key": 15,
"endpoint": 10,
},
)