"""Test the node model.""" import asyncio from copy import deepcopy from datetime import UTC, datetime import json from typing import Any from unittest.mock import AsyncMock, patch import pytest from zwave_js_server.const import ( INTERVIEW_FAILED, CommandClass, CommandStatus, NodeStatus, PowerLevel, ProtocolDataRate, ProtocolVersion, Protocols, RFRegion, SecurityClass, SupervisionStatus, Weekday, ) from zwave_js_server.const.command_class.entry_control import ( EntryControlDataType, EntryControlEventType, ) from zwave_js_server.const.command_class.multilevel_switch import ( MultilevelSwitchCommand, ) 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, RssiErrorReceived, UnwriteableValue, ) from zwave_js_server.model import endpoint as endpoint_pkg, node as node_pkg from zwave_js_server.model.node import Node from zwave_js_server.model.node.firmware import ( NodeFirmwareUpdateInfo, NodeFirmwareUpdateStatus, ) from zwave_js_server.model.node.health_check import ( LifelineHealthCheckResultDataType, RouteHealthCheckResultDataType, ) from zwave_js_server.model.node.statistics import NodeStatistics from zwave_js_server.model.value import ( ConfigurationValue, ConfigurationValueFormat, SetConfigParameterResult, get_value_id_str, ) from .. import load_fixture # pylint: disable=unused-argument FIRMWARE_UPDATE_INFO = { "version": "1.0.0", "changelog": "changelog", "channel": "stable", "files": [{"target": 0, "url": "http://example.com", "integrity": "test"}], "downgrade": True, "normalizedVersion": "1.0.0", "device": { "manufacturerId": 1, "productType": 2, "productId": 3, "firmwareVersion": "0.4.4", "rfRegion": 1, }, } def test_firmware(): """Test NodeFirmwareUpdateInfo.""" firmware_update_info = NodeFirmwareUpdateInfo.from_dict(FIRMWARE_UPDATE_INFO) assert firmware_update_info.version == "1.0.0" assert firmware_update_info.changelog == "changelog" assert firmware_update_info.channel == "stable" assert len(firmware_update_info.files) == 1 assert firmware_update_info.files[0].target == 0 assert firmware_update_info.files[0].url == "http://example.com" assert firmware_update_info.files[0].integrity == "test" assert firmware_update_info.downgrade assert firmware_update_info.normalized_version == "1.0.0" assert firmware_update_info.device.manufacturer_id == 1 assert firmware_update_info.device.product_type == 2 assert firmware_update_info.device.product_id == 3 assert firmware_update_info.device.firmware_version == "0.4.4" assert firmware_update_info.device.rf_region == RFRegion.USA assert firmware_update_info.to_dict() == FIRMWARE_UPDATE_INFO def test_from_state(client): """Test from_state method.""" state = json.loads(load_fixture("basic_dump.txt").split("\n")[0])["result"]["state"] node = node_pkg.Node(client, state["nodes"][0]) assert node.node_id == 1 assert node.index == 0 assert node.status == 4 assert node.ready is True assert node.device_class.basic.key == 2 assert node.device_class.generic.label == "Static Controller" assert node.is_listening is True assert node.is_frequent_listening is False assert node.is_routing is False assert node.max_data_rate == 100000 assert node.supported_data_rates == [40000, 100000] assert node.is_secure is False assert node.protocol is None assert node.protocol_version == ProtocolVersion.VERSION_4_5X_OR_6_0X assert node.supports_beaming is True assert node.supports_security is False assert node.zwave_plus_node_type is None assert node.zwave_plus_role_type is None assert node.manufacturer_id == 134 assert node.product_id == 90 assert node.product_type == 257 assert node.label == "ZW090" assert node.interview_attempts == 0 assert node.installer_icon is None assert node.user_icon is None assert node.firmware_version is None assert node.name is None assert node.zwave_plus_version is None assert node.location is None assert node.endpoint_count_is_dynamic is None assert node.endpoints_have_identical_capabilities is None assert node.individual_endpoint_count is None assert node.aggregated_endpoint_count is None assert node.interview_stage == "Neighbors" assert not node.is_controller_node assert not node.keep_awake assert len(node.command_classes) == 0 assert len(node.endpoints) == 1 assert node.endpoints[0].index == 0 assert node.endpoints[0].installer_icon is None assert node.endpoints[0].user_icon is None assert node.endpoints[0].command_classes == [] assert node.endpoints[0].endpoint_label is None device_class = node.endpoints[0].device_class assert device_class.basic.key == 2 assert device_class.generic.key == 2 assert device_class.specific.key == 1 stats = node.statistics assert ( stats.commands_dropped_rx == stats.commands_dropped_tx == stats.commands_rx == stats.commands_tx == stats.timeout_response == 0 ) assert node == node_pkg.Node(client, state["nodes"][0]) assert node != node.node_id assert hash(node) == hash((client.driver, node.node_id)) assert node.endpoints[0] == endpoint_pkg.Endpoint( client, node, state["nodes"][0]["endpoints"][0], {} ) assert node.endpoints[0] != node.endpoints[0].index assert hash(node.endpoints[0]) == hash((client.driver, node.node_id, 0)) assert node.last_seen is None event = Event( "statistics updated", { "source": "node", "event": "statistics updated", "nodeId": node.node_id, "statistics": { "commandsTX": 1, "commandsRX": 2, "commandsDroppedTX": 3, "commandsDroppedRX": 4, "timeoutResponse": 5, "rssi": 7, "lastSeen": "2023-07-18T15:42:34.701Z", }, }, ) node.receive_event(event) assert node.last_seen == datetime(2023, 7, 18, 15, 42, 34, 701000, UTC) async def test_last_seen(lock_schlage_be469): """Test last seen property.""" assert lock_schlage_be469.last_seen == datetime( 2023, 7, 18, 15, 42, 34, 701000, UTC ) assert ( lock_schlage_be469.last_seen == lock_schlage_be469.statistics.last_seen == datetime.fromisoformat(lock_schlage_be469.statistics.data.get("lastSeen")) ) async def test_highest_security_value(lock_schlage_be469, ring_keypad): """Test the highest_security_class property.""" assert lock_schlage_be469.highest_security_class == SecurityClass.S0_LEGACY assert ring_keypad.highest_security_class is None async def test_command_classes(endpoints_with_command_classes: Node) -> None: """Test command_classes property on endpoint.""" node = endpoints_with_command_classes assert len(node.endpoints[0].command_classes) == 17 command_class_info = node.endpoints[0].command_classes[0] assert command_class_info.id == 38 assert command_class_info.command_class == CommandClass.SWITCH_MULTILEVEL assert command_class_info.name == "Multilevel Switch" assert command_class_info.version == 2 assert command_class_info.is_secure is False assert command_class_info.to_dict() == command_class_info.data async def test_device_config( wallmote_central_scene, climate_radio_thermostat_ct100_plus ): """Test a device config.""" node: node_pkg.Node = wallmote_central_scene device_config = node.device_config assert device_config.is_embedded assert device_config.filename == ( "/usr/src/app/node_modules/@zwave-js/config/config/devices/0x0086/zw130.json" ) assert device_config.manufacturer == "AEON Labs" assert device_config.manufacturer_id == 134 assert device_config.label == "ZW130" assert device_config.description == "WallMote Quad" assert len(device_config.devices) == 3 assert device_config.devices[0].product_id == 130 assert device_config.devices[0].product_type == 2 assert device_config.firmware_version.min == "0.0" assert device_config.firmware_version.max == "255.255" assert device_config.metadata.inclusion == ( "To add the ZP3111 to the Z-Wave network (inclusion), place the Z-Wave " "primary controller into inclusion mode. Press the Program Switch of ZP3111 " "for sending the NIF. After sending NIF, Z-Wave will send the auto inclusion, " "otherwise, ZP3111 will go to sleep after 20 seconds." ) assert device_config.metadata.exclusion == ( "To remove the ZP3111 from the Z-Wave network (exclusion), place the Z-Wave " "primary controller into \u201cexclusion\u201d mode, and following its " "instruction to delete the ZP3111 to the controller. Press the Program Switch " "of ZP3111 once to be excluded." ) assert device_config.metadata.reset == ( "Remove cover to trigged tamper switch, LED flash once & send out Alarm " "Report. Press Program Switch 10 times within 10 seconds, ZP3111 will send " "the \u201cDevice Reset Locally Notification\u201d command and reset to the " "factory default. (Remark: This is to be used only in the case of primary " "controller being inoperable or otherwise unavailable.)" ) assert device_config.metadata.manual == ( "https://products.z-wavealliance.org/ProductManual/File?folder=&filename=" "MarketCertificationFiles/2479/ZP3111-5_R2_20170316.pdf" ) assert device_config.metadata.wakeup is None assert device_config.metadata.comments == [{"level": "info", "text": "test"}] assert device_config.associations == {} assert device_config.param_information == {"_map": {}} assert device_config.supports_zwave_plus is None assert climate_radio_thermostat_ct100_plus.device_config.metadata.comments == [] async def test_protocol(client, wallmote_central_scene_state): """Test protocol of a node.""" node_state = deepcopy(wallmote_central_scene_state) node_state["protocol"] = 0 node = node_pkg.Node(client, node_state) assert node.protocol is Protocols.ZWAVE async def test_endpoint_no_device_class(climate_radio_thermostat_ct100_plus): """Test endpoint without a device class.""" assert climate_radio_thermostat_ct100_plus.endpoints[0].device_class is None async def test_unknown_values(cover_qubino_shutter): """Test that values that are unknown return as None.""" node = cover_qubino_shutter assert ( "5-38-0-currentValue" in node.values and node.values["5-38-0-currentValue"].value is None ) assert ( "5-37-0-currentValue" in node.values and node.values["5-37-0-currentValue"].value is None ) async def test_device_database_url(cover_qubino_shutter): """Test that the device database URL is available.""" assert ( cover_qubino_shutter.device_database_url == "https://devices.zwave-js.io/?jumpTo=0x0159:0x0003:0x0053:0.0" ) async def test_values_without_property_key_name(multisensor_6): """Test that values with property key and without property key name can be found.""" node = multisensor_6 assert "52-112-0-101-1" in node.values assert "52-112-0-101-16" in node.values async def test_hash(climate_radio_thermostat_ct100_plus): """Test node hash.""" node = climate_radio_thermostat_ct100_plus assert hash(node) == hash((node.client.driver, node.node_id)) async def test_command_class_values(climate_radio_thermostat_ct100_plus): """Test node methods to get command class values.""" node = climate_radio_thermostat_ct100_plus assert node.node_id == 13 switch_values = node.get_command_class_values(CommandClass.SENSOR_MULTILEVEL) assert len(switch_values) == 2 with pytest.raises(UnwriteableValue): await node.async_set_value("13-112-0-2", 1) async def test_set_value(multisensor_6, uuid4, mock_command): """Test set value.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.set_value", "nodeId": node.node_id}, {"success": True}, ) value_id = "52-32-0-targetValue" assert await node.async_set_value(value_id, 42) is None assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.set_value", "nodeId": node.node_id, "valueId": {"commandClass": 32, "endpoint": 0, "property": "targetValue"}, "value": 42, "messageId": uuid4, } # Set value with options assert await node.async_set_value(value_id, 42, {"transitionDuration": 1}) is None assert len(ack_commands) == 2 assert ack_commands[1] == { "command": "node.set_value", "nodeId": node.node_id, "valueId": {"commandClass": 32, "endpoint": 0, "property": "targetValue"}, "value": 42, "options": {"transitionDuration": 1}, "messageId": uuid4, } # Set value with illegal option with pytest.raises(NotFoundError): await node.async_set_value(value_id, 42, {"fake_option": 1}) # Use invalid value with pytest.raises(NotFoundError): await node.async_set_value(f"{value_id}_fake_value", 42) async def test_set_value_node_status_change(driver, multisensor_6_state): """Test set value when node status changes.""" async def async_send_command( message: dict[str, Any], require_schema: int | None = None ) -> dict: """Send a mock command that never returns.""" block_event = asyncio.Event() await block_event.wait() with patch("zwave_js_server.client.Client", autospec=True) as client_class: client = client_class.return_value client.driver = driver client.async_send_command = AsyncMock(side_effect=async_send_command) node = node_pkg.Node(client, multisensor_6_state) # wake up node event = Event(type="wake up") node.handle_wake_up(event) task = asyncio.create_task(node.async_send_command("mock_cmd")) task_2 = asyncio.create_task(node.endpoints[0].async_send_command("mock_cmd")) await asyncio.sleep(0.01) # we are waiting for the response assert not task.done() assert not task_2.done() # node goes to sleep event = Event(type="sleep") node.handle_sleep(event) await asyncio.sleep(0.01) # we are no longer waiting for the response assert task.done() assert task.result() is None assert task_2.done() assert task_2.result() is None # mark node as alive event = Event(type="alive") node.handle_alive(event) task = asyncio.create_task(node.async_send_command("mock_cmd")) task_2 = asyncio.create_task(node.endpoints[0].async_send_command("mock_cmd")) await asyncio.sleep(0.01) # we are waiting for the response assert not task.done() assert not task_2.done() # node is marked dead event = Event(type="dead") node.handle_dead(event) await asyncio.sleep(0.01) # we are no longer waiting for the response assert task.done() assert task.result() is None assert task_2.done() assert task_2.result() is None async def test_poll_value(multisensor_6, uuid4, mock_command): """Test poll value.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.poll_value", "nodeId": node.node_id}, {"result": "something"}, ) value_id = "52-32-0-currentValue" assert await node.async_poll_value(value_id) is None assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.poll_value", "nodeId": node.node_id, "valueId": {"commandClass": 32, "endpoint": 0, "property": "currentValue"}, "messageId": uuid4, } async def test_ping(multisensor_6, uuid4, mock_command): """Test ping.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.ping", "nodeId": node.node_id}, {"responded": True}, ) assert await node.async_ping() assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.ping", "nodeId": node.node_id, "messageId": uuid4, } async def test_refresh_info(multisensor_6, uuid4, mock_command): """Test refresh info.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.refresh_info", "nodeId": node.node_id}, {}, ) assert await node.async_refresh_info() is None assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.refresh_info", "nodeId": node.node_id, "messageId": uuid4, } async def test_value_added_event(multisensor_6): """Test Node value removed event.""" node = multisensor_6 assert "52-112-0-2" in node.values event = Event( type="value removed", data={ "source": "node", "event": "value removed", "nodeId": 52, "args": { "commandClassName": "Configuration", "commandClass": 112, "endpoint": 0, "property": 2, "propertyName": "Stay Awake in Battery Mode", "metadata": { "type": "number", "readable": True, "writeable": True, "valueSize": 1, "min": 0, "max": 1, "default": 0, "format": 0, "allowManualEntry": False, "states": {"0": "Disable", "1": "Enable"}, "label": "Stay Awake in Battery Mode", "description": "Stay awake for 10 minutes at power on", "isFromConfig": True, }, "value": 0, }, }, ) node.handle_value_removed(event) assert "52-112-0-2" not in node.values async def test_get_defined_value_ids(multisensor_6, uuid4, mock_command): """Test get defined value ids.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.get_defined_value_ids", "nodeId": node.node_id}, { "valueIds": [ { "commandClassName": "Wake Up", "commandClass": 132, "endpoint": 0, "property": "wakeUpInterval", "propertyName": "wakeUpInterval", }, { "commandClassName": "Wake Up", "commandClass": 132, "endpoint": 0, "property": "controllerNodeId", "propertyName": "controllerNodeId", }, ] }, ) result = await node.async_get_defined_value_ids() assert len(result) == 2 assert result[0].command_class_name == "Wake Up" assert result[0].command_class == 132 assert result[0].endpoint == 0 assert result[0].property_ == "wakeUpInterval" assert result[0].property_name == "wakeUpInterval" assert result[1].command_class_name == "Wake Up" assert result[1].command_class == 132 assert result[1].endpoint == 0 assert result[1].property_ == "controllerNodeId" assert result[1].property_name == "controllerNodeId" assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_defined_value_ids", "nodeId": node.node_id, "messageId": uuid4, } async def test_get_value_metadata(multisensor_6, uuid4, mock_command): """Test get value metadata.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.get_value_metadata", "nodeId": node.node_id}, { "type": "any", "readable": True, "writeable": False, "label": "Node ID of the controller", "description": "Description of the value metadata", }, ) result = await node.async_get_value_metadata("52-32-0-targetValue") assert result.type == "any" assert result.readable is True assert result.writeable is False assert result.label == "Node ID of the controller" assert result.description == "Description of the value metadata" assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_value_metadata", "nodeId": node.node_id, "valueId": {"commandClass": 32, "endpoint": 0, "property": "targetValue"}, "messageId": uuid4, } ack_commands.clear() async def test_abort_firmware_update(multisensor_6, uuid4, mock_command): """Test abort firmware update.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.abort_firmware_update", "nodeId": node.node_id}, {}, ) assert await node.async_abort_firmware_update() is None assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.abort_firmware_update", "nodeId": node.node_id, "messageId": uuid4, } def test_node_inclusion(multisensor_6_state): """Emulate a node being added.""" # when a node node is added, it has minimal info first node = node_pkg.Node( None, {"nodeId": 52, "status": 1, "ready": False, "values": [], "endpoints": []} ) assert node.node_id == 52 assert node.status == 1 assert not node.ready assert len(node.values) == 0 assert node.device_config.manufacturer is None # the ready event contains a full (and complete) dump of the node, including values event = Event( "ready", { "event": "ready", "source": "node", "nodeId": node.node_id, "nodeState": multisensor_6_state, "result": [], }, ) node.receive_event(event) assert node.device_config.manufacturer == "AEON Labs" assert len(node.values) > 0 new_state = deepcopy(multisensor_6_state) new_state["values"].append( { "commandClassName": "Binary Sensor", "commandClass": 48, "endpoint": 0, "property": "test", "propertyName": "test", "metadata": { "type": "boolean", "readable": True, "writeable": False, "label": "Any", "ccSpecific": {"sensorType": 255}, }, "value": False, } ) new_state["endpoints"].append( {"nodeId": 52, "index": 1, "installerIcon": 3079, "userIcon": 3079} ) event = Event( "ready", { "event": "ready", "source": "node", "nodeId": node.node_id, "nodeState": new_state, "result": [], }, ) node.receive_event(event) assert "52-48-0-test" in node.values assert 1 in node.endpoints new_state = deepcopy(new_state) new_state["endpoints"].pop(1) event = Event( "ready", { "event": "ready", "source": "node", "nodeId": node.node_id, "nodeState": multisensor_6_state, "result": [], }, ) node.receive_event(event) assert 1 not in node.endpoints def test_node_ready_event(switch_enbrighten_zw3010_state): """Emulate a node ready event.""" # when a node node is added, it has minimal info first node = node_pkg.Node( None, {"nodeId": 2, "status": 1, "ready": False, "values": [], "endpoints": []} ) assert node.node_id == 2 assert node.status == 1 assert not node.ready assert len(node.values) == 0 assert node.device_config.manufacturer is None # the ready event contains a full (and complete) dump of the node, including values event = Event( "ready", { "event": "ready", "source": "node", "nodeId": node.node_id, "nodeState": switch_enbrighten_zw3010_state, "result": [], }, ) # This will fail if the schema is invalid node.receive_event(event) assert len(node.values) > 0 async def test_node_status_events(multisensor_6): """Test Node status events.""" node = multisensor_6 assert node.status == NodeStatus.ASLEEP # mock node wake up event event = Event(type="wake up") node.handle_wake_up(event) assert node.status == NodeStatus.AWAKE # mock node dead event event = Event(type="dead") node.handle_dead(event) assert node.status == NodeStatus.DEAD # mock node alive event event = Event(type="alive") node.handle_alive(event) assert node.status == NodeStatus.ALIVE # mock node sleep event event = Event(type="sleep") node.handle_sleep(event) assert node.status == NodeStatus.ASLEEP async def test_value_added_events(multisensor_6): """Test Node value added events for new value.""" node = multisensor_6 value_id = "52-112-0-6" # Validate that the value doesn't exist in the node state data assert value_id not in node.values event = Event( type="value added", data={ "source": "node", "event": "value added", "nodeId": 52, "args": { "commandClassName": "Configuration", "commandClass": 112, "endpoint": 0, "property": 6, "propertyName": "Stay Awake in Battery Mode", "metadata": { "type": "number", "readable": True, "writeable": True, "valueSize": 1, "min": 0, "max": 1, "default": 0, "format": 0, "allowManualEntry": False, "states": {"0": "Disable", "1": "Enable"}, "label": "Stay Awake in Battery Mode", "description": "Stay awake for 10 minutes at power on", "isFromConfig": True, }, "value": 0, }, }, ) node.handle_value_added(event) assert isinstance(event.data["value"], ConfigurationValue) assert isinstance(node.values[value_id], ConfigurationValue) # ensure that the value was added to the node's state data assert value_id in node.values async def test_value_updated_events(multisensor_6): """Test Node value updated events.""" node = multisensor_6 value_id = "52-112-0-2" # ensure that the value is in the node's state data assert value_id in node.values # assert the old value of the ZwaveValue assert (value_data := node.values[value_id].data) is not None assert value_data["value"] == node.values[value_id].value == 0 event = Event( type="value updated", data={ "source": "node", "event": "value updated", "nodeId": 52, "args": { "commandClassName": "Configuration", "commandClass": 112, "endpoint": 0, "property": 2, "propertyName": "Stay Awake in Battery Mode", "value": -1, "newValue": 1, "prevValue": 0, }, }, ) node.handle_value_updated(event) assert isinstance(event.data["value"], ConfigurationValue) assert isinstance(node.values[value_id], ConfigurationValue) # ensure that the value is in to the node's state data assert value_id in node.values # ensure that the node's state data was updated and that old keys were removed assert (value_data := node.values[value_id].data) is not None assert value_data["metadata"] assert value_data["value"] == 1 assert "newValue" not in value_data assert "prevValue" not in value_data # ensure that the value's state data was updated and that old keys were removed val = node.values[value_id] assert val.data["value"] == 1 assert val.value == 1 assert "newValue" not in val.data assert "prevValue" not in val.data async def test_value_removed_events(multisensor_6): """Test Node value removed events.""" node = multisensor_6 value_id = "52-112-0-2" event = Event( type="value removed", data={ "source": "node", "event": "value removed", "nodeId": 52, "args": { "commandClassName": "Configuration", "commandClass": 112, "endpoint": 0, "property": 2, "propertyName": "Stay Awake in Battery Mode", "prevValue": 0, }, }, ) node.handle_value_removed(event) assert isinstance(event.data["value"], ConfigurationValue) # ensure that the value was removed from the nodes value's dict assert node.values.get(value_id) is None # ensure that the value was removed from the node's state data assert value_id not in node.values async def test_value_notification(wallmote_central_scene: node_pkg.Node): """Test value notification events.""" node = wallmote_central_scene # Validate that metadata gets added to notification when it's not included event = Event( type="value notification", data={ "source": "node", "event": "value notification", "nodeId": 35, "args": { "commandClass": 91, "commandClassName": "Central Scene", "property": "scene", "propertyKey": "002", "propertyName": "scene", "propertyKeyName": "002", "ccVersion": 2, "value": 1, }, }, ) node.handle_value_notification(event) assert event.data["value_notification"].metadata.states assert event.data["value_notification"].endpoint is not None assert event.data["value_notification"].value == 1 # Let's make sure that the Value was not updated by the value notification event assert node.values["35-91-0-scene-002"].value is None # Validate that a value notification event for an unknown value gets returned as is event = Event( type="value notification", data={ "source": "node", "event": "value notification", "nodeId": 35, "args": { "commandClass": 91, "commandClassName": "Central Scene", "property": "scene", "propertyKey": "005", "propertyName": "scene", "propertyKeyName": "005", "ccVersion": 2, "value": 2, }, }, ) node.handle_value_notification(event) assert event.data["value_notification"].command_class == 91 assert event.data["value_notification"].command_class_name == "Central Scene" assert event.data["value_notification"].property_ == "scene" assert event.data["value_notification"].property_name == "scene" assert event.data["value_notification"].property_key == "005" assert event.data["value_notification"].property_key_name == "005" assert event.data["value_notification"].value == 2 async def test_metadata_updated(climate_radio_thermostat_ct100_plus: node_pkg.Node): """Test metadata updated events.""" node = climate_radio_thermostat_ct100_plus value = node.values["13-135-1-value"] assert not value.metadata.states # Validate that states becomes available on a value that doesn't have a state when # a metadata updated event with states is received event = Event( type="value notification", data={ "source": "node", "event": "metadata updated", "nodeId": 13, "args": { "commandClassName": "Indicator", "commandClass": 135, "endpoint": 1, "property": "value", "propertyName": "value", "metadata": { "type": "number", "readable": True, "writeable": True, "min": 0, "max": 255, "label": "Indicator value", "ccSpecific": {"indicatorId": 0}, "states": { "0": "Idle", "1": "Heating", "2": "Cooling", "3": "Fan Only", "4": "Pending Heat", "5": "Pending Cool", "6": "Vent/Economizer", "7": "Aux Heating", "8": "2nd Stage Heating", "9": "2nd Stage Cooling", "10": "2nd Stage Aux Heat", "11": "3rd Stage Aux Heat", }, }, "value": 0, }, }, ) node.handle_metadata_updated(event) assert value.metadata.states 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, "endpointIndex": 0, "args": { "eventType": 0, "eventTypeLabel": "a", "dataType": 0, "dataTypeLabel": "b", "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"].endpoint_idx == 0 assert event.data["notification"].event_type == EntryControlEventType.CACHING assert event.data["notification"].event_type_label == "a" assert event.data["notification"].data_type == EntryControlDataType.NONE assert event.data["notification"].data_type_label == "b" assert event.data["notification"].event_data == "test" # Validate that Notification CC notification event is received as expected event = Event( type="notification", data={ "source": "node", "event": "notification", "nodeId": 23, "endpointIndex": 0, "ccId": 113, "args": { "type": 6, "event": 5, "label": "Access Control", "eventLabel": "Keypad lock operation", "parameters": {"userId": 1}, }, }, ) node.handle_notification(event) assert event.data["notification"].command_class == CommandClass.NOTIFICATION assert event.data["notification"].node_id == 23 assert event.data["notification"].endpoint_idx == 0 assert event.data["notification"].type_ == 6 assert event.data["notification"].event == 5 assert event.data["notification"].label == "Access Control" 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, "endpointIndex": 0, "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"].endpoint_idx == 0 assert event.data["notification"].test_node_id == 1 assert event.data["notification"].status == PowerLevelTestStatus.FAILED assert event.data["notification"].acknowledged_frames == 2 # Validate that Multilevel Switch CC notification event is received as expected event = Event( type="notification", data={ "source": "node", "event": "notification", "nodeId": 23, "endpointIndex": 0, "ccId": CommandClass.SWITCH_MULTILEVEL.value, "args": {"direction": "up", "eventType": 4, "eventTypeLabel": "c"}, }, ) node.handle_notification(event) assert event.data["notification"].command_class == CommandClass.SWITCH_MULTILEVEL assert event.data["notification"].node_id == 23 assert event.data["notification"].endpoint_idx == 0 assert event.data["notification"].direction == "up" assert ( event.data["notification"].event_type == MultilevelSwitchCommand.START_LEVEL_CHANGE ) assert event.data["notification"].event_type_label == "c" # Validate that Multilevel Switch CC notification event without a direction is valid event = Event( type="notification", data={ "source": "node", "event": "notification", "nodeId": 23, "endpointIndex": 0, "ccId": CommandClass.SWITCH_MULTILEVEL.value, "args": {"eventType": 4, "eventTypeLabel": "c"}, }, ) node.handle_notification(event) assert event.data["notification"].command_class == CommandClass.SWITCH_MULTILEVEL assert event.data["notification"].node_id == 23 assert event.data["notification"].endpoint_idx == 0 assert event.data["notification"].direction is None assert ( event.data["notification"].event_type == MultilevelSwitchCommand.START_LEVEL_CHANGE ) assert event.data["notification"].event_type_label == "c" async def test_notification_unknown(lock_schlage_be469: node_pkg.Node, caplog): """Test unrecognized command class notification events.""" # Validate that an unrecognized CC notification event raises Exception node = lock_schlage_be469 event = Event( type="notification", data={ "source": "node", "event": "notification", "nodeId": 23, "ccId": 0, }, ) node.handle_notification(event) assert "notification" not in event.data async def test_entry_control_notification(ring_keypad): """Test entry control CC notification events.""" node = ring_keypad # Validate that Entry Control CC notification event is received as expected event = Event( type="notification", data={ "source": "node", "event": "notification", "nodeId": 10, "endpointIndex": 0, "ccId": 111, "args": { "eventType": 5, "eventTypeLabel": "foo", "dataType": 2, "dataTypeLabel": "bar", "eventData": "cat", }, }, ) node.handle_notification(event) assert event.data["notification"].command_class == CommandClass.ENTRY_CONTROL assert event.data["notification"].node_id == 10 assert event.data["notification"].endpoint_idx == 0 assert event.data["notification"].event_type == EntryControlEventType.ARM_AWAY assert event.data["notification"].event_type_label == "foo" assert event.data["notification"].data_type == EntryControlDataType.ASCII assert event.data["notification"].data_type_label == "bar" assert event.data["notification"].event_data == "cat" async def test_interview_events(multisensor_6): """Test Node interview events.""" node = multisensor_6 assert node.interview_stage is None assert node.ready assert not node.in_interview event = Event( type="interview started", data={ "source": "node", "event": "interview started", "nodeId": 52, }, ) node.handle_interview_started(event) assert node.interview_stage is None assert not node.ready assert not node.in_interview assert node.awaiting_manual_interview event = Event( type="interview stage completed", data={ "source": "node", "event": "interview stage completed", "nodeId": 52, "stageName": "test", }, ) node.handle_interview_stage_completed(event) assert node.interview_stage == "test" assert not node.ready assert node.in_interview event = Event( type="interview failed", data={ "source": "node", "event": "interview failed", "nodeId": 52, }, ) node.handle_interview_failed(event) assert node.interview_stage == INTERVIEW_FAILED assert not node.ready assert not node.in_interview event = Event( type="interview completed", data={ "source": "node", "event": "interview completed", "nodeId": 52, }, ) node.handle_interview_completed(event) assert node.ready assert not node.in_interview async def test_refresh_values(multisensor_6, uuid4, mock_command): """Test refresh_values and refresh_cc_values commands.""" node: node_pkg.Node = multisensor_6 ack_commands = mock_command( {"command": "node.refresh_values", "nodeId": node.node_id}, {"success": True}, ) await node.async_refresh_values() assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.refresh_values", "nodeId": node.node_id, "messageId": uuid4, } ack_commands = mock_command( { "command": "node.refresh_cc_values", "nodeId": node.node_id, "commandClass": 112, }, {"success": True}, ) await node.async_refresh_cc_values(CommandClass.CONFIGURATION) assert len(ack_commands) == 2 assert ack_commands[1] == { "command": "node.refresh_cc_values", "nodeId": node.node_id, "commandClass": 112, "messageId": uuid4, } async def test_firmware_events(wallmote_central_scene: node_pkg.Node): """Test firmware events.""" node = wallmote_central_scene assert node.firmware_update_progress is None event = Event( type="firmware update progress", data={ "source": "node", "event": "firmware update progress", "nodeId": 35, "progress": { "currentFile": 1, "totalFiles": 1, "sentFragments": 1, "totalFragments": 10, "progress": 10.0, }, }, ) node.handle_firmware_update_progress(event) progress = event.data["firmware_update_progress"] assert progress.current_file == 1 assert progress.total_files == 1 assert progress.sent_fragments == 1 assert progress.total_fragments == 10 assert progress.progress == 10.0 assert node.firmware_update_progress assert node.firmware_update_progress.current_file == 1 assert node.firmware_update_progress.total_files == 1 assert node.firmware_update_progress.sent_fragments == 1 assert node.firmware_update_progress.total_fragments == 10 assert node.firmware_update_progress.progress == 10.0 event = Event( type="firmware update finished", data={ "source": "node", "event": "firmware update finished", "nodeId": 35, "result": { "status": 255, "success": True, "waitTime": 10, "reInterview": False, }, }, ) node.handle_firmware_update_finished(event) result = event.data["firmware_update_finished"] assert result.status == NodeFirmwareUpdateStatus.OK_RESTART_PENDING assert result.success assert result.wait_time == 10 assert not result.reinterview assert node.firmware_update_progress is None async def test_value_added_value_exists(climate_radio_thermostat_ct100_plus): """Test value added event when value exists.""" node: node_pkg.Node = climate_radio_thermostat_ct100_plus value_id = f"{node.node_id}-128-1-isHigh" value = node.values.get(value_id) assert value event = Event( "value added", { "source": "node", "event": "value added", "nodeId": node.node_id, "args": { "commandClassName": "Battery", "commandClass": 128, "endpoint": 1, "property": "isHigh", "propertyName": "isHigh", "metadata": { "type": "boolean", "readable": True, "writeable": False, "label": "High battery level", }, "value": True, }, }, ) node.receive_event(event) assert value_id in node.values assert node.values[value_id] is value async def test_value_added_new_value(climate_radio_thermostat_ct100_plus): """Test value added event when new value is added.""" node: node_pkg.Node = climate_radio_thermostat_ct100_plus event = Event( "value added", { "source": "node", "event": "value added", "nodeId": node.node_id, "args": { "commandClassName": "Battery", "commandClass": 128, "endpoint": 1, "property": "isMedium", "propertyName": "isMedium", "metadata": { "type": "boolean", "readable": True, "writeable": False, "label": "Medium battery level", }, "value": True, }, }, ) node.receive_event(event) assert f"{node.node_id}-128-1-isMedium" in node.values async def test_invoke_cc_api(client, lock_schlage_be469, uuid4, mock_command): """Test endpoint.invoke_cc_api commands.""" node = lock_schlage_be469 ack_commands = mock_command( {"command": "endpoint.invoke_cc_api", "nodeId": node.node_id, "endpoint": 0}, {"response": "ok"}, ) assert ( await node.async_invoke_cc_api(CommandClass.USER_CODE, "set", 1, 1, "1234") == "ok" ) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "endpoint.invoke_cc_api", "nodeId": node.node_id, "endpoint": 0, "commandClass": 99, "methodName": "set", "args": [1, 1, "1234"], "messageId": uuid4, } assert ( await node.async_invoke_cc_api( CommandClass.USER_CODE, "set", 2, 2, "1234", wait_for_result=True ) == "ok" ) assert len(ack_commands) == 2 assert ack_commands[1] == { "command": "endpoint.invoke_cc_api", "nodeId": node.node_id, "endpoint": 0, "commandClass": 99, "methodName": "set", "args": [2, 2, "1234"], "messageId": uuid4, } with pytest.raises(NotFoundError): await node.async_invoke_cc_api(CommandClass.ANTITHEFT, "test", 1) async def test_supports_cc_api(multisensor_6, uuid4, mock_command): """Test endpoint.supports_cc_api commands.""" node = multisensor_6 ack_commands = mock_command( {"command": "endpoint.supports_cc_api", "nodeId": node.node_id, "endpoint": 0}, {"supported": True}, ) assert await node.async_supports_cc_api(CommandClass.USER_CODE) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "endpoint.supports_cc_api", "nodeId": node.node_id, "endpoint": 0, "commandClass": 99, "messageId": uuid4, } # Test that command fails when client is disconnected with patch("zwave_js_server.client.asyncio.Event.wait", return_value=True): await node.client.disconnect() with pytest.raises(FailedCommand): await node.async_supports_cc_api(CommandClass.USER_CODE) async def test_supports_cc(multisensor_6, uuid4, mock_command): """Test endpoint.supports_cc_api commands.""" node = multisensor_6 ack_commands = mock_command( {"command": "endpoint.supports_cc", "nodeId": node.node_id, "endpoint": 0}, {"supported": True}, ) assert await node.async_supports_cc(CommandClass.USER_CODE) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "endpoint.supports_cc", "nodeId": node.node_id, "endpoint": 0, "commandClass": 99, "messageId": uuid4, } # Test that command fails when client is disconnected with patch("zwave_js_server.client.asyncio.Event.wait", return_value=True): await node.client.disconnect() with pytest.raises(FailedCommand): await node.async_supports_cc(CommandClass.USER_CODE) async def test_controls_cc(multisensor_6, uuid4, mock_command): """Test endpoint.controls_cc commands.""" node = multisensor_6 ack_commands = mock_command( {"command": "endpoint.controls_cc", "nodeId": node.node_id, "endpoint": 0}, {"controlled": True}, ) assert await node.async_controls_cc(CommandClass.USER_CODE) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "endpoint.controls_cc", "nodeId": node.node_id, "endpoint": 0, "commandClass": 99, "messageId": uuid4, } # Test that command fails when client is disconnected with patch("zwave_js_server.client.asyncio.Event.wait", return_value=True): await node.client.disconnect() with pytest.raises(FailedCommand): await node.async_controls_cc(CommandClass.USER_CODE) async def test_is_cc_secure(multisensor_6, uuid4, mock_command): """Test endpoint.is_cc_secure commands.""" node = multisensor_6 ack_commands = mock_command( {"command": "endpoint.is_cc_secure", "nodeId": node.node_id, "endpoint": 0}, {"secure": True}, ) assert await node.async_is_cc_secure(CommandClass.USER_CODE) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "endpoint.is_cc_secure", "nodeId": node.node_id, "endpoint": 0, "commandClass": 99, "messageId": uuid4, } # Test that command fails when client is disconnected with patch("zwave_js_server.client.asyncio.Event.wait", return_value=True): await node.client.disconnect() with pytest.raises(FailedCommand): await node.async_is_cc_secure(CommandClass.USER_CODE) async def test_get_cc_version(multisensor_6, uuid4, mock_command): """Test endpoint.get_cc_version commands.""" node = multisensor_6 ack_commands = mock_command( {"command": "endpoint.get_cc_version", "nodeId": node.node_id, "endpoint": 0}, {"version": 1}, ) assert await node.async_get_cc_version(CommandClass.USER_CODE) == 1 assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "endpoint.get_cc_version", "nodeId": node.node_id, "endpoint": 0, "commandClass": 99, "messageId": uuid4, } # Test that command fails when client is disconnected with patch("zwave_js_server.client.asyncio.Event.wait", return_value=True): await node.client.disconnect() with pytest.raises(FailedCommand): await node.async_get_cc_version(CommandClass.USER_CODE) async def test_get_node_unsafe(multisensor_6, uuid4, mock_command): """Test endpoint.get_node_unsafe commands.""" node = multisensor_6 ack_commands = mock_command( {"command": "endpoint.get_node_unsafe", "nodeId": node.node_id, "endpoint": 0}, {"node": multisensor_6}, ) assert await node.async_get_node_unsafe() == multisensor_6 assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "endpoint.get_node_unsafe", "nodeId": node.node_id, "endpoint": 0, "messageId": uuid4, } # Test that command fails when client is disconnected with patch("zwave_js_server.client.asyncio.Event.wait", return_value=True): await node.client.disconnect() with pytest.raises(FailedCommand): await node.async_get_node_unsafe() async def test_statistics_updated( wallmote_central_scene: node_pkg.Node, multisensor_6, ring_keypad ): """Test that statistics get updated on events.""" node = wallmote_central_scene assert node.statistics.commands_rx == 0 event = Event( "statistics updated", { "source": "node", "event": "statistics updated", "nodeId": node.node_id, "statistics": { "commandsTX": 1, "commandsRX": 2, "commandsDroppedTX": 3, "commandsDroppedRX": 4, "timeoutResponse": 5, "rtt": 6, "rssi": 7, "lwr": { "protocolDataRate": 1, "repeaters": [f"{wallmote_central_scene.node_id}"], "repeaterRSSI": [1], "routeFailedBetween": [ f"{ring_keypad.node_id}", f"{multisensor_6.node_id}", ], }, "nlwr": { "protocolDataRate": 2, "repeaters": [], "repeaterRSSI": [127], "routeFailedBetween": [ f"{multisensor_6.node_id}", f"{ring_keypad.node_id}", ], }, }, }, ) node.receive_event(event) # Event should be modified with the NodeStatistics object assert "statistics_updated" in event.data event_stats: NodeStatistics = event.data["statistics_updated"] assert isinstance(event_stats, NodeStatistics) assert event_stats.commands_tx == 1 assert event_stats.commands_rx == 2 assert event_stats.commands_dropped_tx == 3 assert event_stats.commands_dropped_rx == 4 assert event_stats.timeout_response == 5 assert event_stats.rtt == 6 assert event_stats.rssi == 7 assert event_stats.lwr assert event_stats.lwr.protocol_data_rate == ProtocolDataRate.ZWAVE_9K6 assert event_stats.nlwr assert event_stats.nlwr.protocol_data_rate == ProtocolDataRate.ZWAVE_40K assert node.statistics == event_stats assert event_stats.lwr.as_dict() == { "protocol_data_rate": 1, "repeaters": [wallmote_central_scene], "repeater_rssi": [1], "rssi": None, "route_failed_between": ( ring_keypad, multisensor_6, ), } statistics_data = { "commandsTX": 1, "commandsRX": 2, "commandsDroppedTX": 3, "commandsDroppedRX": 4, "timeoutResponse": 5, } assert node.data.get("statistics") != statistics_data event = Event( "statistics updated", { "source": "node", "event": "statistics updated", "nodeId": node.node_id, "statistics": statistics_data, }, ) node.receive_event(event) # Event should be modified with the NodeStatistics object assert "statistics_updated" in event.data event_stats: NodeStatistics = event.data["statistics_updated"] assert isinstance(event_stats, NodeStatistics) assert event_stats.commands_tx == 1 assert event_stats.commands_rx == 2 assert event_stats.commands_dropped_tx == 3 assert event_stats.commands_dropped_rx == 4 assert event_stats.timeout_response == 5 assert not event_stats.rtt assert not event_stats.rssi assert not event_stats.lwr assert not event_stats.nlwr assert node.statistics == event_stats assert node.data["statistics"] == statistics_data # Test that invalid protocol data rate doesn't raise error event = Event( "statistics updated", { "source": "node", "event": "statistics updated", "nodeId": node.node_id, "statistics": { "commandsTX": 1, "commandsRX": 2, "commandsDroppedTX": 3, "commandsDroppedRX": 4, "timeoutResponse": 5, "rtt": 6, "rssi": 7, "lwr": { "protocolDataRate": 0, "repeaters": [], "repeaterRSSI": [], "routeFailedBetween": [], }, "nlwr": { "protocolDataRate": 0, "repeaters": [], "repeaterRSSI": [], "routeFailedBetween": [], }, }, }, ) node.receive_event(event) # Event should be modified with the NodeStatistics object assert "statistics_updated" in event.data event_stats: NodeStatistics = event.data["statistics_updated"] assert isinstance(event_stats, NodeStatistics) assert not event_stats.lwr assert not event_stats.nlwr async def test_statistics_updated_rssi_error( wallmote_central_scene: node_pkg.Node, multisensor_6, ring_keypad ): """Test that statistics get updated on events and rssi error is handled.""" node = wallmote_central_scene assert node.statistics.commands_rx == 0 event = Event( "statistics updated", { "source": "node", "event": "statistics updated", "nodeId": node.node_id, "statistics": { "commandsTX": 1, "commandsRX": 2, "commandsDroppedTX": 3, "commandsDroppedRX": 4, "timeoutResponse": 5, "rtt": 6, "rssi": 127, "lwr": { "protocolDataRate": 1, "repeaters": [f"{wallmote_central_scene.node_id}"], "repeaterRSSI": [1], "routeFailedBetween": [ f"{ring_keypad.node_id}", f"{multisensor_6.node_id}", ], }, "nlwr": { "protocolDataRate": 2, "repeaters": [], "repeaterRSSI": [127], "routeFailedBetween": [ f"{multisensor_6.node_id}", f"{ring_keypad.node_id}", ], }, }, }, ) node.receive_event(event) # Event should be modified with the NodeStatistics object assert "statistics_updated" in event.data event_stats: NodeStatistics = event.data["statistics_updated"] assert isinstance(event_stats, NodeStatistics) with pytest.raises(RssiErrorReceived): assert event_stats.rssi async def test_has_security_class(multisensor_6: node_pkg.Node, uuid4, mock_command): """Test node.has_security_class command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.has_security_class", "nodeId": node.node_id}, {"hasSecurityClass": True}, ) assert await node.async_has_security_class(SecurityClass.S2_AUTHENTICATED) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.has_security_class", "nodeId": node.node_id, "securityClass": SecurityClass.S2_AUTHENTICATED.value, "messageId": uuid4, } async def test_has_security_class_undefined( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.has_security_class command response is undefined.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.has_security_class", "nodeId": node.node_id}, {}, ) assert await node.async_has_security_class(SecurityClass.S2_AUTHENTICATED) is None assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.has_security_class", "nodeId": node.node_id, "securityClass": SecurityClass.S2_AUTHENTICATED.value, "messageId": uuid4, } async def test_get_highest_security_class( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.get_highest_security_class command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.get_highest_security_class", "nodeId": node.node_id}, {"highestSecurityClass": SecurityClass.S2_AUTHENTICATED.value}, ) assert ( await node.async_get_highest_security_class() == SecurityClass.S2_AUTHENTICATED ) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_highest_security_class", "nodeId": node.node_id, "messageId": uuid4, } async def test_get_highest_security_class_undefined( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.get_highest_security_class command response is undefined.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.get_highest_security_class", "nodeId": node.node_id}, {}, ) assert await node.async_get_highest_security_class() is None assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_highest_security_class", "nodeId": node.node_id, "messageId": uuid4, } async def test_test_power_level( multisensor_6: node_pkg.Node, wallmote_central_scene: node_pkg.Node, uuid4, mock_command, ): """Test node.test_powerlevel command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.test_powerlevel", "nodeId": node.node_id}, {"framesAcked": 1}, ) assert ( await node.async_test_power_level( wallmote_central_scene, PowerLevel.DBM_MINUS_1, 3 ) == 1 ) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.test_powerlevel", "nodeId": node.node_id, "testNodeId": wallmote_central_scene.node_id, "powerlevel": PowerLevel.DBM_MINUS_1.value, "testFrameCount": 3, "messageId": uuid4, } async def test_test_power_level_progress_event( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test test power level progress event.""" event = Event( "test powerlevel progress", { "source": "node", "event": "test powerlevel progress", "nodeId": multisensor_6.node_id, "acknowledged": 1, "total": 2, }, ) multisensor_6.receive_event(event) assert event.data["test_power_level_progress"] assert event.data["test_power_level_progress"].acknowledged == 1 assert event.data["test_power_level_progress"].total == 2 async def test_check_lifeline_health(multisensor_6: node_pkg.Node, uuid4, mock_command): """Test node.check_lifeline_health command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.check_lifeline_health", "nodeId": node.node_id}, { "summary": { "rating": 10, "results": [ LifelineHealthCheckResultDataType( latency=1, numNeighbors=2, failedPingsNode=3, rating=9, routeChanges=4, minPowerlevel=5, failedPingsController=6, snrMargin=7, ) ], } }, ) summary = await node.async_check_lifeline_health(1) assert summary.rating == 10 assert summary.results[0].latency == 1 assert summary.results[0].num_neighbors == 2 assert summary.results[0].failed_pings_node == 3 assert summary.results[0].rating == 9 assert summary.results[0].route_changes == 4 assert summary.results[0].min_power_level == PowerLevel.DBM_MINUS_5 assert summary.results[0].failed_pings_controller == 6 assert summary.results[0].snr_margin == 7 assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.check_lifeline_health", "nodeId": node.node_id, "rounds": 1, "messageId": uuid4, } async def test_check_lifeline_health_progress_event( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test check lifeline health progress event.""" event = Event( "check lifeline health progress", { "source": "node", "event": "check lifeline health progress", "nodeId": multisensor_6.node_id, "rounds": 1, "totalRounds": 2, "lastRating": 10, "lastResult": LifelineHealthCheckResultDataType( latency=1, numNeighbors=2, failedPingsNode=3, rating=9, routeChanges=4, minPowerlevel=5, failedPingsController=6, snrMargin=7, ), }, ) multisensor_6.receive_event(event) assert event.data["check_lifeline_health_progress"] assert event.data["check_lifeline_health_progress"].rounds == 1 assert event.data["check_lifeline_health_progress"].total_rounds == 2 assert event.data["check_lifeline_health_progress"].last_rating == 10 assert event.data["check_lifeline_health_progress"].last_result.latency == 1 async def test_check_route_health( multisensor_6: node_pkg.Node, wallmote_central_scene: node_pkg.Node, uuid4, mock_command, ): """Test node.check_route_health command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.check_route_health", "nodeId": node.node_id}, { "summary": { "rating": 10, "results": [ RouteHealthCheckResultDataType( numNeighbors=1, rating=10, failedPingsToSource=2, failedPingsToTarget=3, minPowerlevelSource=4, minPowerlevelTarget=5, ) ], } }, ) summary = await node.async_check_route_health(wallmote_central_scene, 1) assert summary.rating == 10 assert summary.results[0].num_neighbors == 1 assert summary.results[0].rating == 10 assert summary.results[0].failed_pings_to_source == 2 assert summary.results[0].failed_pings_to_target == 3 assert summary.results[0].min_power_level_source == PowerLevel.DBM_MINUS_4 assert summary.results[0].min_power_level_target == PowerLevel.DBM_MINUS_5 assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.check_route_health", "nodeId": node.node_id, "targetNodeId": wallmote_central_scene.node_id, "rounds": 1, "messageId": uuid4, } async def test_check_route_health_progress_event( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test check route health progress event.""" event = Event( "check route health progress", { "source": "node", "event": "check route health progress", "nodeId": multisensor_6.node_id, "rounds": 1, "totalRounds": 2, "lastRating": 10, "lastResult": RouteHealthCheckResultDataType( numNeighbors=1, rating=10, failedPingsToSource=2, failedPingsToTarget=3, minPowerlevelSource=4, minPowerlevelTarget=5, ), }, ) multisensor_6.receive_event(event) assert event.data["check_route_health_progress"] assert event.data["check_route_health_progress"].rounds == 1 assert event.data["check_route_health_progress"].total_rounds == 2 assert event.data["check_route_health_progress"].last_rating == 10 assert event.data["check_route_health_progress"].last_result.num_neighbors == 1 async def test_get_state( multisensor_6: node_pkg.Node, multisensor_6_state: node_pkg.NodeDataType, uuid4, mock_command, ): """Test node.get_state command.""" node = multisensor_6 value_id = get_value_id_str(node, 32, "currentValue", 0) # Verify original values assert node.endpoints[0].installer_icon == 3079 assert node.values[value_id].value == 255 new_state = deepcopy(multisensor_6_state) # Update endpoint 0 installer icon new_state["endpoints"][0]["installerIcon"] = 1 # Update value of {nodeId}-32-0-currentValue new_state["values"][0] = { "commandClassName": "Basic", "commandClass": 32, "endpoint": 0, "property": "currentValue", "propertyName": "currentValue", "metadata": { "type": "number", "readable": True, "writeable": False, "min": 0, "max": 99, "label": "Current value", }, "value": 0, } ack_commands = mock_command( {"command": "node.get_state", "nodeId": node.node_id}, {"state": new_state}, ) # Verify new values assert await node.async_get_state() == new_state # Verify original values are still the same assert node.endpoints[0].installer_icon != 1 assert node.values[value_id].value != 0 assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_state", "nodeId": node.node_id, "messageId": uuid4, } async def test_update_endpoints( shelly_wave_shutter_state: node_pkg.NodeDataType, shelly_wave_shutter: node_pkg.Node, ) -> None: """Test updating endpoints of a node.""" node = shelly_wave_shutter assert len(node.endpoints) == 3 for endpoint_idx, endpoint in node.endpoints.items(): assert endpoint.node_id == node.node_id assert endpoint.index == endpoint_idx for value in endpoint.values.values(): assert value.node.node_id == node.node_id assert value.endpoint == endpoint_idx node_data = deepcopy(shelly_wave_shutter_state) new_endpoints = [ endpoint_pkg.EndpointDataType( nodeId=node.node_id, index=0, commandClasses=[], ), endpoint_pkg.EndpointDataType( nodeId=node.node_id, index=1, commandClasses=[], ), endpoint_pkg.EndpointDataType( nodeId=node.node_id, index=3, commandClasses=[], ), ] node_data["endpoints"] = new_endpoints node.update(node_data) assert len(node.endpoints) == 3 for endpoint_data in new_endpoints: assert endpoint_data["index"] in node.endpoints assert node.endpoints[endpoint_data["index"]].node_id == node.node_id assert node.endpoints[endpoint_data["index"]].index == endpoint_data["index"] for endpoint_idx, endpoint in node.endpoints.items(): assert endpoint.node_id == node.node_id assert endpoint.index == endpoint_idx for value in endpoint.values.values(): assert value.node.node_id == node.node_id assert value.endpoint == endpoint_idx async def test_set_name(multisensor_6: node_pkg.Node, uuid4, mock_command): """Test node.set_name command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.set_name", "nodeId": node.node_id}, {}, ) assert node.name != "new_name" assert await node.async_set_name("new_name", False) is None assert node.name == "new_name" assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.set_name", "nodeId": node.node_id, "name": "new_name", "updateCC": False, "messageId": uuid4, } async def test_set_location(multisensor_6: node_pkg.Node, uuid4, mock_command): """Test node.set_location command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.set_location", "nodeId": node.node_id}, {}, ) assert node.location != "new_location" assert await node.async_set_location("new_location", False) is None assert node.location == "new_location" assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.set_location", "nodeId": node.node_id, "location": "new_location", "updateCC": False, "messageId": uuid4, } async def test_set_keep_awake(multisensor_6: node_pkg.Node, uuid4, mock_command): """Test node.set_keep_awake command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.set_keep_awake", "nodeId": node.node_id}, {}, ) assert node.keep_awake assert await node.async_set_keep_awake(False) is None assert node.keep_awake is False assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.set_keep_awake", "nodeId": node.node_id, "keepAwake": False, "messageId": uuid4, } async def test_get_firmware_update_capabilities( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.get_firmware_update_capabilities command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.get_firmware_update_capabilities", "nodeId": node.node_id}, { "capabilities": { "firmwareUpgradable": True, "firmwareTargets": [0], "continuesToFunction": True, "supportsActivation": True, } }, ) capabilities = await node.async_get_firmware_update_capabilities() assert capabilities.firmware_upgradable assert capabilities.firmware_targets == [0] assert capabilities.continues_to_function assert capabilities.supports_activation assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_firmware_update_capabilities", "nodeId": node.node_id, "messageId": uuid4, } assert capabilities.to_dict() == { "firmware_upgradable": True, "firmware_targets": [0], "continues_to_function": True, "supports_activation": True, } async def test_get_firmware_update_capabilities_false( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.get_firmware_update_capabilities cmd without firmware support.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.get_firmware_update_capabilities", "nodeId": node.node_id}, {"capabilities": {"firmwareUpgradable": False}}, ) capabilities = await node.async_get_firmware_update_capabilities() assert not capabilities.firmware_upgradable with pytest.raises(TypeError): assert capabilities.firmware_targets with pytest.raises(TypeError): assert capabilities.continues_to_function with pytest.raises(TypeError): assert capabilities.supports_activation assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_firmware_update_capabilities", "nodeId": node.node_id, "messageId": uuid4, } assert capabilities.to_dict() == {"firmware_upgradable": False} async def test_get_firmware_update_capabilities_string( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.get_firmware_update_capabilities cmd without firmware support.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.get_firmware_update_capabilities", "nodeId": node.node_id}, { "capabilities": { "firmwareUpgradable": True, "firmwareTargets": [0], "continuesToFunction": "unknown", "supportsActivation": "unknown", } }, ) capabilities = await node.async_get_firmware_update_capabilities() assert capabilities.firmware_upgradable assert capabilities.firmware_targets == [0] assert capabilities.continues_to_function is None assert capabilities.supports_activation is None assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_firmware_update_capabilities", "nodeId": node.node_id, "messageId": uuid4, } async def test_get_firmware_update_capabilities_cached( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.get_firmware_update_capabilities_cached command.""" node = multisensor_6 ack_commands = mock_command( { "command": "node.get_firmware_update_capabilities_cached", "nodeId": node.node_id, }, { "capabilities": { "firmwareUpgradable": True, "firmwareTargets": [0], "continuesToFunction": True, "supportsActivation": True, } }, ) capabilities = await node.async_get_firmware_update_capabilities_cached() assert capabilities.firmware_upgradable assert capabilities.firmware_targets == [0] assert capabilities.continues_to_function assert capabilities.supports_activation assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_firmware_update_capabilities_cached", "nodeId": node.node_id, "messageId": uuid4, } assert capabilities.to_dict() == { "firmware_upgradable": True, "firmware_targets": [0], "continues_to_function": True, "supports_activation": True, } async def test_is_firmware_update_in_progress( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.is_firmware_update_in_progress command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.is_firmware_update_in_progress", "nodeId": node.node_id}, {"progress": True}, ) assert await node.async_is_firmware_update_in_progress() assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.is_firmware_update_in_progress", "nodeId": node.node_id, "messageId": uuid4, } async def test_interview(multisensor_6: node_pkg.Node, uuid4, mock_command): """Test node.interview command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.interview", "nodeId": node.node_id}, {}, ) await node.async_interview() assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.interview", "nodeId": node.node_id, "messageId": uuid4, } async def test_manually_idle_notification_value( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.manually_idle_notification_value command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.manually_idle_notification_value", "nodeId": node.node_id}, {}, ) await node.async_manually_idle_notification_value( f"{node.node_id}-113-0-Home Security-Cover status" ) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.manually_idle_notification_value", "nodeId": node.node_id, "valueId": { "commandClass": 113, "endpoint": 0, "property": "Home Security", "propertyKey": "Cover status", }, "messageId": uuid4, } # Raise ValueError if the value is not for the right CommandClass with pytest.raises(ValueError): await node.async_manually_idle_notification_value(f"{node.node_id}-112-0-255") async def test_set_date_and_time_no_wait( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.set_date_and_time command without waiting.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.set_date_and_time", "nodeId": node.node_id}, {"success": True}, ) assert await node.async_set_date_and_time(datetime(2020, 1, 1, 12, 0, 0)) is None assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.set_date_and_time", "nodeId": node.node_id, "date": "2020-01-01T12:00:00", "messageId": uuid4, } async def test_set_date_and_time( climate_radio_thermostat_ct100_plus: node_pkg.Node, uuid4, mock_command ): """Test node.set_date_and_time command while waiting for response.""" node = climate_radio_thermostat_ct100_plus ack_commands = mock_command( {"command": "node.set_date_and_time", "nodeId": node.node_id}, {"success": True}, ) assert await node.async_set_date_and_time(datetime(2020, 1, 1, 12, 0, 0)) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.set_date_and_time", "nodeId": node.node_id, "date": "2020-01-01T12:00:00", "messageId": uuid4, } async def test_get_date_and_time(multisensor_6: node_pkg.Node, uuid4, mock_command): """Test node.get_date_and_time command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.get_date_and_time", "nodeId": node.node_id}, {"dateAndTime": {"hour": 1, "minute": 2, "weekday": 5}}, ) date_and_time = await node.async_get_date_and_time() assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_date_and_time", "nodeId": node.node_id, "messageId": uuid4, } assert date_and_time.hour == 1 assert date_and_time.minute == 2 assert date_and_time.weekday == Weekday.FRIDAY async def test_get_value_timestamp(multisensor_6: node_pkg.Node, uuid4, mock_command): """Test node.get_value_timestamp command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.get_value_timestamp", "nodeId": node.node_id}, {"timestamp": 1234567890}, ) val = node.values["52-32-0-targetValue"] assert await node.async_get_value_timestamp(val) == 1234567890 assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.get_value_timestamp", "nodeId": node.node_id, "valueId": {"commandClass": 32, "endpoint": 0, "property": "targetValue"}, "messageId": uuid4, } assert await node.async_get_value_timestamp("52-112-0-2") == 1234567890 assert len(ack_commands) == 2 assert ack_commands[1] == { "command": "node.get_value_timestamp", "nodeId": node.node_id, "valueId": {"commandClass": 112, "endpoint": 0, "property": 2}, "messageId": uuid4, } async def test_is_health_check_in_progress( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test node.is_health_check_in_progress command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.is_health_check_in_progress", "nodeId": node.node_id}, {"progress": True}, ) assert await node.async_is_health_check_in_progress() assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.is_health_check_in_progress", "nodeId": node.node_id, "messageId": uuid4, } async def test_abort_health_check(multisensor_6: node_pkg.Node, uuid4, mock_command): """Test node.abort_health_check command.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.abort_health_check", "nodeId": node.node_id}, {}, ) await node.async_abort_health_check() assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.abort_health_check", "nodeId": node.node_id, "messageId": uuid4, } async def test_unknown_event(multisensor_6: node_pkg.Node): """Test that an unknown event type causes an exception.""" with pytest.raises(KeyError): assert multisensor_6.receive_event(Event("unknown_event", {"source": "node"})) async def test_default_volume(multisensor_6: node_pkg.Node, uuid4, mock_command): """Test default volume.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.set_default_volume", "nodeId": node.node_id}, {}, ) assert node.default_volume is None await node.async_set_default_volume(10) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.set_default_volume", "defaultVolume": 10, "nodeId": node.node_id, "messageId": uuid4, } assert node.default_volume == 10 async def test_default_transition_duration( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test default transition duration.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.set_default_transition_duration", "nodeId": node.node_id}, {}, ) assert node.default_transition_duration is None await node.async_set_default_transition_duration(10) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.set_default_transition_duration", "defaultTransitionDuration": 10, "nodeId": node.node_id, "messageId": uuid4, } assert node.default_transition_duration == 10 async def test_has_device_config_changed( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test has device config changed.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.has_device_config_changed", "nodeId": node.node_id}, {"changed": True}, ) assert await node.async_has_device_config_changed() assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.has_device_config_changed", "nodeId": node.node_id, "messageId": uuid4, } async def test_has_device_config_changed_undefined( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test has device config changed returns undefined.""" node = multisensor_6 ack_commands = mock_command( {"command": "node.has_device_config_changed", "nodeId": node.node_id}, {}, ) assert await node.async_has_device_config_changed() is None assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "node.has_device_config_changed", "nodeId": node.node_id, "messageId": uuid4, } async def test_is_secure_none(client, multisensor_6_state): """Test is_secure when it's not included in the dump.""" node_state = deepcopy(multisensor_6_state) node_state.pop("isSecure") node = node_pkg.Node(client, node_state) assert node.is_secure is None async def test_set_raw_config_parameter_value( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test set raw config parameter value.""" node = multisensor_6 ack_commands = mock_command( {"command": "endpoint.set_raw_config_parameter_value", "nodeId": node.node_id}, {}, ) result = await node.async_set_raw_config_parameter_value(1, 101, 1) assert result == SetConfigParameterResult(CommandStatus.QUEUED) assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "endpoint.set_raw_config_parameter_value", "nodeId": node.node_id, "endpoint": 0, "parameter": 101, "bitMask": 1, "value": 1, "messageId": uuid4, } result = await node.async_set_raw_config_parameter_value(2, 0) assert result == SetConfigParameterResult(CommandStatus.QUEUED) assert len(ack_commands) == 2 assert ack_commands[1] == { "command": "endpoint.set_raw_config_parameter_value", "nodeId": node.node_id, "endpoint": 0, "parameter": 0, "value": 2, "messageId": uuid4, } # wake up node event = Event( "wake up", {"source": "node", "event": "wake up", "nodeId": node.node_id} ) node.receive_event(event) result = await node.async_set_raw_config_parameter_value( 1, 2, value_size=1, value_format=ConfigurationValueFormat.SIGNED_INTEGER ) assert result == SetConfigParameterResult(CommandStatus.ACCEPTED) assert len(ack_commands) == 3 assert ack_commands[2] == { "command": "endpoint.set_raw_config_parameter_value", "nodeId": node.node_id, "endpoint": 0, "parameter": 2, "valueSize": 1, "valueFormat": 0, "value": 1, "messageId": uuid4, } # Test failures with pytest.raises(ValueError): await node.async_set_raw_config_parameter_value(1, 101, 1, 1) with pytest.raises(ValueError): await node.async_set_raw_config_parameter_value( 1, 101, 1, value_format=ConfigurationValueFormat.SIGNED_INTEGER ) with pytest.raises(ValueError): await node.async_set_raw_config_parameter_value( 1, 101, 1, 1, ConfigurationValueFormat.SIGNED_INTEGER ) async def test_get_raw_config_parameter_value( multisensor_6: node_pkg.Node, uuid4, mock_command ): """Test get raw config parameter value.""" node = multisensor_6 ack_commands = mock_command( {"command": "endpoint.get_raw_config_parameter_value", "nodeId": node.node_id}, {"value": 1}, ) value = await node.async_get_raw_config_parameter_value(101) assert value == 1 assert len(ack_commands) == 1 assert ack_commands[0] == { "command": "endpoint.get_raw_config_parameter_value", "nodeId": node.node_id, "endpoint": 0, "parameter": 101, "messageId": uuid4, } async def test_supervision_result(inovelli_switch: node_pkg.Node, uuid4, mock_command): """Test Supervision Result.""" node = inovelli_switch mock_command( {"command": "endpoint.set_raw_config_parameter_value", "nodeId": node.node_id}, {"result": {"status": 1, "remainingDuration": "default"}}, ) result = await node.async_set_raw_config_parameter_value(1, 1) assert result.result.status is SupervisionStatus.WORKING duration = result.result.remaining_duration assert duration.unit == "default" async def test_supervision_result_invalid( inovelli_switch: node_pkg.Node, uuid4, mock_command ): """Test invalid Supervision Result.""" node = inovelli_switch mock_command( {"command": "endpoint.set_raw_config_parameter_value", "nodeId": node.node_id}, {"result": {"status": 1}}, ) with pytest.raises(ValueError): await node.async_set_raw_config_parameter_value(1, 1)