Add event platform to isy994 integration (#169782)

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
shbatm
2026-09-16 11:00:33 +00:00
committed by GitHub
co-authored by Claude Opus 4.7 Copilot Autofix powered by AI Joost Lekkerkerker
parent d5aca4c291
commit 10aebaf423
9 changed files with 945 additions and 3 deletions
+48
View File
@@ -80,6 +80,8 @@ DEFAULT_VAR_SENSOR_STRING = "HA."
KEY_ACTIONS = "actions"
KEY_STATUS = "status"
EVENT_ISY994_CONTROL = "isy994_control"
NODE_PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.CLIMATE,
@@ -106,10 +108,15 @@ PROGRAM_PLATFORMS = [
ROOT_NODE_PLATFORMS = [Platform.BUTTON]
VARIABLE_PLATFORMS = [Platform.NUMBER, Platform.SENSOR]
# Platforms that classify in parallel with NODE_PLATFORMS — a node placed in
# one of these still falls through to its primary platform classification.
NODE_PARALLEL_PLATFORMS = [Platform.EVENT]
# Set of all platforms used by integration
PLATFORMS = {
*NODE_PLATFORMS,
*NODE_AUX_PROP_PLATFORMS,
*NODE_PARALLEL_PLATFORMS,
*PROGRAM_PLATFORMS,
*ROOT_NODE_PLATFORMS,
*VARIABLE_PLATFORMS,
@@ -315,6 +322,47 @@ NODE_FILTERS: dict[Platform, dict[str, list[str]]] = {
FILTER_INSTEON_TYPE: ["4.8", TYPE_CATEGORY_CLIMATE],
FILTER_ZWAVE_CAT: ["140"],
},
# Additive: a node matching here still gets its primary classification.
Platform.EVENT: {
FILTER_UOM: [],
FILTER_STATES: [],
FILTER_NODE_DEF_ID: [
"BallastRelayLampSwitch",
"BallastRelayLampSwitch_ADV",
"DimmerLampSwitch",
"DimmerLampSwitch_ADV",
"DimmerSwitchOnly",
"DimmerSwitchOnly_ADV",
"KeypadButton",
"KeypadButton_ADV",
"KeypadDimmer",
"KeypadDimmer_ADV",
"KeypadRelay",
"KeypadRelay_ADV",
"RelayLampOnly",
"RelayLampOnly_ADV",
"RelayLampSwitch",
"RelayLampSwitch_ADV",
"RelaySwitchOnlyPlusQuery",
"RelaySwitchOnlyPlusQuery_ADV",
],
# Type prefixes derived from observed eisy node families
# (SwitchLinc / KeypadLinc / InLineLinc / BallastLinc) — catches
# legacy non-_ADV firmware variants of the same hardware.
FILTER_INSTEON_TYPE: [
"1.14.",
"1.32.",
"1.45.",
"1.65.",
"1.66.",
"2.42.",
"2.44.",
"2.55.",
"2.57.",
"3.32.",
],
FILTER_ZWAVE_CAT: [],
},
}
NODE_AUX_FILTERS: dict[str, Platform] = {
PROP_ON_LEVEL: Platform.NUMBER,
+2 -2
View File
@@ -25,7 +25,7 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity, EntityDescription
from .const import DOMAIN
from .const import DOMAIN, EVENT_ISY994_CONTROL
class ISYEntity(Entity):
@@ -82,7 +82,7 @@ class ISYEntity(Entity):
# New state attributes may be available, update the state.
self.async_write_ha_state()
self.hass.bus.async_fire("isy994_control", event_data)
self.hass.bus.async_fire(EVENT_ISY994_CONTROL, event_data)
class ISYNodeEntity(ISYEntity):
+205
View File
@@ -0,0 +1,205 @@
"""Event entities for ISY Insteon load and keypad-button nodes.
Each entity represents a physical button on the device and emits one of the
standard button event types (architecture#1377) when its corresponding
control event arrives from the ISY.
"""
from typing import TYPE_CHECKING, Final, NamedTuple, override
from pyisy.constants import (
ATTR_ACTION,
CMD_FADE_DOWN,
CMD_FADE_STOP,
CMD_FADE_UP,
CMD_OFF,
CMD_OFF_FAST,
CMD_ON,
CMD_ON_FAST,
ES_CONNECTED,
NC_NODE_ENABLED,
TAG_ADDRESS,
)
from pyisy.helpers import NodeProperty
from pyisy.nodes import Node, NodeChangedEvent
from homeassistant.components.event import (
ATTR_MULTI_PRESS_COUNT,
ButtonEventType,
EventDeviceClass,
EventEntity,
EventEntityDescription,
)
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .entity import ISYNodeEntity
if TYPE_CHECKING:
from .models import IsyConfigEntry
EVENT_BUTTON_UNIQUE_ID_SUFFIX = "_button"
ATTR_DIRECTION = "direction"
DIRECTION_UP = "up"
DIRECTION_DOWN = "down"
class _ControlEvent(NamedTuple):
"""Standard event type + direction/count for one ISY control command."""
event_type: ButtonEventType
direction: str
multi_press_count: int | None = None
# Maps to the architecture#1377 standard button event types. `direction`
# distinguishes the two paddle positions of a single physical button rather
# than splitting into two entities. CMD_FADE_STOP
# (long-press end) isn't listed here: its direction is whichever fade most
# recently started, tracked in `ISYButtonEvent._last_fade_direction`.
CONTROL_TO_EVENT: Final[dict[str, _ControlEvent]] = {
CMD_ON: _ControlEvent(ButtonEventType.PRESS_END, DIRECTION_UP),
CMD_OFF: _ControlEvent(ButtonEventType.PRESS_END, DIRECTION_DOWN),
CMD_ON_FAST: _ControlEvent(ButtonEventType.MULTI_PRESS_END, DIRECTION_UP, 2),
CMD_OFF_FAST: _ControlEvent(ButtonEventType.MULTI_PRESS_END, DIRECTION_DOWN, 2),
CMD_FADE_UP: _ControlEvent(ButtonEventType.LONG_PRESS_START, DIRECTION_UP),
CMD_FADE_DOWN: _ControlEvent(ButtonEventType.LONG_PRESS_START, DIRECTION_DOWN),
}
BUTTON_DESCRIPTION: Final[EventEntityDescription] = EventEntityDescription(
key="button",
translation_key="button",
device_class=EventDeviceClass.BUTTON,
event_types=[
ButtonEventType.PRESS_END,
ButtonEventType.MULTI_PRESS_END,
ButtonEventType.LONG_PRESS_START,
ButtonEventType.LONG_PRESS_END,
],
)
def _sub_button_name(node: Node) -> str:
"""Return the sub-button label with the parent device prefix stripped.
ISY users commonly label KeypadLinc sub-buttons as ``"<device> <suffix>"``
(e.g. ``"Hallway Keypad B"``), which would render as ``"Hallway Keypad
Hallway Keypad B"`` under ``has_entity_name=True``. Falls back to the raw
node name when the prefix doesn't match. The label is user-supplied in the
ISY admin console and is not translatable.
"""
parent_name: str = node.parent_node.name
name: str = node.name
if name.startswith(parent_name) and (
len(name) == len(parent_name) or name[len(parent_name)] in " -_:."
):
return name[len(parent_name) :].lstrip(" -_:.") or name
return name
async def async_setup_entry(
hass: HomeAssistant,
entry: IsyConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the ISY event platform."""
isy_data = entry.runtime_data
device_info = isy_data.devices
async_add_entities(
ISYButtonEvent(node, device_info.get(node.primary_node))
for node in isy_data.nodes[Platform.EVENT]
)
class ISYButtonEvent(ISYNodeEntity, EventEntity):
"""Event entity that emits press/fast/fade events from an ISY node."""
entity_description = BUTTON_DESCRIPTION
_attr_has_entity_name = True
def __init__(self, node: Node, device_info: DeviceInfo | None = None) -> None:
"""Initialize the ISY button event entity."""
super().__init__(node, device_info=device_info)
self._attr_unique_id = (
f"{node.isy.uuid}_{node.address}{EVENT_BUTTON_UNIQUE_ID_SUFFIX}"
)
self._last_fade_direction: str | None = None
if node.parent_node is None:
self._attr_name = None
else:
self._attr_name = _sub_button_name(node)
# Disabled by default — a typical KeypadLinc exposes 6-8 of
# these and most users only automate a few.
self._attr_entity_registry_enabled_default = False
@override
# pylint: disable-next=home-assistant-missing-super-call
async def async_added_to_hass(self) -> None:
"""Subscribe to control events and node enabled/disabled changes only.
Skipping the base class's status_events subscription avoids a state
write on every value update; availability still tracks the node's
enabled flag via a filtered subscription.
"""
self._control_handler = self._node.control_events.subscribe(
self.async_on_control
)
self.async_on_remove(self._control_handler.unsubscribe)
self._change_handler = self._node.isy.nodes.status_events.subscribe(
self._async_on_availability_change,
event_filter={
TAG_ADDRESS: self._node.address,
ATTR_ACTION: NC_NODE_ENABLED,
},
key=self.unique_id,
)
self.async_on_remove(self._change_handler.unsubscribe)
@callback
def _async_on_availability_change(self, event: NodeChangedEvent, key: str) -> None:
"""Refresh state when the node is enabled or disabled."""
self.async_write_ha_state()
@callback
@override
def async_on_control(self, event: NodeProperty) -> None:
"""Trigger the entity, bypassing the base class's bus.fire.
The load entity for the same node still fires `isy994_control` via
the base class, so we don't fire it here to avoid double-emission.
Suppressed while the websocket isn't fully connected -- PyISY
replays the current status of every node on (re)connect before
settling, and without this guard that replay fires stale button
events on every startup, config-entry reload, and reconnect.
"""
websocket = self._node.isy.websocket
if websocket is not None and websocket.status != ES_CONNECTED:
return
if event.control == CMD_FADE_STOP:
# FADE_STOP carries no direction of its own, so it consumes the
# one from the fade it ends; a stop with no preceding start
# reports no direction rather than a stale one.
direction = self._last_fade_direction
self._last_fade_direction = None
self._trigger_event(
ButtonEventType.LONG_PRESS_END,
{ATTR_DIRECTION: direction} if direction is not None else None,
)
self.async_write_ha_state()
return
control_event = CONTROL_TO_EVENT.get(event.control)
if control_event is None:
return
if control_event.event_type == ButtonEventType.LONG_PRESS_START:
self._last_fade_direction = control_event.direction
event_attributes: dict[str, str | int] = {
ATTR_DIRECTION: control_event.direction
}
if control_event.multi_press_count is not None:
event_attributes[ATTR_MULTI_PRESS_COUNT] = control_event.multi_press_count
self._trigger_event(control_event.event_type, event_attributes)
self.async_write_ha_state()
@@ -39,6 +39,7 @@ from .const import (
LOGGER,
NODE_AUX_FILTERS,
NODE_FILTERS,
NODE_PARALLEL_PLATFORMS,
NODE_PLATFORMS,
PROGRAM_PLATFORMS,
SUBNODE_CLIMATE_COOL,
@@ -368,6 +369,17 @@ def _categorize_nodes(
continue
isy_data.aux_properties[Platform.SENSOR].append((node, control))
# Must run before the sensor_identifier override below -- Platform.EVENT
# is additive, not exclusive with a name/path-forced Platform.SENSOR.
for parallel_platform in NODE_PARALLEL_PLATFORMS:
if _check_for_node_def(isy_data, node, single_platform=parallel_platform):
continue
if _check_for_insteon_type(
isy_data, node, single_platform=parallel_platform
):
continue
_check_for_zwave_cat(isy_data, node, single_platform=parallel_platform)
if sensor_identifier in path or sensor_identifier in node.name:
# User has specified to treat this as a sensor. First we need to
# determine if it should be a binary_sensor.
+13 -1
View File
@@ -17,11 +17,13 @@ from homeassistant.helpers.device_registry import DeviceInfo
from .const import (
CONF_NETWORK,
NODE_AUX_PROP_PLATFORMS,
NODE_PARALLEL_PLATFORMS,
NODE_PLATFORMS,
PROGRAM_PLATFORMS,
ROOT_NODE_PLATFORMS,
VARIABLE_PLATFORMS,
)
from .event import EVENT_BUTTON_UNIQUE_ID_SUFFIX
type IsyConfigEntry = ConfigEntry[IsyData]
@@ -41,7 +43,7 @@ class IsyData:
def __init__(self) -> None:
"""Initialize an empty ISY data class."""
self.nodes = {p: [] for p in NODE_PLATFORMS}
self.nodes = {p: [] for p in (*NODE_PLATFORMS, *NODE_PARALLEL_PLATFORMS)}
self.root_nodes = {p: [] for p in ROOT_NODE_PLATFORMS}
self.aux_properties = {p: [] for p in NODE_AUX_PROP_PLATFORMS}
self.programs = {p: [] for p in PROGRAM_PLATFORMS}
@@ -95,4 +97,14 @@ class IsyData:
for node in self.net_resources:
current_unique_ids.add((Platform.BUTTON, self.uid_base(node)))
# Separate from the NODE_PLATFORMS loop above: event unique ids carry
# a suffix, since the same node also has a primary-platform entity.
for node in self.nodes[Platform.EVENT]:
current_unique_ids.add(
(
Platform.EVENT,
f"{self.uid_base(node)}{EVENT_BUTTON_UNIQUE_ID_SUFFIX}",
)
)
return current_unique_ids
@@ -35,6 +35,21 @@
}
}
},
"entity": {
"event": {
"button": {
"state_attributes": {
"direction": {
"name": "Direction",
"state": {
"down": "Down",
"up": "Up"
}
}
}
}
}
},
"options": {
"step": {
"init": {
+2
View File
@@ -2,6 +2,7 @@
from unittest.mock import AsyncMock, MagicMock, patch
from pyisy.constants import ES_CONNECTED
from pyisy.nodes import Node
import pytest
@@ -42,6 +43,7 @@ def mock_isy():
mock.networking.nobjs = []
mock.clock = MagicMock()
mock.websocket = MagicMock()
mock.websocket.status = ES_CONNECTED
mock.conf = {
"name": "Skynet ISY",
"model": "IoX",
@@ -0,0 +1,261 @@
# serializer version: 1
# name: test_event_entity_snapshot[event.garage_relay-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<EventEntityCapabilityAttribute.EVENT_TYPES: 'event_types'>: list([
<ButtonEventType.PRESS_END: 'press_end'>,
<ButtonEventType.MULTI_PRESS_END: 'multi_press_end'>,
<ButtonEventType.LONG_PRESS_START: 'long_press_start'>,
<ButtonEventType.LONG_PRESS_END: 'long_press_end'>,
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.garage_relay',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.BUTTON: 'button'>,
'original_icon': None,
'original_name': None,
'platform': 'isy994',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'button',
'unique_id': '00:00:00:00:00:00_22 22 22 1_button',
'unit_of_measurement': None,
})
# ---
# name: test_event_entity_snapshot[event.garage_relay-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'button',
<EventEntityStateAttribute.EVENT_TYPE: 'event_type'>: None,
<EventEntityCapabilityAttribute.EVENT_TYPES: 'event_types'>: list([
<ButtonEventType.PRESS_END: 'press_end'>,
<ButtonEventType.MULTI_PRESS_END: 'multi_press_end'>,
<ButtonEventType.LONG_PRESS_START: 'long_press_start'>,
<ButtonEventType.LONG_PRESS_END: 'long_press_end'>,
]),
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Garage Relay',
}),
'context': <ANY>,
'entity_id': 'event.garage_relay',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_event_entity_snapshot[event.hallway_keypad-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<EventEntityCapabilityAttribute.EVENT_TYPES: 'event_types'>: list([
<ButtonEventType.PRESS_END: 'press_end'>,
<ButtonEventType.MULTI_PRESS_END: 'multi_press_end'>,
<ButtonEventType.LONG_PRESS_START: 'long_press_start'>,
<ButtonEventType.LONG_PRESS_END: 'long_press_end'>,
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.hallway_keypad',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.BUTTON: 'button'>,
'original_icon': None,
'original_name': None,
'platform': 'isy994',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'button',
'unique_id': '00:00:00:00:00:00_33 33 33 1_button',
'unit_of_measurement': None,
})
# ---
# name: test_event_entity_snapshot[event.hallway_keypad-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'button',
<EventEntityStateAttribute.EVENT_TYPE: 'event_type'>: None,
<EventEntityCapabilityAttribute.EVENT_TYPES: 'event_types'>: list([
<ButtonEventType.PRESS_END: 'press_end'>,
<ButtonEventType.MULTI_PRESS_END: 'multi_press_end'>,
<ButtonEventType.LONG_PRESS_START: 'long_press_start'>,
<ButtonEventType.LONG_PRESS_END: 'long_press_end'>,
]),
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Hallway Keypad',
}),
'context': <ANY>,
'entity_id': 'event.hallway_keypad',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_event_entity_snapshot[event.hallway_keypad_b-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<EventEntityCapabilityAttribute.EVENT_TYPES: 'event_types'>: list([
<ButtonEventType.PRESS_END: 'press_end'>,
<ButtonEventType.MULTI_PRESS_END: 'multi_press_end'>,
<ButtonEventType.LONG_PRESS_START: 'long_press_start'>,
<ButtonEventType.LONG_PRESS_END: 'long_press_end'>,
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.hallway_keypad_b',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'B',
'options': dict({
}),
'original_device_class': <EventDeviceClass.BUTTON: 'button'>,
'original_icon': None,
'original_name': 'B',
'platform': 'isy994',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'button',
'unique_id': '00:00:00:00:00:00_33 33 33 2_button',
'unit_of_measurement': None,
})
# ---
# name: test_event_entity_snapshot[event.hallway_keypad_b-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'button',
<EventEntityStateAttribute.EVENT_TYPE: 'event_type'>: None,
<EventEntityCapabilityAttribute.EVENT_TYPES: 'event_types'>: list([
<ButtonEventType.PRESS_END: 'press_end'>,
<ButtonEventType.MULTI_PRESS_END: 'multi_press_end'>,
<ButtonEventType.LONG_PRESS_START: 'long_press_start'>,
<ButtonEventType.LONG_PRESS_END: 'long_press_end'>,
]),
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Hallway Keypad B',
}),
'context': <ANY>,
'entity_id': 'event.hallway_keypad_b',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_event_entity_snapshot[event.living_room_switch-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': dict({
<EventEntityCapabilityAttribute.EVENT_TYPES: 'event_types'>: list([
<ButtonEventType.PRESS_END: 'press_end'>,
<ButtonEventType.MULTI_PRESS_END: 'multi_press_end'>,
<ButtonEventType.LONG_PRESS_START: 'long_press_start'>,
<ButtonEventType.LONG_PRESS_END: 'long_press_end'>,
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'event',
'entity_category': None,
'entity_id': 'event.living_room_switch',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': <EventDeviceClass.BUTTON: 'button'>,
'original_icon': None,
'original_name': None,
'platform': 'isy994',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'button',
'unique_id': '00:00:00:00:00:00_11 11 11 1_button',
'unit_of_measurement': None,
})
# ---
# name: test_event_entity_snapshot[event.living_room_switch-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.DEVICE_CLASS: 'device_class'>: 'button',
<EventEntityStateAttribute.EVENT_TYPE: 'event_type'>: None,
<EventEntityCapabilityAttribute.EVENT_TYPES: 'event_types'>: list([
<ButtonEventType.PRESS_END: 'press_end'>,
<ButtonEventType.MULTI_PRESS_END: 'multi_press_end'>,
<ButtonEventType.LONG_PRESS_START: 'long_press_start'>,
<ButtonEventType.LONG_PRESS_END: 'long_press_end'>,
]),
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Living Room Switch',
}),
'context': <ANY>,
'entity_id': 'event.living_room_switch',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
+387
View File
@@ -0,0 +1,387 @@
"""Test the ISY994 event platform."""
from collections.abc import Callable, Generator
from typing import Any
from unittest.mock import MagicMock, patch
from pyisy.constants import (
CMD_FADE_DOWN,
CMD_FADE_STOP,
CMD_FADE_UP,
CMD_OFF,
CMD_OFF_FAST,
CMD_ON,
CMD_ON_FAST,
ES_SYNCING,
)
from pyisy.helpers import NodeProperty
from pyisy.nodes import NodeChangedEvent
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.isy994.const import EVENT_ISY994_CONTROL
from homeassistant.components.isy994.event import _sub_button_name
from homeassistant.const import STATE_UNAVAILABLE, Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, async_capture_events, snapshot_platform
SWITCH_ENTITY_ID = "switch.garage_relay"
@pytest.fixture
def platforms() -> list[Platform]:
"""Return the platforms to set up, overridden by parametrization."""
return [Platform.EVENT]
@pytest.fixture(autouse=True)
def mock_event_platform(platforms: list[Platform]) -> Generator[None]:
"""Mock the platforms that are set up."""
with patch("homeassistant.components.isy994.PLATFORMS", platforms):
yield
def _make_button_nodes(
mock_isy: MagicMock, mock_node: Callable[..., Any]
) -> list[tuple[str, MagicMock]]:
"""Build a representative set of button-emitting Insteon nodes."""
nodes: list[tuple[str, MagicMock]] = []
# Primary loads — enabled by default
primary = mock_node(
mock_isy, "11 11 11 1", "Living Room Switch", "DimmerLampSwitch_ADV"
)
nodes.append(("Living Room Switch", primary))
relay = mock_node(mock_isy, "22 22 22 1", "Garage Relay", "RelayLampSwitch_ADV")
nodes.append(("Garage Relay", relay))
keypad_load = mock_node(
mock_isy, "33 33 33 1", "Hallway Keypad", "KeypadDimmer_ADV"
)
nodes.append(("Hallway Keypad", keypad_load))
# Secondary keypad button — disabled by default
sub_button = mock_node(
mock_isy, "33 33 33 2", "Hallway Keypad B", "KeypadButton_ADV"
)
sub_button.parent_node = keypad_load
sub_button.primary_node = "33 33 33 1"
nodes.append(("Hallway Keypad B", sub_button))
return nodes
async def test_event_entity_snapshot(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
entity_registry: er.EntityRegistry,
mock_isy: MagicMock,
mock_node: Callable[..., Any],
) -> None:
"""Snapshot the event entities created for supported Insteon nodes."""
mock_config_entry.add_to_hass(hass)
mock_isy.nodes.__iter__.return_value = _make_button_nodes(mock_isy, mock_node)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
assert any(entry.disabled_by is not None for entry in entries)
for entry in entries:
if entry.disabled_by:
entity_registry.async_update_entity(entry.entity_id, disabled_by=None)
await hass.config_entries.async_reload(mock_config_entry.entry_id)
await hass.async_block_till_done()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize(
("parent_name", "node_name", "expected"),
[
("Hallway Keypad", "Hallway Keypad B", "B"),
("Hall", "Hallway B", "Hallway B"), # prefix with no separator: unchanged
],
)
def test_sub_button_name_requires_separator(
parent_name: str, node_name: str, expected: str
) -> None:
"""A parent name that is a bare prefix (no separator) must not be stripped.
"Hall" is a prefix of "Hallway B" with no separator between them, so the
sub-button label must stay "Hallway B" rather than being corrupted to
"way B".
"""
node = MagicMock()
node.parent_node.name = parent_name
node.name = node_name
assert _sub_button_name(node) == expected
@pytest.mark.parametrize(
("control", "expected_event_type", "expected_direction", "expected_count"),
[
(CMD_ON, "press_end", "up", None),
(CMD_OFF, "press_end", "down", None),
(CMD_ON_FAST, "multi_press_end", "up", 2),
(CMD_OFF_FAST, "multi_press_end", "down", 2),
(CMD_FADE_UP, "long_press_start", "up", None),
(CMD_FADE_DOWN, "long_press_start", "down", None),
],
)
async def test_control_event_triggers_entity(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_isy: MagicMock,
mock_node: Callable[..., Any],
control: str,
expected_event_type: str,
expected_direction: str,
expected_count: int | None,
) -> None:
"""Control events from pyisy translate into the standard button event types."""
mock_config_entry.add_to_hass(hass)
node = mock_node(mock_isy, "11 11 11 1", "Test Switch", "DimmerLampSwitch_ADV")
mock_isy.nodes.__iter__.return_value = [("Test Switch", node)]
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
handler = node.control_events.subscribe.call_args.args[0]
handler(MagicMock(spec=NodeProperty, control=control))
await hass.async_block_till_done()
entity_ids = hass.states.async_entity_ids("event")
assert len(entity_ids) == 1
state = hass.states.get(entity_ids[0])
assert state is not None
assert state.attributes["event_type"] == expected_event_type
assert state.attributes.get("direction") == expected_direction
assert state.attributes.get("multi_press_count") == expected_count
@pytest.mark.parametrize(
("controls", "expected_direction"),
[
pytest.param([CMD_FADE_DOWN, CMD_FADE_STOP], "down", id="paired_stop"),
pytest.param([CMD_FADE_STOP], None, id="orphan_stop"),
pytest.param(
[CMD_FADE_DOWN, CMD_FADE_STOP, CMD_FADE_STOP], None, id="duplicate_stop"
),
],
)
async def test_fade_stop_direction(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_isy: MagicMock,
mock_node: Callable[..., Any],
controls: list[str],
expected_direction: str | None,
) -> None:
"""CMD_FADE_STOP reports the direction of the fade it ends, once.
The remembered direction is consumed by the stop that uses it, so a stop
with no preceding fade start reports no direction instead of a stale one.
"""
mock_config_entry.add_to_hass(hass)
node = mock_node(mock_isy, "11 11 11 1", "Test Switch", "DimmerLampSwitch_ADV")
mock_isy.nodes.__iter__.return_value = [("Test Switch", node)]
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
handler = node.control_events.subscribe.call_args.args[0]
for control in controls:
handler(MagicMock(spec=NodeProperty, control=control))
await hass.async_block_till_done()
entity_ids = hass.states.async_entity_ids("event")
state = hass.states.get(entity_ids[0])
assert state is not None
assert state.attributes["event_type"] == "long_press_end"
assert state.attributes.get("direction") == expected_direction
# An unknown direction omits the attribute entirely rather than
# publishing it as null.
assert ("direction" in state.attributes) is (expected_direction is not None)
async def test_disabling_node_marks_entity_unavailable(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_isy: MagicMock,
mock_node: Callable[..., Any],
) -> None:
"""Availability tracks the node's enabled flag.
The entity skips the base class's status_events subscription, so only the
filtered enabled/disabled subscription keeps `available` current.
"""
mock_config_entry.add_to_hass(hass)
node = mock_node(mock_isy, "11 11 11 1", "Test Switch", "DimmerLampSwitch_ADV")
mock_isy.nodes.__iter__.return_value = [("Test Switch", node)]
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_id = hass.states.async_entity_ids("event")[0]
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
node.enabled = False
handler = mock_isy.nodes.status_events.subscribe.call_args.args[0]
handler(MagicMock(spec=NodeChangedEvent), "key")
await hass.async_block_till_done()
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
async def test_control_event_suppressed_while_websocket_syncing(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_isy: MagicMock,
mock_node: Callable[..., Any],
) -> None:
"""Control events are dropped while the websocket replays status on connect.
Without this guard, PyISY's post-connect status replay fires stale
button events on every startup, config-entry reload, and reconnect.
"""
mock_config_entry.add_to_hass(hass)
node = mock_node(mock_isy, "11 11 11 1", "Test Switch", "DimmerLampSwitch_ADV")
mock_isy.nodes.__iter__.return_value = [("Test Switch", node)]
mock_isy.websocket.status = ES_SYNCING
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
handler = node.control_events.subscribe.call_args.args[0]
handler(MagicMock(spec=NodeProperty, control=CMD_ON))
await hass.async_block_till_done()
entity_ids = hass.states.async_entity_ids("event")
state = hass.states.get(entity_ids[0])
assert state is not None
assert state.attributes.get("event_type") is None
@pytest.mark.parametrize("platforms", [[Platform.EVENT, Platform.SWITCH]])
@pytest.mark.parametrize(
"control", [CMD_ON, CMD_OFF, CMD_ON_FAST, CMD_OFF_FAST, CMD_FADE_UP, CMD_FADE_STOP]
)
async def test_legacy_control_bus_event_not_duplicated(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_isy: MagicMock,
mock_node: Callable[..., Any],
control: str,
) -> None:
"""The legacy isy994_control bus event still fires exactly once per control.
The node's primary (switch) entity keeps firing the bus event from the
base class, while the event entity overrides async_on_control and must
not fire a second one for the same control.
"""
mock_config_entry.add_to_hass(hass)
node = mock_node(mock_isy, "22 22 22 1", "Garage Relay", "RelayLampSwitch_ADV")
mock_isy.nodes.__iter__.return_value = [("Garage Relay", node)]
events = async_capture_events(hass, EVENT_ISY994_CONTROL)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get(SWITCH_ENTITY_ID) is not None
assert len(hass.states.async_entity_ids("event")) == 1
for call in node.control_events.subscribe.call_args_list:
call.args[0](MagicMock(spec=NodeProperty, control=control))
await hass.async_block_till_done()
assert len(events) == 1
assert events[0].data["entity_id"] == SWITCH_ENTITY_ID
assert events[0].data["control"] == control
async def test_unsupported_control_is_ignored(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_isy: MagicMock,
mock_node: Callable[..., Any],
) -> None:
"""Control events not in the mapping must not trigger the entity."""
mock_config_entry.add_to_hass(hass)
node = mock_node(mock_isy, "11 11 11 1", "Test Switch", "DimmerLampSwitch_ADV")
mock_isy.nodes.__iter__.return_value = [("Test Switch", node)]
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
handler = node.control_events.subscribe.call_args.args[0]
handler(MagicMock(spec=NodeProperty, control="ST"))
await hass.async_block_till_done()
entity_ids = hass.states.async_entity_ids("event")
assert len(entity_ids) == 1
state = hass.states.get(entity_ids[0])
assert state is not None
assert state.attributes.get("event_type") is None
@pytest.mark.parametrize(
("node_type", "expect_event_entity"),
[
("1.14.1", True), # SwitchLinc prefix from FILTER_INSTEON_TYPE
("2.44.1", True), # KeypadLinc dimmer prefix from FILTER_INSTEON_TYPE
("3.32.1", True), # BallastLinc prefix from FILTER_INSTEON_TYPE
("1.20.1", False), # Not in FILTER_INSTEON_TYPE
],
)
async def test_legacy_insteon_type_fallback(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_isy: MagicMock,
mock_node: Callable[..., Any],
node_type: str,
expect_event_entity: bool,
) -> None:
"""Pre-5.0-firmware nodes with no node_def_id fall back to type-prefix matching."""
mock_config_entry.add_to_hass(hass)
node = mock_node(mock_isy, "11 11 11 1", "Legacy Switch", None, node_type=node_type)
mock_isy.nodes.__iter__.return_value = [("Legacy Switch", node)]
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
entity_ids = hass.states.async_entity_ids("event")
assert (len(entity_ids) == 1) is expect_event_entity
async def test_event_entity_created_despite_sensor_string_override(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_isy: MagicMock,
mock_node: Callable[..., Any],
) -> None:
"""A node forced into Platform.SENSOR by the sensor_string option still gets its event entity.
Platform.EVENT is a parallel classification, not exclusive with the
user's sensor_string override -- a SwitchLinc/KeypadLinc whose name
happens to contain the (default "sensor") override string must still be
matched against NODE_PARALLEL_PLATFORMS.
"""
mock_config_entry.add_to_hass(hass)
node = mock_node(
mock_isy, "11 11 11 1", "Garage sensor Switch", "DimmerLampSwitch_ADV"
)
mock_isy.nodes.__iter__.return_value = [("Garage sensor Switch", node)]
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert len(hass.states.async_entity_ids("event")) == 1