diff --git a/homeassistant/components/zwave_js/__init__.py b/homeassistant/components/zwave_js/__init__.py index 1682edb66a3..5a0a7bbf29f 100644 --- a/homeassistant/components/zwave_js/__init__.py +++ b/homeassistant/components/zwave_js/__init__.py @@ -12,6 +12,7 @@ from zwave_js_server.model.node import Node as ZwaveNode from zwave_js_server.model.notification import ( EntryControlNotification, NotificationNotification, + PowerLevelNotification, ) from zwave_js_server.model.value import Value, ValueNotification @@ -41,6 +42,7 @@ from homeassistant.helpers.typing import ConfigType from .addon import AddonError, AddonManager, AddonState, get_addon_manager from .api import async_register_api from .const import ( + ATTR_ACKNOWLEDGED_FRAMES, ATTR_COMMAND_CLASS, ATTR_COMMAND_CLASS_NAME, ATTR_DATA_TYPE, @@ -57,6 +59,8 @@ from .const import ( ATTR_PROPERTY_KEY, ATTR_PROPERTY_KEY_NAME, ATTR_PROPERTY_NAME, + ATTR_STATUS, + ATTR_TEST_NODE_ID, ATTR_TYPE, ATTR_VALUE, ATTR_VALUE_RAW, @@ -392,7 +396,9 @@ async def async_setup_entry( # noqa: C901 @callback def async_on_notification( - notification: EntryControlNotification | NotificationNotification, + notification: EntryControlNotification + | NotificationNotification + | PowerLevelNotification, ) -> None: """Relay stateless notification events from Z-Wave nodes to hass.""" device = dev_reg.async_get_device({get_device_id(client, notification.node)}) @@ -415,7 +421,7 @@ async def async_setup_entry( # noqa: C901 ATTR_EVENT_DATA: notification.event_data, } ) - else: + elif isinstance(notification, NotificationNotification): event_data.update( { ATTR_COMMAND_CLASS_NAME: "Notification", @@ -426,6 +432,17 @@ async def async_setup_entry( # noqa: C901 ATTR_PARAMETERS: notification.parameters, } ) + elif isinstance(notification, PowerLevelNotification): + event_data.update( + { + ATTR_COMMAND_CLASS_NAME: "Power Level", + ATTR_TEST_NODE_ID: notification.test_node_id, + ATTR_STATUS: notification.status, + ATTR_ACKNOWLEDGED_FRAMES: notification.acknowledged_frames, + } + ) + else: + raise TypeError(f"Unhandled notification type: {notification}") hass.bus.async_fire(ZWAVE_JS_NOTIFICATION_EVENT, event_data) diff --git a/homeassistant/components/zwave_js/const.py b/homeassistant/components/zwave_js/const.py index 2d16c0113c9..bd46497f629 100644 --- a/homeassistant/components/zwave_js/const.py +++ b/homeassistant/components/zwave_js/const.py @@ -56,6 +56,9 @@ ATTR_EVENT_DATA = "event_data" ATTR_DATA_TYPE = "data_type" ATTR_WAIT_FOR_RESULT = "wait_for_result" ATTR_OPTIONS = "options" +ATTR_TEST_NODE_ID = "test_node_id" +ATTR_STATUS = "status" +ATTR_ACKNOWLEDGED_FRAMES = "acknowledged_frames" ATTR_NODE = "node" ATTR_ZWAVE_VALUE = "zwave_value" diff --git a/tests/components/zwave_js/test_events.py b/tests/components/zwave_js/test_events.py index d66cd52be40..32859ae3c37 100644 --- a/tests/components/zwave_js/test_events.py +++ b/tests/components/zwave_js/test_events.py @@ -1,4 +1,7 @@ """Test Z-Wave JS (value notification) events.""" +from unittest.mock import AsyncMock + +import pytest from zwave_js_server.const import CommandClass from zwave_js_server.event import Event @@ -259,3 +262,48 @@ async def test_value_updated(hass, vision_security_zl7432, integration, client): await hass.async_block_till_done() # We should only still have captured one event assert len(events) == 1 + + +async def test_power_level_notification(hass, hank_binary_switch, integration, client): + """Test power level notification events.""" + # just pick a random node to fake the notification event + node = hank_binary_switch + events = async_capture_events(hass, "zwave_js_notification") + + event = Event( + type="notification", + data={ + "source": "node", + "event": "notification", + "nodeId": 7, + "ccId": 115, + "args": { + "commandClassName": "Powerlevel", + "commandClass": 115, + "testNodeId": 1, + "status": 0, + "acknowledgedFrames": 2, + }, + }, + ) + node.receive_event(event) + await hass.async_block_till_done() + assert len(events) == 1 + assert events[0].data["command_class_name"] == "Power Level" + assert events[0].data["command_class"] == 115 + assert events[0].data["test_node_id"] == 1 + assert events[0].data["status"] == 0 + assert events[0].data["acknowledged_frames"] == 2 + + +async def test_unknown_notification(hass, hank_binary_switch, integration, client): + """Test behavior of unknown notification type events.""" + # just pick a random node to fake the notification event + node = hank_binary_switch + + # We emit the event directly so we can skip any validation and event handling + # by the lib. We will use a class that is guaranteed not to be recognized + notification_obj = AsyncMock() + notification_obj.node = node + with pytest.raises(TypeError): + node.emit("notification", {"notification": notification_obj})