Handle unrecognized notification types

test/model/test_node.py

@@ -17,8 +17,14 @@
     EntryControlDataType,
     EntryControlEventType,
 )
+from zwave_js_server.const.command_class.power_level import PowerLevelTestStatus
 from zwave_js_server.event import Event
-from zwave_js_server.exceptions import FailedCommand, NotFoundError, UnwriteableValue
+from zwave_js_server.exceptions import (
+    FailedCommand,
+    NotFoundError,
+    NotificationHasUnsupportedCommandClass,
+    UnwriteableValue,
+)
 from zwave_js_server.model import node as node_pkg
 from zwave_js_server.model.firmware import FirmwareUpdateStatus
 from zwave_js_server.model.node_health_check import (
@@ -712,6 +718,29 @@ async def test_notification(lock_schlage_be469: node_pkg.Node):
     """Test notification CC notification events."""
     node = lock_schlage_be469
 
+    # Validate that Entry Control CC notification event is received as expected
+    event = Event(
+        type="notification",
+        data={
+            "source": "node",
+            "event": "notification",
+            "nodeId": 23,
+            "ccId": 111,
+            "args": {
+                "eventType": 0,
+                "dataType": 0,
+                "eventData": "test",
+            },
+        },
+    )
+
+    node.handle_notification(event)
+    assert event.data["notification"].command_class == CommandClass.ENTRY_CONTROL
+    assert event.data["notification"].node_id == 23
+    assert event.data["notification"].event_type == EntryControlEventType.CACHING
+    assert event.data["notification"].data_type == EntryControlDataType.NONE
+    assert event.data["notification"].event_data == "test"
+
     # Validate that Notification CC notification event is received as expected
     event = Event(
         type="notification",
@@ -739,6 +768,39 @@ async def test_notification(lock_schlage_be469: node_pkg.Node):
     assert event.data["notification"].event_label == "Keypad lock operation"
     assert event.data["notification"].parameters == {"userId": 1}
 
+    # Validate that Power Level CC notification event is received as expected
+    event = Event(
+        type="notification",
+        data={
+            "source": "node",
+            "event": "notification",
+            "nodeId": 23,
+            "ccId": CommandClass.POWERLEVEL.value,
+            "args": {"testNodeId": 1, "status": 0, "acknowledgedFrames": 2},
+        },
+    )
+
+    node.handle_notification(event)
+    assert event.data["notification"].command_class == CommandClass.POWERLEVEL
+    assert event.data["notification"].node_id == 23
+    assert event.data["notification"].test_node_id == 1
+    assert event.data["notification"].status == PowerLevelTestStatus.FAILED
+    assert event.data["notification"].acknowledged_frames == 2
+
+    # Validate that an unrecognized CC notification event raises Exception
+    event = Event(
+        type="notification",
+        data={
+            "source": "node",
+            "event": "notification",
+            "nodeId": 23,
+            "ccId": 0,
+        },
+    )
+
+    with pytest.raises(NotificationHasUnsupportedCommandClass):
+        node.handle_notification(event)
+
 
 async def test_entry_control_notification(ring_keypad):
     """Test entry control CC notification events."""

zwave_js_server/const/command_class/power_level.py

@@ -0,0 +1,11 @@
+"""Constants for the Power Level Command Class."""
+from enum import IntEnum
+
+
+class PowerLevelTestStatus(IntEnum):
+    """Enum with all known power level test statuses."""
+
+    # https://github.com/zwave-js/node-zwave-js/blob/master/packages/zwave-js/src/lib/commandclass/PowerlevelCC.ts#L52
+    FAILED = 0
+    SUCCESS = 1
+    IN_PROGRESS = 2

zwave_js_server/exceptions.py

@@ -4,6 +4,7 @@
 from .const import CommandClass
 
 if TYPE_CHECKING:
+    from .event import Event
     from .model.value import Value
 
 
@@ -156,3 +157,16 @@ def __init__(self, value: "Value", path: str) -> None:
             "upstream issue with the driver or missing support for this data in the "
             "library"
         )
+
+
+class NotificationHasUnsupportedCommandClass(BaseZwaveJSServerError):
+    """Exception raised when notification is received for an unsupported CC."""
+
+    def __init__(self, event: "Event", command_class: CommandClass) -> None:
+        """Initialize an invalid Command Class error."""
+        self.event_data = event.data
+        self.command_class = command_class
+        super().__init__(
+            "Notification received with unsupported command class "
+            f"{command_class.name}: {event.data}"
+        )

zwave_js_server/model/node.py

@@ -12,6 +12,7 @@
 from ..exceptions import (
     FailedCommand,
     NotFoundError,
+    NotificationHasUnsupportedCommandClass,
     UnparseableValue,
     UnwriteableValue,
 )
@@ -37,6 +38,8 @@
     EntryControlNotificationDataType,
     NotificationNotification,
     NotificationNotificationDataType,
+    PowerLevelNotification,
+    PowerLevelNotificationDataType,
 )
 from .value import (
     ConfigurationValue,
@@ -811,14 +814,21 @@ def handle_metadata_updated(self, event: Event) -> None:
 
     def handle_notification(self, event: Event) -> None:
         """Process a node notification event."""
-        if event.data["ccId"] == CommandClass.NOTIFICATION.value:
+        command_class = CommandClass(event.data["ccId"])
+        if command_class == CommandClass.NOTIFICATION:
             event.data["notification"] = NotificationNotification(
                 self, cast(NotificationNotificationDataType, event.data)
             )
-        else:
+        elif command_class == CommandClass.ENTRY_CONTROL:
             event.data["notification"] = EntryControlNotification(
                 self, cast(EntryControlNotificationDataType, event.data)
             )
+        elif command_class == CommandClass.POWERLEVEL:
+            event.data["notification"] = PowerLevelNotification(
+                self, cast(PowerLevelNotificationDataType, event.data)
+            )
+        else:
+            raise NotificationHasUnsupportedCommandClass(event, command_class)
 
     def handle_firmware_update_progress(self, event: Event) -> None:
         """Process a node firmware update progress event."""

zwave_js_server/model/notification.py

@@ -5,15 +5,16 @@
 """
 
 from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, TypedDict, Union
+from zwave_js_server.const.command_class.power_level import PowerLevelTestStatus
 
 from zwave_js_server.util.helpers import parse_buffer
 
 if TYPE_CHECKING:
     from .node import Node
 
 
-class NotificationDataType(TypedDict):
-    """Represent a generic notification event data dict type."""
+class BaseNotificationDataType(TypedDict):
+    """Represent a base notification event data dict type."""
 
     source: Literal["node"]  # required
     event: Literal["notification"]  # required
@@ -29,12 +30,48 @@ class EntryControlNotificationArgsDataType(TypedDict, total=False):
     eventData: Union[str, Dict[str, Any]]
 
 
-class EntryControlNotificationDataType(NotificationDataType):
+class EntryControlNotificationDataType(BaseNotificationDataType):
     """Represent an Entry Control CC notification event data dict type."""
 
     args: EntryControlNotificationArgsDataType  # required
 
 
+class EntryControlNotification:
+    """Model for a Zwave Node's Entry Control CC notification event."""
+
+    def __init__(self, node: "Node", data: EntryControlNotificationDataType) -> None:
+        """Initialize."""
+        self.node = node
+        self.data = data
+
+    @property
+    def node_id(self) -> int:
+        """Return node ID property."""
+        return self.data["nodeId"]
+
+    @property
+    def command_class(self) -> int:
+        """Return command class."""
+        return self.data["ccId"]
+
+    @property
+    def event_type(self) -> int:
+        """Return event type property."""
+        return self.data["args"]["eventType"]
+
+    @property
+    def data_type(self) -> int:
+        """Return data type property."""
+        return self.data["args"]["dataType"]
+
+    @property
+    def event_data(self) -> Optional[str]:
+        """Return event data property."""
+        if event_data := self.data["args"].get("eventData"):
+            return parse_buffer(event_data)
+        return None
+
+
 class NotificationNotificationArgsDataType(TypedDict, total=False):
     """Represent args for a Notification CC notification event data dict type."""
 
@@ -45,7 +82,7 @@ class NotificationNotificationArgsDataType(TypedDict, total=False):
     parameters: Dict[str, Any]
 
 
-class NotificationNotificationDataType(NotificationDataType):
+class NotificationNotificationDataType(BaseNotificationDataType):
     """Represent a Notification CC notification event data dict type."""
 
     args: NotificationNotificationArgsDataType  # required
@@ -95,10 +132,24 @@ def parameters(self) -> Dict[str, Any]:
         return self.data["args"].get("parameters", {})
 
 
-class EntryControlNotification:
-    """Model for a Zwave Node's Entry Control CC notification event."""
+class PowerLevelNotificationArgsDataType(TypedDict):
+    """Represent args for a Power Level CC notification event data dict type."""
 
-    def __init__(self, node: "Node", data: EntryControlNotificationDataType) -> None:
+    testNodeId: int
+    status: int
+    acknowledgedFrames: int
+
+
+class PowerLevelNotificationDataType(BaseNotificationDataType):
+    """Represent a Power Level CC notification event data dict type."""
+
+    args: PowerLevelNotificationArgsDataType  # required
+
+
+class PowerLevelNotification:
+    """Model for a Zwave Node's Power Level CC notification event."""
+
+    def __init__(self, node: "Node", data: PowerLevelNotificationDataType) -> None:
         """Initialize."""
         self.node = node
         self.data = data
@@ -114,18 +165,16 @@ def command_class(self) -> int:
         return self.data["ccId"]
 
     @property
-    def event_type(self) -> int:
-        """Return event type property."""
-        return self.data["args"]["eventType"]
+    def test_node_id(self) -> int:
+        """Return test node ID property."""
+        return self.data["args"]["testNodeId"]
 
     @property
-    def data_type(self) -> int:
-        """Return data type property."""
-        return self.data["args"]["dataType"]
+    def status(self) -> PowerLevelTestStatus:
+        """Return status."""
+        return PowerLevelTestStatus(self.data["args"]["status"])
 
     @property
-    def event_data(self) -> Optional[str]:
-        """Return event data property."""
-        if event_data := self.data["args"].get("eventData"):
-            return parse_buffer(event_data)
-        return None
+    def acknowledged_frames(self) -> int:
+        """Return acknowledged frames property."""
+        return self.data["args"]["acknowledgedFrames"]