starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 09:23:17 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add Z-Wave JS node status, configuration parameter, and value conditions (#181133)

Browse Source 
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Raman Gupta
2026-09-10 11:26:45 +02:00
committed by GitHub
co-authored by Claude Fable 5.1
parent 0092cb2dca
commit 805abfbb46
15 changed files with 1414 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/zwave_js/condition.py
+273
View File
@@ -0,0 +1,273 @@
"""Offer Z-Wave JS automation conditions."""
import abc
from collections.abc import Callable, Iterable
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Unpack, override
import voluptuous as vol
from zwave_js_server.const import CommandClass
from zwave_js_server.model.node import Node as ZwaveNode
from homeassistant.const import ATTR_DEVICE_ID, CONF_OPTIONS
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv, device_registry as dr
from homeassistant.helpers.automation import move_top_level_schema_fields_to_options
from homeassistant.helpers.condition import (
ATTR_BEHAVIOR,
BEHAVIOR_ALL,
BEHAVIOR_ANY,
Condition,
ConditionCheckParams,
ConditionConfig,
)
from homeassistant.helpers.typing import ConfigType
from .config_validation import BITMASK_SCHEMA, COMMAND_CLASS_SCHEMA
from .const import (
ATTR_COMMAND_CLASS,
ATTR_CONFIG_PARAMETER,
ATTR_CONFIG_PARAMETER_BITMASK,
ATTR_ENDPOINT,
ATTR_PROPERTY,
ATTR_PROPERTY_KEY,
ATTR_VALUE,
NODE_STATUSES,
)
from .helpers import (
async_bypass_dynamic_config_validation,
async_get_node_from_device_id,
get_zwave_value_from_config,
node_status_matches,
value_matches_state,
)
CONF_STATUS = "status"
# Conditions compare against state labels, so strings must be kept as given
_CONDITION_VALUE_SCHEMA = vol.Any(bool, int, float, dict, cv.string)
_BASE_SCHEMA_DICT: dict[vol.Marker, Any] = {
vol.Required(ATTR_DEVICE_ID): vol.All(cv.ensure_list, [cv.string]),
vol.Required(ATTR_BEHAVIOR, default=BEHAVIOR_ANY): vol.In(
[BEHAVIOR_ANY, BEHAVIOR_ALL]
),
}
_NODE_STATUS_OPTIONS_SCHEMA_DICT: dict[vol.Marker, Any] = {
**_BASE_SCHEMA_DICT,
vol.Required(CONF_STATUS): vol.In(NODE_STATUSES),
}
_VALUE_OPTIONS_SCHEMA_DICT: dict[vol.Marker, Any] = {
**_BASE_SCHEMA_DICT,
vol.Required(ATTR_COMMAND_CLASS): COMMAND_CLASS_SCHEMA,
vol.Required(ATTR_PROPERTY): vol.Any(vol.Coerce(int), cv.string),
vol.Optional(ATTR_ENDPOINT): vol.Coerce(int),
vol.Optional(ATTR_PROPERTY_KEY): vol.Any(vol.Coerce(int), cv.string),
vol.Required(ATTR_VALUE): _CONDITION_VALUE_SCHEMA,
}
_CONFIG_PARAMETER_OPTIONS_SCHEMA_DICT: dict[vol.Marker, Any] = {
**_BASE_SCHEMA_DICT,
vol.Required(ATTR_CONFIG_PARAMETER): vol.Coerce(int),
vol.Optional(ATTR_CONFIG_PARAMETER_BITMASK): vol.Any(
vol.Coerce(int), BITMASK_SCHEMA
),
vol.Optional(ATTR_ENDPOINT, default=0): vol.Coerce(int),
vol.Required(ATTR_VALUE): _CONDITION_VALUE_SCHEMA,
}
def _condition_schema(options_schema_dict: dict[vol.Marker, Any]) -> vol.Schema:
"""Return the condition schema for an options schema dict."""
return vol.Schema({vol.Required(CONF_OPTIONS, default={}): options_schema_dict})
@dataclass(slots=True)
class _ResolvedNodes:
"""Z-Wave nodes resolved from the targeted devices."""
nodes: set[ZwaveNode] = field(default_factory=set)
unresolved: int = 0
@callback
def _async_resolve_nodes(
hass: HomeAssistant, device_ids: Iterable[str]
) -> _ResolvedNodes:
"""Resolve targeted device IDs to Z-Wave nodes."""
dev_reg = dr.async_get(hass)
resolved = _ResolvedNodes()
for device_id in set(device_ids):
try:
node = async_get_node_from_device_id(hass, device_id, dev_reg)
except ValueError:
resolved.unresolved += 1
else:
resolved.nodes.add(node)
return resolved
class _ZwaveNodeCondition(Condition):
"""Base for conditions evaluated per Z-Wave node."""
options_schema_dict: dict[vol.Marker, Any]
_schema: vol.Schema
@classmethod
@override
async def async_validate_complete_config(
cls, hass: HomeAssistant, complete_config: ConfigType
) -> ConfigType:
"""Validate complete config."""
complete_config = move_top_level_schema_fields_to_options(
complete_config, cls.options_schema_dict
)
return await super().async_validate_complete_config(hass, complete_config)
@classmethod
@override
async def async_validate_config(
cls, hass: HomeAssistant, config: ConfigType
) -> ConfigType:
"""Validate config."""
config = cls._schema(config)
device_ids = config[CONF_OPTIONS][ATTR_DEVICE_ID]
if async_bypass_dynamic_config_validation(hass, {ATTR_DEVICE_ID: device_ids}):
return config
resolved = _async_resolve_nodes(hass, device_ids)
if not resolved.nodes:
raise vol.Invalid("No nodes found for the given devices")
cls._validate_nodes(resolved.nodes, config[CONF_OPTIONS])
return config
@classmethod
def _validate_nodes(cls, nodes: set[ZwaveNode], options: dict[str, Any]) -> None:
"""Validate the options against the resolved nodes."""
def __init__(self, hass: HomeAssistant, config: ConditionConfig) -> None:
"""Initialize condition."""
super().__init__(hass, config)
if TYPE_CHECKING:
assert config.options is not None
self._options = config.options
@abc.abstractmethod
def _node_matches(self, node: ZwaveNode) -> bool:
"""Return whether a node satisfies the condition."""
@override
def _async_check(self, **kwargs: Unpack[ConditionCheckParams]) -> bool:
"""Test the condition against all targeted nodes."""
resolved = _async_resolve_nodes(self._hass, self._options[ATTR_DEVICE_ID])
if not resolved.nodes:
return False
behavior_all = self._options[ATTR_BEHAVIOR] == BEHAVIOR_ALL
if behavior_all and resolved.unresolved:
return False
combine: Callable[[Iterable[object]], bool] = all if behavior_all else any
return combine(self._node_matches(node) for node in resolved.nodes)
class NodeStatusCondition(_ZwaveNodeCondition):
"""Test the status of Z-Wave nodes."""
options_schema_dict = _NODE_STATUS_OPTIONS_SCHEMA_DICT
_schema = _condition_schema(_NODE_STATUS_OPTIONS_SCHEMA_DICT)
@override
def _node_matches(self, node: ZwaveNode) -> bool:
return node_status_matches(node, self._options[CONF_STATUS])
class _ZwaveValueCondition(_ZwaveNodeCondition):
"""Base for conditions comparing a Z-Wave value."""
@classmethod
@abc.abstractmethod
def _value_config(cls, options: dict[str, Any]) -> dict[str, Any]:
"""Return the value lookup config for get_zwave_value_from_config."""
@classmethod
@abc.abstractmethod
def _value_description(cls, options: dict[str, Any]) -> str:
"""Return a human readable description of the looked up value."""
@classmethod
@override
def _validate_nodes(cls, nodes: set[ZwaveNode], options: dict[str, Any]) -> None:
value_config = cls._value_config(options)
for node in nodes:
try:
get_zwave_value_from_config(node, value_config)
except vol.Invalid:
continue
return
raise vol.Invalid(f"No targeted node has {cls._value_description(options)}")
@override
def _node_matches(self, node: ZwaveNode) -> bool:
try:
value = get_zwave_value_from_config(node, self._value_config(self._options))
except vol.Invalid:
return False
return value_matches_state(value, self._options[ATTR_VALUE])
class ValueCondition(_ZwaveValueCondition):
"""Test a Z-Wave value."""
options_schema_dict = _VALUE_OPTIONS_SCHEMA_DICT
_schema = _condition_schema(_VALUE_OPTIONS_SCHEMA_DICT)
@classmethod
@override
def _value_config(cls, options: dict[str, Any]) -> dict[str, Any]:
return {
ATTR_COMMAND_CLASS: options[ATTR_COMMAND_CLASS],
ATTR_PROPERTY: options[ATTR_PROPERTY],
ATTR_ENDPOINT: options.get(ATTR_ENDPOINT),
ATTR_PROPERTY_KEY: options.get(ATTR_PROPERTY_KEY),
}
@classmethod
@override
def _value_description(cls, options: dict[str, Any]) -> str:
command_class = CommandClass(options[ATTR_COMMAND_CLASS])
return f"value {command_class.name}-{options[ATTR_PROPERTY]}"
class ConfigParameterCondition(_ZwaveValueCondition):
"""Test a Z-Wave configuration parameter."""
options_schema_dict = _CONFIG_PARAMETER_OPTIONS_SCHEMA_DICT
_schema = _condition_schema(_CONFIG_PARAMETER_OPTIONS_SCHEMA_DICT)
@classmethod
@override
def _value_config(cls, options: dict[str, Any]) -> dict[str, Any]:
return {
ATTR_COMMAND_CLASS: CommandClass.CONFIGURATION,
ATTR_PROPERTY: options[ATTR_CONFIG_PARAMETER],
ATTR_PROPERTY_KEY: options.get(ATTR_CONFIG_PARAMETER_BITMASK),
ATTR_ENDPOINT: options[ATTR_ENDPOINT],
}
@classmethod
@override
def _value_description(cls, options: dict[str, Any]) -> str:
return f"configuration parameter {options[ATTR_CONFIG_PARAMETER]}"
CONDITIONS: dict[str, type[Condition]] = {
"node_status": NodeStatusCondition,
"config_parameter": ConfigParameterCondition,
"value": ValueCondition,
}
async def async_get_conditions(hass: HomeAssistant) -> dict[str, type[Condition]]:
"""Return the Z-Wave JS conditions."""
return CONDITIONS
homeassistant/components/zwave_js/conditions.yaml
+213
View File
@@ -0,0 +1,213 @@
# Describes the format for available Z-Wave JS conditions
.device_id: &device_id
required: true
example: 8f4219cfa57e23f6f669c4616c2205e2
selector:
device:
filter:
- integration: zwave_js
multiple: true
.behavior: &behavior
required: true
default: any
selector:
automation_behavior:
mode: condition
.value: &value
required: true
example: 255
selector:
object:
node_status:
fields:
device_id: *device_id
behavior: *behavior
status:
required: true
selector:
select:
translation_key: node_status
options:
- alive
- asleep
- awake
- dead
config_parameter:
fields:
device_id: *device_id
behavior: *behavior
parameter:
required: true
example: 3
selector:
number:
min: 0
mode: box
bitmask:
required: false
example: "0x1"
selector:
text:
endpoint:
required: false
default: 0
selector:
number:
min: 0
mode: box
value: *value
value:
fields:
device_id: *device_id
behavior: *behavior
command_class:
required: true
selector:
select:
translation_key: command_class
sort: true
options:
- "0"
- "32"
- "33"
- "34"
- "35"
- "37"
- "38"
- "39"
- "40"
- "41"
- "43"
- "44"
- "45"
- "48"
- "49"
- "50"
- "51"
- "52"
- "53"
- "54"
- "55"
- "57"
- "58"
- "59"
- "60"
- "61"
- "62"
- "63"
- "64"
- "65"
- "66"
- "67"
- "68"
- "69"
- "70"
- "71"
- "72"
- "73"
- "74"
- "75"
- "76"
- "77"
- "78"
- "79"
- "80"
- "81"
- "82"
- "83"
- "84"
- "85"
- "86"
- "87"
- "88"
- "89"
- "90"
- "91"
- "92"
- "93"
- "94"
- "95"
- "96"
- "97"
- "98"
- "99"
- "100"
- "102"
- "103"
- "104"
- "105"
- "106"
- "107"
- "108"
- "109"
- "110"
- "111"
- "112"
- "113"
- "114"
- "115"
- "116"
- "117"
- "118"
- "119"
- "120"
- "121"
- "122"
- "123"
- "124"
- "125"
- "126"
- "128"
- "129"
- "130"
- "132"
- "133"
- "134"
- "135"
- "136"
- "137"
- "138"
- "139"
- "140"
- "142"
- "143"
- "144"
- "145"
- "146"
- "147"
- "148"
- "152"
- "154"
- "155"
- "156"
- "157"
- "158"
- "159"
- "160"
- "161"
- "162"
- "163"
property:
required: true
example: currentValue
selector:
text:
endpoint:
required: false
example: 1
selector:
number:
min: 0
mode: box
property_key:
required: false
example: 1
selector:
text:
value: *value
homeassistant/components/zwave_js/const.py
+2
View File
@@ -181,6 +181,8 @@ ATTR_TWIST_ASSIST = "twist_assist"
ADDON_SLUG = "core_zwave_js"
NODE_STATUSES = ["asleep", "awake", "dead", "alive"]
# Sensor entity description constants
ENTITY_DESC_KEY_BATTERY_LIST_STATE = "battery_list_state"
ENTITY_DESC_KEY_BATTERY_MAXIMUM_CAPACITY = "battery_maximum_capacity"
homeassistant/components/zwave_js/device_automation_helpers.py
-2
View File
@@ -8,8 +8,6 @@ from homeassistant.helpers import device_registry as dr
from .const import DOMAIN
NODE_STATUSES = ["asleep", "awake", "dead", "alive"]
CONF_SUBTYPE = "subtype"
CONF_VALUE_ID = "value_id"
homeassistant/components/zwave_js/device_condition.py
+10 -9
View File
@@ -1,10 +1,7 @@
"""Provide the device conditions for Z-Wave JS."""
from typing import cast
import voluptuous as vol
from zwave_js_server.const import CommandClass
from zwave_js_server.model.value import ConfigurationValue
from homeassistant.components.device_automation import InvalidDeviceAutomationConfig
from homeassistant.const import CONF_CONDITION, CONF_DEVICE_ID, CONF_DOMAIN, CONF_TYPE
@@ -21,11 +18,11 @@ from .const import (
ATTR_PROPERTY_KEY,
ATTR_VALUE,
DOMAIN,
NODE_STATUSES,
)
from .device_automation_helpers import (
CONF_SUBTYPE,
CONF_VALUE_ID,
NODE_STATUSES,
async_bypass_dynamic_config_validation,
generate_config_parameter_subtype,
)
@@ -34,7 +31,9 @@ from .helpers import (
check_type_schema_map,
get_value_state_schema,
get_zwave_value_from_config,
node_status_matches,
remove_keys_with_empty_values,
value_matches_state,
)
CONF_STATUS = "status"
@@ -168,7 +167,7 @@ def async_condition_from_config(
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])
return node_status_matches(node, config[CONF_STATUS])
if condition_type == NODE_STATUS_TYPE:
return test_node_status
@@ -177,8 +176,9 @@ def async_condition_from_config(
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])
return value_matches_state(
node.values[config[CONF_VALUE_ID]], config[ATTR_VALUE]
)
if condition_type == CONFIG_PARAMETER_TYPE:
return test_config_parameter
@@ -187,8 +187,9 @@ def async_condition_from_config(
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])
return value_matches_state(
get_zwave_value_from_config(node, config), config[ATTR_VALUE]
)
if condition_type == VALUE_TYPE:
return test_value
homeassistant/components/zwave_js/device_trigger.py
+1 -1
View File
@@ -44,12 +44,12 @@ from .const import (
ATTR_VALUE,
ATTR_VALUE_RAW,
DOMAIN,
NODE_STATUSES,
ZWAVE_JS_NOTIFICATION_EVENT,
ZWAVE_JS_VALUE_NOTIFICATION_EVENT,
)
from .device_automation_helpers import (
CONF_SUBTYPE,
NODE_STATUSES,
async_bypass_dynamic_config_validation,
generate_config_parameter_subtype,
)
homeassistant/components/zwave_js/helpers.py
+54 -5
View File
@@ -34,6 +34,7 @@ from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import (
ATTR_AREA_ID,
ATTR_CONFIG_ENTRY_ID,
ATTR_DEVICE_ID,
ATTR_ENTITY_ID,
CONF_TYPE,
@@ -468,9 +469,9 @@ def get_zwave_value_from_config(node: ZwaveNode, config: ConfigType) -> ZwaveVal
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]
property_key = config.get(ATTR_PROPERTY_KEY)
if property_key == "":
property_key = None
value_id = get_value_id_str(
node,
config[ATTR_COMMAND_CLASS],
@@ -483,7 +484,24 @@ def get_zwave_value_from_config(node: ZwaveNode, config: ConfigType) -> ZwaveVal
return node.values[value_id]
def _zwave_js_config_entry(hass: HomeAssistant, device: dr.DeviceEntry) -> str | None:
def node_status_matches(node: ZwaveNode, status: str) -> bool:
"""Return whether the node has the given status name."""
return node.status.name.lower() == status
def value_matches_state(value: ZwaveValue, expected: Any) -> bool:
"""Return whether a value matches the expected raw value, string form or label."""
current = value.value
return expected in (
current,
str(current),
value.metadata.states.get(str(current), current),
)
def get_zwave_js_config_entry_id(
hass: HomeAssistant, device: dr.DeviceEntry
) -> str | None:
"""Find zwave_js config entry from a device."""
_, config_entry = dr.async_get_device_and_config_entry_for_domain(
hass, device.id, domain=DOMAIN
@@ -506,7 +524,7 @@ def async_get_node_status_sensor_entity_id(
if not (device := dev_reg.async_get(device_id, include_child_devices=False)):
raise HomeAssistantError("Invalid Device ID provided")
if not (entry_id := _zwave_js_config_entry(hass, device)):
if not (entry_id := get_zwave_js_config_entry_id(hass, device)):
return None
entry = hass.config_entries.async_get_entry(entry_id)
@@ -670,3 +688,34 @@ def async_wait_for_driver_ready_event(
class CannotConnect(HomeAssistantError):
"""Indicate connection error."""
@callback
def async_bypass_dynamic_config_validation(
hass: HomeAssistant, config: ConfigType
) -> bool:
"""Return whether a referenced zwave_js config entry is not loaded or ready."""
dev_reg = dr.async_get(hass)
ent_reg = er.async_get(hass)
devices = config.get(ATTR_DEVICE_ID, [])
entities = config.get(ATTR_ENTITY_ID, [])
for entry in hass.config_entries.async_entries(DOMAIN):
if not (
entry.entry_id == config.get(ATTR_CONFIG_ENTRY_ID)
or any(
device.id in devices
for device in dr.async_entries_for_config_entry(dev_reg, entry.entry_id)
)
or any(
entity.entity_id in entities
for entity in er.async_entries_for_config_entry(ent_reg, entry.entry_id)
)
):
continue
if entry.state is not ConfigEntryState.LOADED:
return True
# The driver may not be ready when the config entry is loaded.
if entry.runtime_data.client.driver is None:
return True
return False
homeassistant/components/zwave_js/icons.json
+11
View File
@@ -1,4 +1,15 @@
{
"conditions": {
"config_parameter": {
"condition": "mdi:cog"
},
"node_status": {
"condition": "mdi:heart-pulse"
},
"value": {
"condition": "mdi:update"
}
},
"entity": {
"button": {
"ping": {
homeassistant/components/zwave_js/strings.json
+102
View File
@@ -1,4 +1,98 @@
{
"common": {
"condition_behavior_description": "Whether any or every targeted node must match.",
"condition_behavior_name": "Condition passes if",
"condition_device_id_description": "The Z-Wave JS devices whose nodes to test.",
"condition_device_id_name": "Devices",
"condition_endpoint_description": "Endpoint of the value.",
"condition_endpoint_name": "Endpoint",
"condition_value_description": "The value to compare with, either the raw value or its state label.",
"condition_value_name": "Value"
},
"conditions": {
"config_parameter": {
"description": "Tests if a configuration parameter on one or more Z-Wave JS nodes has the given value.",
"fields": {
"behavior": {
"description": "[%key:component::zwave_js::common::condition_behavior_description%]",
"name": "[%key:component::zwave_js::common::condition_behavior_name%]"
},
"bitmask": {
"description": "Bitmask of a partial parameter, as a number or hexadecimal string, if the parameter is split into parts.",
"name": "Bitmask"
},
"device_id": {
"description": "[%key:component::zwave_js::common::condition_device_id_description%]",
"name": "[%key:component::zwave_js::common::condition_device_id_name%]"
},
"endpoint": {
"description": "[%key:component::zwave_js::common::condition_endpoint_description%]",
"name": "[%key:component::zwave_js::common::condition_endpoint_name%]"
},
"parameter": {
"description": "Number of the configuration parameter.",
"name": "Parameter"
},
"value": {
"description": "[%key:component::zwave_js::common::condition_value_description%]",
"name": "[%key:component::zwave_js::common::condition_value_name%]"
}
},
"name": "Z-Wave JS configuration parameter"
},
"node_status": {
"description": "Tests if one or more Z-Wave JS nodes have the given status.",
"fields": {
"behavior": {
"description": "[%key:component::zwave_js::common::condition_behavior_description%]",
"name": "[%key:component::zwave_js::common::condition_behavior_name%]"
},
"device_id": {
"description": "[%key:component::zwave_js::common::condition_device_id_description%]",
"name": "[%key:component::zwave_js::common::condition_device_id_name%]"
},
"status": {
"description": "The node status to test for.",
"name": "Status"
}
},
"name": "Z-Wave JS node status"
},
"value": {
"description": "Tests if a Z-Wave value on one or more nodes equals the given value.",
"fields": {
"behavior": {
"description": "[%key:component::zwave_js::common::condition_behavior_description%]",
"name": "[%key:component::zwave_js::common::condition_behavior_name%]"
},
"command_class": {
"description": "Command class of the value.",
"name": "Command class"
},
"device_id": {
"description": "[%key:component::zwave_js::common::condition_device_id_description%]",
"name": "[%key:component::zwave_js::common::condition_device_id_name%]"
},
"endpoint": {
"description": "[%key:component::zwave_js::common::condition_endpoint_description%]",
"name": "[%key:component::zwave_js::common::condition_endpoint_name%]"
},
"property": {
"description": "Property of the value.",
"name": "Property"
},
"property_key": {
"description": "Property key of the value.",
"name": "Property key"
},
"value": {
"description": "[%key:component::zwave_js::common::condition_value_description%]",
"name": "[%key:component::zwave_js::common::condition_value_name%]"
}
},
"name": "Z-Wave JS value"
}
},
"config": {
"abort": {
"addon_already_configured": "A configuration entry using the Z-Wave JS app already exists. Reconfigure or migrate that entry instead.",
@@ -562,6 +656,14 @@
"existing": "It already exists",
"new": "It's new"
}
},
"node_status": {
"options": {
"alive": "Alive",
"asleep": "Asleep",
"awake": "Awake",
"dead": "Dead"
}
}
},
"services": {
homeassistant/components/zwave_js/triggers/event.py
+1 -1
View File
@@ -37,12 +37,12 @@ from ..const import (
DOMAIN,
)
from ..helpers import (
async_bypass_dynamic_config_validation,
async_get_config_entry_from_node,
async_get_nodes_from_targets,
get_device_id,
get_home_and_node_id_from_device_entry,
)
from .trigger_helpers import async_bypass_dynamic_config_validation
# Relative platform type should be <SUBMODULE_NAME>
RELATIVE_PLATFORM_TYPE = f"{__name__.rsplit('.', maxsplit=1)[-1]}"
homeassistant/components/zwave_js/triggers/trigger_helpers.py
-42
View File
@@ -1,42 +0,0 @@
"""Helpers for Z-Wave JS custom triggers."""
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import ATTR_CONFIG_ENTRY_ID, 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
from ..const import DOMAIN
@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, [])
for entry in hass.config_entries.async_entries(DOMAIN):
if entry.state is not 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)
)
):
return True
# The driver may not be ready when the config entry is loaded.
client = entry.runtime_data.client
if client.driver is None:
return True
return False
homeassistant/components/zwave_js/triggers/value_updated.py
+1 -1
View File
@@ -40,11 +40,11 @@ from ..const import (
EVENT_VALUE_UPDATED,
)
from ..helpers import (
async_bypass_dynamic_config_validation,
async_get_config_entry_from_node,
async_get_nodes_from_targets,
get_device_id,
)
from .trigger_helpers import async_bypass_dynamic_config_validation
# Relative platform type should be <SUBMODULE_NAME>
RELATIVE_PLATFORM_TYPE = f"{__name__.rsplit('.', maxsplit=1)[-1]}"
tests/components/zwave_js/test_condition.py
+658
View File
@@ -0,0 +1,658 @@
"""The tests for Z-Wave JS conditions."""
from typing import Any
from unittest.mock import MagicMock
import pytest
import voluptuous as vol
from zwave_js_server.const import CommandClass
from zwave_js_server.event import Event
from zwave_js_server.model.node import Node
from homeassistant.components.zwave_js import DOMAIN
from homeassistant.components.zwave_js.condition import CONDITIONS
from homeassistant.components.zwave_js.helpers import get_device_id
from homeassistant.core import HomeAssistant
from homeassistant.helpers import (
condition,
config_validation as cv,
device_registry as dr,
)
from homeassistant.helpers.translation import async_get_translations
from .common import COMMAND_CLASS_MARKERS
from tests.common import MockConfigEntry
async def _checker(
hass: HomeAssistant, config: dict[str, Any]
) -> condition.ConditionChecker:
"""Validate a condition config and build its checker."""
validated = await condition.async_validate_condition_config(
hass, cv.CONDITION_SCHEMA(config)
)
return await condition.async_from_config(hass, validated)
def _device_id(
device_registry: dr.DeviceRegistry,
client: MagicMock,
node: Node,
entry: MockConfigEntry,
) -> str:
"""Return the device registry ID for a node."""
device = device_registry.async_get_device_by_identifier(
get_device_id(client.driver, node), entry.entry_id
)
assert device
return device.id
@pytest.mark.parametrize(
("condition_type", "options", "expected"),
[
pytest.param("node_status", {"status": "alive"}, True, id="node_status_match"),
pytest.param(
"node_status", {"status": "dead"}, False, id="node_status_mismatch"
),
pytest.param(
"config_parameter", {"parameter": 3, "value": 255}, True, id="param_raw"
),
pytest.param(
"config_parameter",
{"parameter": 3, "value": "Enable Beeper"},
True,
id="param_label",
),
pytest.param(
"config_parameter", {"parameter": 3, "value": 0}, False, id="param_mismatch"
),
pytest.param(
"value",
{"command_class": "98", "property": "currentMode", "value": "Unsecured"},
True,
id="value_label",
),
pytest.param(
"value",
{"command_class": 98, "property": "currentMode", "value": 255},
False,
id="value_mismatch",
),
],
)
async def test_condition_by_device(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
condition_type: str,
options: dict[str, Any],
expected: bool,
) -> None:
"""Test each condition targeted by device."""
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.{condition_type}",
"options": {
"device_id": _device_id(
device_registry, client, lock_schlage_be469, integration
),
**options,
},
},
)
assert checker.async_check() is expected
@pytest.mark.parametrize(
("node_name", "condition_type", "options", "expected"),
[
pytest.param(
"iblinds_v3",
"config_parameter",
{"parameter": 3, "value": "Enable"},
True,
id="config_parameter_label_match",
),
pytest.param(
"iblinds_v3",
"config_parameter",
{"parameter": 3, "value": "Disable"},
False,
id="config_parameter_label_mismatch",
),
pytest.param(
"iblinds_v3",
"value",
{"command_class": 112, "property": 3, "value": "0"},
True,
id="value_raw_string_form",
),
pytest.param(
"gdc_zw062",
"value",
{
"command_class": 102,
"property": "signalingState",
"property_key": 1,
"value": "On",
},
True,
id="value_on_label",
),
],
)
async def test_condition_value_state_labels(
hass: HomeAssistant,
client: MagicMock,
iblinds_v3: Node,
gdc_zw062: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
node_name: str,
condition_type: str,
options: dict[str, Any],
expected: bool,
) -> None:
"""Test state labels and raw string values are compared without coercion."""
nodes = {"iblinds_v3": iblinds_v3, "gdc_zw062": gdc_zw062}
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.{condition_type}",
"options": {
"device_id": _device_id(
device_registry, client, nodes[node_name], integration
),
**options,
},
},
)
assert checker.async_check() is expected
@pytest.mark.parametrize(
("behavior", "target_kind", "expected"),
[
pytest.param("any", "two_nodes", True, id="any_one_alive"),
pytest.param("all", "two_nodes", False, id="all_one_alive"),
pytest.param("all", "same_node_twice", True, id="all_deduplicated_node"),
],
)
async def test_node_status_behavior(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
multisensor_6: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
behavior: str,
target_kind: str,
expected: bool,
) -> None:
"""Test any/all behavior, including that a node targeted twice is deduplicated."""
lock_id = _device_id(device_registry, client, lock_schlage_be469, integration)
device_ids = {
"two_nodes": [
lock_id,
_device_id(device_registry, client, multisensor_6, integration),
],
"same_node_twice": [lock_id, lock_id],
}
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.node_status",
"options": {
"device_id": device_ids[target_kind],
"behavior": behavior,
"status": "alive",
},
},
)
assert checker.async_check() is expected
async def test_node_status_follows_events(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test the node status condition reflects status changes."""
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.node_status",
"options": {
"device_id": _device_id(
device_registry, client, lock_schlage_be469, integration
),
"status": "dead",
},
},
)
assert checker.async_check() is False
lock_schlage_be469.receive_event(
Event(
"dead",
data={
"source": "node",
"event": "dead",
"nodeId": lock_schlage_be469.node_id,
},
)
)
assert checker.async_check() is True
async def test_node_status_all_two_nodes_match(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
multisensor_6: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test an all behavior only matches once every targeted node matches."""
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.node_status",
"options": {
"device_id": [
_device_id(
device_registry, client, lock_schlage_be469, integration
),
_device_id(device_registry, client, multisensor_6, integration),
],
"behavior": "all",
"status": "alive",
},
},
)
assert checker.async_check() is False
multisensor_6.receive_event(
Event(
"alive",
data={
"source": "node",
"event": "alive",
"nodeId": multisensor_6.node_id,
},
)
)
assert checker.async_check() is True
async def test_value_missing_on_node(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
multisensor_6: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test a node without the value does not match and validation needs one node with it."""
lock_id = _device_id(device_registry, client, lock_schlage_be469, integration)
sensor_id = _device_id(device_registry, client, multisensor_6, integration)
options = {"command_class": 98, "property": "currentMode", "value": 0}
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.value",
"options": {
"device_id": [lock_id, sensor_id],
**options,
"behavior": "all",
},
},
)
assert checker.async_check() is False
with pytest.raises(vol.Invalid, match="No targeted node has value"):
await _checker(
hass,
{
"condition": f"{DOMAIN}.value",
"options": {"device_id": sensor_id, **options},
},
)
async def test_value_property_key_zero(
hass: HomeAssistant,
client: MagicMock,
bulb_6_multi_color: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test a property key of 0 is not treated as an absent property key."""
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.value",
"options": {
"device_id": _device_id(
device_registry, client, bulb_6_multi_color, integration
),
"command_class": 51,
"property": "currentColor",
"property_key": 0,
"value": 255,
},
},
)
assert checker.async_check() is True
async def test_no_nodes_resolved(
hass: HomeAssistant,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test validation rejects devices that resolve to no Z-Wave nodes."""
other = device_registry.async_get_or_create(
config_entry_id=integration.entry_id, identifiers={("other", "1")}
)
with pytest.raises(vol.Invalid, match="No nodes found"):
await _checker(
hass,
{
"condition": f"{DOMAIN}.node_status",
"options": {"device_id": other.id, "status": "alive"},
},
)
async def test_validation_bypassed_when_not_loaded(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test dynamic validation is skipped while the config entry is not loaded."""
device_id = _device_id(device_registry, client, lock_schlage_be469, integration)
await hass.config_entries.async_unload(integration.entry_id)
validated = await condition.async_validate_condition_config(
hass,
cv.CONDITION_SCHEMA(
{
"condition": f"{DOMAIN}.value",
"options": {
"device_id": device_id,
"command_class": 98,
"property": "nope",
"value": 0,
},
}
),
)
assert validated["options"]["property"] == "nope"
async def test_config_parameter_with_bitmask(
hass: HomeAssistant,
client: MagicMock,
multisensor_6: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test a config parameter condition with a partial parameter bitmask."""
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.config_parameter",
"options": {
"device_id": _device_id(
device_registry, client, multisensor_6, integration
),
"parameter": 101,
"bitmask": "0x1",
"value": 1,
},
},
)
assert checker.async_check() is True
async def test_top_level_fields_moved_to_options(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test top level option fields are moved into the options block."""
device_id = _device_id(device_registry, client, lock_schlage_be469, integration)
validated = await condition.async_validate_condition_config(
hass,
cv.CONDITION_SCHEMA(
{
"condition": f"{DOMAIN}.node_status",
"device_id": device_id,
"status": "alive",
"behavior": "all",
}
),
)
assert validated["options"] == {
"device_id": [device_id],
"behavior": "all",
"status": "alive",
}
assert "status" not in validated
async def test_check_false_when_nodes_disappear(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test the condition is False once the devices no longer resolve to nodes."""
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.node_status",
"options": {
"device_id": _device_id(
device_registry, client, lock_schlage_be469, integration
),
"status": "alive",
},
},
)
assert checker.async_check() is True
await hass.config_entries.async_unload(integration.entry_id)
assert checker.async_check() is False
@pytest.mark.parametrize(
("behavior", "expected"),
[("any", True), ("all", False)],
)
async def test_partially_unresolved_target(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
multisensor_6: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
behavior: str,
expected: bool,
) -> None:
"""Test a targeted Z-Wave node that cannot be resolved fails an all behavior."""
device_ids = [
_device_id(device_registry, client, lock_schlage_be469, integration),
_device_id(device_registry, client, multisensor_6, integration),
]
del client.driver.controller.nodes[multisensor_6.node_id]
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.node_status",
"options": {
"device_id": device_ids,
"behavior": behavior,
"status": "alive",
},
},
)
assert checker.async_check() is expected
async def test_config_parameter_missing_on_node(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test validation fails when no node in the target has the parameter."""
device_id = _device_id(device_registry, client, lock_schlage_be469, integration)
with pytest.raises(vol.Invalid, match="configuration parameter"):
await _checker(
hass,
{
"condition": f"{DOMAIN}.config_parameter",
"options": {
"device_id": device_id,
"parameter": 9999,
"value": 1,
},
},
)
@pytest.mark.parametrize("condition_type", list(CONDITIONS), ids=list(CONDITIONS))
@pytest.mark.usefixtures("integration")
async def test_condition_description_fields_match_schema(
hass: HomeAssistant, condition_type: str
) -> None:
"""Test the described fields and required flags match the options schema."""
schema = CONDITIONS[condition_type].options_schema_dict
descriptions = await condition.async_get_all_descriptions(hass)
description = descriptions[f"{DOMAIN}.{condition_type}"]
# Nodes are targeted with a device selector field, not a target selector
assert "target" not in description
fields = description["fields"]
assert set(fields) == {str(key) for key in schema}
assert {name for name, field in fields.items() if field["required"]} == {
str(key) for key in schema if isinstance(key, vol.Required)
}
@pytest.mark.parametrize("condition_type", list(CONDITIONS), ids=list(CONDITIONS))
@pytest.mark.usefixtures("integration")
async def test_condition_device_selector(
hass: HomeAssistant, condition_type: str
) -> None:
"""Test every condition picks nodes with a multiple zwave_js device selector."""
descriptions = await condition.async_get_all_descriptions(hass)
selector = descriptions[f"{DOMAIN}.{condition_type}"]["fields"]["device_id"][
"selector"
]["device"]
assert selector["filter"] == [{"integration": DOMAIN}]
assert selector["multiple"] is True
@pytest.mark.usefixtures("integration")
async def test_value_command_class_options(hass: HomeAssistant) -> None:
"""Test the value condition's command class options match the CommandClass enum."""
expected = {str(cc.value) for cc in CommandClass if cc not in COMMAND_CLASS_MARKERS}
descriptions = await condition.async_get_all_descriptions(hass)
options = descriptions[f"{DOMAIN}.value"]["fields"]["command_class"]["selector"][
"select"
]["options"]
assert len(options) == len(expected)
assert set(options) == expected
@pytest.mark.usefixtures("integration")
async def test_node_status_selector_translations(hass: HomeAssistant) -> None:
"""Test the node status selector options are translated."""
translations = await async_get_translations(hass, "en", "selector", {DOMAIN})
prefix = f"component.{DOMAIN}.selector.node_status.options."
assert {
key.removeprefix(prefix) for key in translations if key.startswith(prefix)
} == {"alive", "asleep", "awake", "dead"}
async def test_non_zwave_device_is_unresolved(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test a device from another integration counts as an unresolved node."""
other_entry = MockConfigEntry(domain="other")
other_entry.add_to_hass(hass)
other_device = device_registry.async_get_or_create(
config_entry_id=other_entry.entry_id, identifiers={("other", "dev")}
)
device_ids = [
_device_id(device_registry, client, lock_schlage_be469, integration),
other_device.id,
]
assert (
await _checker(
hass,
{
"condition": f"{DOMAIN}.node_status",
"options": {
"device_id": device_ids,
"behavior": "any",
"status": "alive",
},
},
)
).async_check() is True
assert (
await _checker(
hass,
{
"condition": f"{DOMAIN}.node_status",
"options": {
"device_id": device_ids,
"behavior": "all",
"status": "alive",
},
},
)
).async_check() is False
async def test_value_empty_property_key(
hass: HomeAssistant,
client: MagicMock,
lock_schlage_be469: Node,
integration: MockConfigEntry,
device_registry: dr.DeviceRegistry,
) -> None:
"""Test an empty property key is treated as no property key."""
checker = await _checker(
hass,
{
"condition": f"{DOMAIN}.value",
"options": {
"device_id": _device_id(
device_registry, client, lock_schlage_be469, integration
),
"command_class": 112,
"property": 3,
"property_key": "",
"value": 255,
},
},
)
assert checker.async_check() is True
tests/components/zwave_js/test_device_condition.py
+9 -1
View File
@@ -372,6 +372,13 @@ async def test_config_parameter_state(
assert service_calls[1].data["some"] == "User Slot Status - event - test_event2"
@pytest.mark.parametrize(
"value",
[
pytest.param(255, id="raw_value"),
pytest.param("Enable Beeper", id="state_label"),
],
)
async def test_value_state(
hass: HomeAssistant,
client,
@@ -379,6 +386,7 @@ async def test_value_state(
integration,
service_calls: list[ServiceCall],
device_registry: dr.DeviceRegistry,
value: int | str,
) -> None:
"""Test for value conditions."""
device = device_registry.async_get_device_by_identifier(
@@ -401,7 +409,7 @@ async def test_value_state(
"type": "value",
"command_class": 112,
"property": 3,
"value": 255,
"value": value,
}
],
"action": {
tests/components/zwave_js/test_trigger.py
+79 -42
View File
@@ -12,17 +12,18 @@ 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.helpers import get_device_id
from homeassistant.components.zwave_js.helpers import (
async_bypass_dynamic_config_validation,
get_device_id,
)
from homeassistant.components.zwave_js.trigger import TRIGGERS
from homeassistant.components.zwave_js.triggers.event import (
_OPTIONS_SCHEMA_DICT as EVENT_OPTIONS_SCHEMA_DICT,
)
from homeassistant.components.zwave_js.triggers.trigger_helpers import (
async_bypass_dynamic_config_validation,
)
from homeassistant.components.zwave_js.triggers.value_updated import (
_OPTIONS_SCHEMA_DICT as VALUE_UPDATED_OPTIONS_SCHEMA_DICT,
)
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import SERVICE_RELOAD
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, trigger
@@ -1145,12 +1146,9 @@ async def test_zwave_js_trigger_config_entry_unloaded(
assert not async_bypass_dynamic_config_validation(
hass,
{
"platform": f"{DOMAIN}.value_updated",
"options": {
"entity_id": SCHLAGE_BE469_LOCK_ENTITY,
"command_class": CommandClass.DOOR_LOCK.value,
"property": "latchStatus",
},
"entity_id": [SCHLAGE_BE469_LOCK_ENTITY],
"command_class": CommandClass.DOOR_LOCK.value,
"property": "latchStatus",
},
)
@@ -1185,66 +1183,105 @@ async def test_zwave_js_trigger_config_entry_unloaded(
assert async_bypass_dynamic_config_validation(
hass,
{
"platform": f"{DOMAIN}.value_updated",
"options": {
"entity_id": SCHLAGE_BE469_LOCK_ENTITY,
"command_class": CommandClass.DOOR_LOCK.value,
"property": "latchStatus",
},
"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",
"options": {
"device_id": device.id,
"command_class": CommandClass.DOOR_LOCK.value,
"property": "latchStatus",
"from": "ajar",
},
"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",
"options": {
"entity_id": SCHLAGE_BE469_LOCK_ENTITY,
"event_source": "node",
"event": "interview stage completed",
},
"entity_id": [SCHLAGE_BE469_LOCK_ENTITY],
"event_source": "node",
"event": "interview stage completed",
},
)
assert async_bypass_dynamic_config_validation(
hass,
{
"platform": f"{DOMAIN}.event",
"options": {
"device_id": device.id,
"event_source": "node",
"event": "interview stage completed",
"event_data": {"stageName": "ProtocolInfo"},
},
"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",
"options": {
"config_entry_id": integration.entry_id,
"event_source": "controller",
"event": "nvm convert progress",
},
"config_entry_id": integration.entry_id,
"event_source": "controller",
"event": "nvm convert progress",
},
)
@pytest.mark.parametrize(
("config_key", "driver", "expected"),
[
pytest.param("loaded_device", MagicMock(), False, id="loaded_device"),
pytest.param("loaded_entity", MagicMock(), False, id="loaded_entity"),
pytest.param("unloaded_device", MagicMock(), True, id="unloaded_device"),
pytest.param(
"unloaded_entry", MagicMock(), True, id="unloaded_config_entry_id"
),
pytest.param("nothing", MagicMock(), False, id="nothing_referenced"),
pytest.param("loaded_device", None, True, id="loaded_device_driver_not_ready"),
pytest.param("nothing", None, False, id="nothing_referenced_driver_not_ready"),
],
)
async def test_bypass_dynamic_config_validation_scoped(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
client: MagicMock,
lock_schlage_be469: Node,
integration: MockConfigEntry,
config_key: str,
driver: MagicMock | None,
expected: bool,
) -> None:
"""Test the bypass check only considers config entries referenced by the config."""
lock_device = device_registry.async_get_device_by_identifier(
get_device_id(client.driver, lock_schlage_be469), integration.entry_id
)
assert lock_device
other_entry = MockConfigEntry(
domain=DOMAIN, data={"url": "ws://test2.org"}, unique_id="other"
)
other_entry.add_to_hass(hass)
other_device = device_registry.async_get_or_create(
config_entry_id=other_entry.entry_id, identifiers={(DOMAIN, "other-node")}
)
assert other_entry.state is not ConfigEntryState.LOADED
configs = {
"loaded_device": {"device_id": [lock_device.id]},
"loaded_entity": {"entity_id": [SCHLAGE_BE469_LOCK_ENTITY]},
"unloaded_device": {"device_id": [other_device.id]},
"unloaded_entry": {"config_entry_id": other_entry.entry_id},
"nothing": {},
}
with patch.object(client, "driver", driver):
assert (
async_bypass_dynamic_config_validation(hass, configs[config_key])
is expected
)
async def test_server_reconnect_event(
hass: HomeAssistant,
client,