Improve zwave_js notification event handling (#66790)

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)
 

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"

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})