From 52fd53f5ec8fe3856e94b6950ef0580de0dcfa4d Mon Sep 17 00:00:00 2001 From: Raman Gupta <7243222+raman325@users.noreply.github.com> Date: Wed, 9 Feb 2022 23:54:33 -0500 Subject: [PATCH 1/3] Handle unrecognized notification types --- test/model/test_node.py | 21 ++++++++++++++++++- test/util/command_class/test_meter.py | 4 ++-- .../command_class/test_multilevel_sensor.py | 4 ++-- zwave_js_server/exceptions.py | 16 +++++++++++++- zwave_js_server/model/node.py | 8 +++++-- zwave_js_server/util/command_class/meter.py | 4 ++-- .../util/command_class/multilevel_sensor.py | 4 ++-- 7 files changed, 49 insertions(+), 12 deletions(-) diff --git a/test/model/test_node.py b/test/model/test_node.py index 1e6502f96..47a746976 100644 --- a/test/model/test_node.py +++ b/test/model/test_node.py @@ -18,7 +18,12 @@ EntryControlEventType, ) 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 ( @@ -739,6 +744,20 @@ 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 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.""" diff --git a/test/util/command_class/test_meter.py b/test/util/command_class/test_meter.py index a658fc465..6899079d3 100644 --- a/test/util/command_class/test_meter.py +++ b/test/util/command_class/test_meter.py @@ -8,7 +8,7 @@ ElectricScale, MeterType, ) -from zwave_js_server.exceptions import InvalidCommandClass, UnknownValueData +from zwave_js_server.exceptions import ValueHasInvalidCommandClass, UnknownValueData from zwave_js_server.model.node import Node from zwave_js_server.model.value import MetaDataType, Value, ValueDataType, get_value_id from zwave_js_server.util.command_class.meter import ( @@ -22,7 +22,7 @@ async def test_get_meter_type(inovelli_switch: Node): node = inovelli_switch value_id = get_value_id(node, CommandClass.SWITCH_BINARY, "currentValue") - with pytest.raises(InvalidCommandClass): + with pytest.raises(ValueHasInvalidCommandClass): get_meter_type(node.values.get(value_id)) value_id = get_value_id(node, CommandClass.METER, "value", property_key=65537) diff --git a/test/util/command_class/test_multilevel_sensor.py b/test/util/command_class/test_multilevel_sensor.py index ab2eefd7f..22f580993 100644 --- a/test/util/command_class/test_multilevel_sensor.py +++ b/test/util/command_class/test_multilevel_sensor.py @@ -6,7 +6,7 @@ MultilevelSensorType, TemperatureScale, ) -from zwave_js_server.exceptions import InvalidCommandClass, UnknownValueData +from zwave_js_server.exceptions import ValueHasInvalidCommandClass, UnknownValueData from zwave_js_server.model.node import Node from zwave_js_server.model.value import MetaDataType, Value, ValueDataType, get_value_id from zwave_js_server.util.command_class.multilevel_sensor import ( @@ -22,7 +22,7 @@ async def test_get_multilevel_sensor_type(multisensor_6: Node): node = multisensor_6 value_id = get_value_id(node, CommandClass.SENSOR_BINARY, "Any") - with pytest.raises(InvalidCommandClass): + with pytest.raises(ValueHasInvalidCommandClass): get_multilevel_sensor_type(node.values.get(value_id)) value_id = get_value_id(node, CommandClass.SENSOR_MULTILEVEL, "Air temperature") diff --git a/zwave_js_server/exceptions.py b/zwave_js_server/exceptions.py index 367855e7f..f49575bcf 100644 --- a/zwave_js_server/exceptions.py +++ b/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 @@ -125,7 +126,7 @@ class BulkSetConfigParameterFailed(BaseZwaveJSServerError): """ -class InvalidCommandClass(BaseZwaveJSServerError): +class ValueHasInvalidCommandClass(BaseZwaveJSServerError): """Exception raised when Zwave Value has an invalid command class.""" def __init__(self, value: "Value", command_class: CommandClass) -> None: @@ -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}" + ) diff --git a/zwave_js_server/model/node.py b/zwave_js_server/model/node.py index 92e491193..9e3012742 100644 --- a/zwave_js_server/model/node.py +++ b/zwave_js_server/model/node.py @@ -12,6 +12,7 @@ from ..exceptions import ( FailedCommand, NotFoundError, + NotificationHasUnsupportedCommandClass, UnparseableValue, UnwriteableValue, ) @@ -811,14 +812,17 @@ 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) ) + else: + raise NotificationHasUnsupportedCommandClass(event, command_class) def handle_firmware_update_progress(self, event: Event) -> None: """Process a node firmware update progress event.""" diff --git a/zwave_js_server/util/command_class/meter.py b/zwave_js_server/util/command_class/meter.py index aa55f53cf..e4dc2d717 100644 --- a/zwave_js_server/util/command_class/meter.py +++ b/zwave_js_server/util/command_class/meter.py @@ -7,14 +7,14 @@ MeterScaleType, MeterType, ) -from ...exceptions import InvalidCommandClass, UnknownValueData +from ...exceptions import ValueHasInvalidCommandClass, UnknownValueData from ...model.value import Value def get_meter_type(value: Value) -> MeterType: """Get the MeterType for a given value.""" if value.command_class != CommandClass.METER: - raise InvalidCommandClass(value, CommandClass.METER) + raise ValueHasInvalidCommandClass(value, CommandClass.METER) try: return MeterType(value.metadata.cc_specific[CC_SPECIFIC_METER_TYPE]) except ValueError: diff --git a/zwave_js_server/util/command_class/multilevel_sensor.py b/zwave_js_server/util/command_class/multilevel_sensor.py index 9601a26ca..ab507fffc 100644 --- a/zwave_js_server/util/command_class/multilevel_sensor.py +++ b/zwave_js_server/util/command_class/multilevel_sensor.py @@ -7,14 +7,14 @@ MultilevelSensorType, MULTILEVEL_SENSOR_TYPE_TO_SCALE_MAP, ) -from ...exceptions import InvalidCommandClass, UnknownValueData +from ...exceptions import ValueHasInvalidCommandClass, UnknownValueData from ...model.value import Value def get_multilevel_sensor_type(value: Value) -> MultilevelSensorType: """Get the MultilevelSensorType for a given value.""" if value.command_class != CommandClass.SENSOR_MULTILEVEL: - raise InvalidCommandClass(value, CommandClass.SENSOR_MULTILEVEL) + raise ValueHasInvalidCommandClass(value, CommandClass.SENSOR_MULTILEVEL) try: return MultilevelSensorType(value.metadata.cc_specific[CC_SPECIFIC_SENSOR_TYPE]) except ValueError: From 25ff0dfaafa34400cd896a77d63a1c023ff79d80 Mon Sep 17 00:00:00 2001 From: Raman Gupta <7243222+raman325@users.noreply.github.com> Date: Thu, 10 Feb 2022 00:28:33 -0500 Subject: [PATCH 2/3] Add support for power level notifications --- test/model/test_node.py | 43 ++++++++++ .../const/command_class/power_level.py | 11 +++ zwave_js_server/model/node.py | 6 ++ zwave_js_server/model/notification.py | 85 +++++++++++++++---- 4 files changed, 127 insertions(+), 18 deletions(-) create mode 100644 zwave_js_server/const/command_class/power_level.py diff --git a/test/model/test_node.py b/test/model/test_node.py index 47a746976..699530dd7 100644 --- a/test/model/test_node.py +++ b/test/model/test_node.py @@ -17,6 +17,7 @@ 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, @@ -717,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", @@ -744,6 +768,25 @@ 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", diff --git a/zwave_js_server/const/command_class/power_level.py b/zwave_js_server/const/command_class/power_level.py new file mode 100644 index 000000000..e947eb4a6 --- /dev/null +++ b/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 diff --git a/zwave_js_server/model/node.py b/zwave_js_server/model/node.py index 9e3012742..1373cd2d1 100644 --- a/zwave_js_server/model/node.py +++ b/zwave_js_server/model/node.py @@ -38,6 +38,8 @@ EntryControlNotificationDataType, NotificationNotification, NotificationNotificationDataType, + PowerLevelNotification, + PowerLevelNotificationDataType, ) from .value import ( ConfigurationValue, @@ -821,6 +823,10 @@ def handle_notification(self, event: Event) -> None: 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) diff --git a/zwave_js_server/model/notification.py b/zwave_js_server/model/notification.py index b1c2c100b..df4893400 100644 --- a/zwave_js_server/model/notification.py +++ b/zwave_js_server/model/notification.py @@ -5,6 +5,7 @@ """ 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 @@ -12,8 +13,8 @@ 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"] From 050964e20080adf8e28b243daa08553dfda88d37 Mon Sep 17 00:00:00 2001 From: Raman Gupta <7243222+raman325@users.noreply.github.com> Date: Thu, 10 Feb 2022 10:06:58 -0500 Subject: [PATCH 3/3] revert exception name change --- test/util/command_class/test_meter.py | 4 ++-- test/util/command_class/test_multilevel_sensor.py | 4 ++-- zwave_js_server/exceptions.py | 2 +- zwave_js_server/util/command_class/meter.py | 4 ++-- zwave_js_server/util/command_class/multilevel_sensor.py | 4 ++-- 5 files changed, 9 insertions(+), 9 deletions(-) diff --git a/test/util/command_class/test_meter.py b/test/util/command_class/test_meter.py index 6899079d3..a658fc465 100644 --- a/test/util/command_class/test_meter.py +++ b/test/util/command_class/test_meter.py @@ -8,7 +8,7 @@ ElectricScale, MeterType, ) -from zwave_js_server.exceptions import ValueHasInvalidCommandClass, UnknownValueData +from zwave_js_server.exceptions import InvalidCommandClass, UnknownValueData from zwave_js_server.model.node import Node from zwave_js_server.model.value import MetaDataType, Value, ValueDataType, get_value_id from zwave_js_server.util.command_class.meter import ( @@ -22,7 +22,7 @@ async def test_get_meter_type(inovelli_switch: Node): node = inovelli_switch value_id = get_value_id(node, CommandClass.SWITCH_BINARY, "currentValue") - with pytest.raises(ValueHasInvalidCommandClass): + with pytest.raises(InvalidCommandClass): get_meter_type(node.values.get(value_id)) value_id = get_value_id(node, CommandClass.METER, "value", property_key=65537) diff --git a/test/util/command_class/test_multilevel_sensor.py b/test/util/command_class/test_multilevel_sensor.py index 22f580993..ab2eefd7f 100644 --- a/test/util/command_class/test_multilevel_sensor.py +++ b/test/util/command_class/test_multilevel_sensor.py @@ -6,7 +6,7 @@ MultilevelSensorType, TemperatureScale, ) -from zwave_js_server.exceptions import ValueHasInvalidCommandClass, UnknownValueData +from zwave_js_server.exceptions import InvalidCommandClass, UnknownValueData from zwave_js_server.model.node import Node from zwave_js_server.model.value import MetaDataType, Value, ValueDataType, get_value_id from zwave_js_server.util.command_class.multilevel_sensor import ( @@ -22,7 +22,7 @@ async def test_get_multilevel_sensor_type(multisensor_6: Node): node = multisensor_6 value_id = get_value_id(node, CommandClass.SENSOR_BINARY, "Any") - with pytest.raises(ValueHasInvalidCommandClass): + with pytest.raises(InvalidCommandClass): get_multilevel_sensor_type(node.values.get(value_id)) value_id = get_value_id(node, CommandClass.SENSOR_MULTILEVEL, "Air temperature") diff --git a/zwave_js_server/exceptions.py b/zwave_js_server/exceptions.py index f49575bcf..b6ee18d37 100644 --- a/zwave_js_server/exceptions.py +++ b/zwave_js_server/exceptions.py @@ -126,7 +126,7 @@ class BulkSetConfigParameterFailed(BaseZwaveJSServerError): """ -class ValueHasInvalidCommandClass(BaseZwaveJSServerError): +class InvalidCommandClass(BaseZwaveJSServerError): """Exception raised when Zwave Value has an invalid command class.""" def __init__(self, value: "Value", command_class: CommandClass) -> None: diff --git a/zwave_js_server/util/command_class/meter.py b/zwave_js_server/util/command_class/meter.py index e4dc2d717..aa55f53cf 100644 --- a/zwave_js_server/util/command_class/meter.py +++ b/zwave_js_server/util/command_class/meter.py @@ -7,14 +7,14 @@ MeterScaleType, MeterType, ) -from ...exceptions import ValueHasInvalidCommandClass, UnknownValueData +from ...exceptions import InvalidCommandClass, UnknownValueData from ...model.value import Value def get_meter_type(value: Value) -> MeterType: """Get the MeterType for a given value.""" if value.command_class != CommandClass.METER: - raise ValueHasInvalidCommandClass(value, CommandClass.METER) + raise InvalidCommandClass(value, CommandClass.METER) try: return MeterType(value.metadata.cc_specific[CC_SPECIFIC_METER_TYPE]) except ValueError: diff --git a/zwave_js_server/util/command_class/multilevel_sensor.py b/zwave_js_server/util/command_class/multilevel_sensor.py index ab507fffc..9601a26ca 100644 --- a/zwave_js_server/util/command_class/multilevel_sensor.py +++ b/zwave_js_server/util/command_class/multilevel_sensor.py @@ -7,14 +7,14 @@ MultilevelSensorType, MULTILEVEL_SENSOR_TYPE_TO_SCALE_MAP, ) -from ...exceptions import ValueHasInvalidCommandClass, UnknownValueData +from ...exceptions import InvalidCommandClass, UnknownValueData from ...model.value import Value def get_multilevel_sensor_type(value: Value) -> MultilevelSensorType: """Get the MultilevelSensorType for a given value.""" if value.command_class != CommandClass.SENSOR_MULTILEVEL: - raise ValueHasInvalidCommandClass(value, CommandClass.SENSOR_MULTILEVEL) + raise InvalidCommandClass(value, CommandClass.SENSOR_MULTILEVEL) try: return MultilevelSensorType(value.metadata.cc_specific[CC_SPECIFIC_SENSOR_TYPE]) except ValueError: