393 lines
13 KiB
Python
393 lines
13 KiB
Python
"""Test node utility functions."""
|
|
|
|
import copy
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from zwave_js_server.const import CommandClass, CommandStatus
|
|
from zwave_js_server.exceptions import (
|
|
BulkSetConfigParameterFailed,
|
|
InvalidNewValue,
|
|
NotFoundError,
|
|
SetValueFailed,
|
|
ValueTypeError,
|
|
)
|
|
from zwave_js_server.model.node import Node
|
|
from zwave_js_server.model.value import ConfigurationValue
|
|
from zwave_js_server.util.node import (
|
|
async_bulk_set_partial_config_parameters,
|
|
async_set_config_parameter,
|
|
dump_node_state,
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("endpoint", [0, 1])
|
|
async def test_configuration_parameter_values(
|
|
endpoint,
|
|
client,
|
|
climate_radio_thermostat_ct100_plus_state,
|
|
inovelli_switch_state,
|
|
uuid4,
|
|
mock_command,
|
|
):
|
|
"""Test node methods to get and set configuration parameter values."""
|
|
node_state = copy.deepcopy(climate_radio_thermostat_ct100_plus_state)
|
|
node_2_state = copy.deepcopy(inovelli_switch_state)
|
|
# Put all config parameters on endpoint we are testing
|
|
for state in (node_state, node_2_state):
|
|
for value in state["values"]:
|
|
if (
|
|
value["commandClass"] == CommandClass.CONFIGURATION
|
|
and value["endpoint"] == 0
|
|
):
|
|
value["endpoint"] = endpoint
|
|
|
|
node: Node = Node(client, node_state)
|
|
node_2: Node = Node(client, node_2_state)
|
|
ack_commands = mock_command(
|
|
{"command": "node.set_value", "nodeId": node.node_id},
|
|
{"result": {"status": 255}},
|
|
)
|
|
|
|
assert node.node_id == 13
|
|
config_values = node.get_configuration_values()
|
|
assert len(config_values) == 12
|
|
|
|
assert node_2.node_id == 31
|
|
config_values_2 = node_2.get_configuration_values()
|
|
assert len(config_values_2) == 15
|
|
|
|
for value in config_values.values():
|
|
assert isinstance(value, ConfigurationValue)
|
|
|
|
# Test setting a configuration parameter that has no metadata
|
|
with pytest.raises(NotImplementedError):
|
|
await async_set_config_parameter(node, 1, 2, endpoint=endpoint)
|
|
|
|
# Test setting a manual entry configuration parameter with a valid value
|
|
ack_commands_2 = mock_command(
|
|
{"command": "node.set_value", "nodeId": node_2.node_id},
|
|
{"result": {"status": 255}},
|
|
)
|
|
|
|
zwave_value, cmd_status = await async_set_config_parameter(
|
|
node_2, 190, 8, 255, endpoint=endpoint
|
|
)
|
|
assert isinstance(zwave_value, ConfigurationValue)
|
|
assert cmd_status.status == CommandStatus.ACCEPTED
|
|
|
|
value = node_2.values[f"31-112-{endpoint}-8-255"]
|
|
assert len(ack_commands_2) == 1
|
|
assert ack_commands_2[0] == {
|
|
"command": "node.set_value",
|
|
"nodeId": node_2.node_id,
|
|
"valueId": {
|
|
"commandClass": 112,
|
|
"endpoint": endpoint,
|
|
"property": 8,
|
|
"propertyKey": 255,
|
|
},
|
|
"value": 190,
|
|
"messageId": uuid4,
|
|
}
|
|
|
|
zwave_value, cmd_status = await async_set_config_parameter(
|
|
node_2, "Blue", 8, 255, endpoint=endpoint
|
|
)
|
|
assert isinstance(zwave_value, ConfigurationValue)
|
|
assert cmd_status.status == CommandStatus.ACCEPTED
|
|
|
|
value = node_2.values[f"31-112-{endpoint}-8-255"]
|
|
assert len(ack_commands_2) == 2
|
|
assert ack_commands_2[1] == {
|
|
"command": "node.set_value",
|
|
"nodeId": node_2.node_id,
|
|
"valueId": {
|
|
"commandClass": 112,
|
|
"endpoint": endpoint,
|
|
"property": 8,
|
|
"propertyKey": 255,
|
|
},
|
|
"value": 170,
|
|
"messageId": uuid4,
|
|
}
|
|
|
|
# Test configuration parameter not found when using an invalid property name
|
|
with pytest.raises(NotFoundError):
|
|
await async_set_config_parameter(
|
|
node, 5, "fake configuration parameter name", endpoint=endpoint
|
|
)
|
|
|
|
# Test using an invalid state label to set a value
|
|
with pytest.raises(InvalidNewValue):
|
|
await async_set_config_parameter(node, "fake state label", 1, endpoint=endpoint)
|
|
|
|
# Test configuration parameter not found when property key is invalid
|
|
with pytest.raises(NotFoundError):
|
|
await async_set_config_parameter(node, 1, 1, property_key=1, endpoint=endpoint)
|
|
|
|
# Test setting a configuration parameter by state label and property name
|
|
zwave_value, cmd_status = await async_set_config_parameter(
|
|
node, "2.0\u00b0 F", "Temperature Reporting Threshold", endpoint=endpoint
|
|
)
|
|
assert isinstance(zwave_value, ConfigurationValue)
|
|
assert cmd_status.status == CommandStatus.ACCEPTED
|
|
|
|
value = node.values[f"13-112-{endpoint}-1"]
|
|
assert len(ack_commands) == 3
|
|
assert ack_commands[2] == {
|
|
"command": "node.set_value",
|
|
"nodeId": node.node_id,
|
|
"valueId": {
|
|
"commandClass": 112,
|
|
"endpoint": endpoint,
|
|
"property": 1,
|
|
},
|
|
"value": 4,
|
|
"messageId": uuid4,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("endpoint", [0, 1])
|
|
async def test_bulk_set_partial_config_parameters(
|
|
endpoint, client, multisensor_6_state, uuid4, mock_command
|
|
):
|
|
"""Test bulk setting partial config parameters."""
|
|
node_state = copy.deepcopy(multisensor_6_state)
|
|
# Put all config parameters on endpoint we are testing
|
|
for value in node_state["values"]:
|
|
if (
|
|
value["commandClass"] == CommandClass.CONFIGURATION
|
|
and value["endpoint"] == 0
|
|
):
|
|
value["endpoint"] = endpoint
|
|
node: Node = Node(client, node_state)
|
|
ack_commands = mock_command(
|
|
{"command": "node.set_value", "nodeId": node.node_id},
|
|
{"result": {"status": 255}},
|
|
)
|
|
cmd_status = await async_bulk_set_partial_config_parameters(
|
|
node, 101, 241, endpoint=endpoint
|
|
)
|
|
assert cmd_status.status == CommandStatus.QUEUED
|
|
assert len(ack_commands) == 1
|
|
assert ack_commands[0] == {
|
|
"command": "node.set_value",
|
|
"nodeId": node.node_id,
|
|
"valueId": {
|
|
"commandClass": CommandClass.CONFIGURATION.value,
|
|
"endpoint": endpoint,
|
|
"property": 101,
|
|
},
|
|
"value": 241,
|
|
"messageId": uuid4,
|
|
}
|
|
|
|
cmd_status = await async_bulk_set_partial_config_parameters(
|
|
node, 101, {128: 1, 64: 1, 32: 1, 16: 1, 1: 1}, endpoint=endpoint
|
|
)
|
|
assert cmd_status.status == CommandStatus.QUEUED
|
|
assert len(ack_commands) == 2
|
|
assert ack_commands[1] == {
|
|
"command": "node.set_value",
|
|
"nodeId": node.node_id,
|
|
"valueId": {
|
|
"commandClass": CommandClass.CONFIGURATION.value,
|
|
"endpoint": endpoint,
|
|
"property": 101,
|
|
},
|
|
"value": 241,
|
|
"messageId": uuid4,
|
|
}
|
|
|
|
# Only set some values so we use cached values for the rest
|
|
cmd_status = await async_bulk_set_partial_config_parameters(
|
|
node, 101, {64: 1, 32: 1, 16: 1, 1: 1}, endpoint=endpoint
|
|
)
|
|
assert cmd_status.status == CommandStatus.QUEUED
|
|
assert len(ack_commands) == 3
|
|
assert ack_commands[2] == {
|
|
"command": "node.set_value",
|
|
"nodeId": node.node_id,
|
|
"valueId": {
|
|
"commandClass": CommandClass.CONFIGURATION.value,
|
|
"endpoint": endpoint,
|
|
"property": 101,
|
|
},
|
|
"value": 241,
|
|
"messageId": uuid4,
|
|
}
|
|
|
|
# Use property key names instead of bitmasks for dict key
|
|
cmd_status = await async_bulk_set_partial_config_parameters(
|
|
node,
|
|
101,
|
|
{
|
|
"Group 1: Send humidity reports": 1,
|
|
"Group 1: Send temperature reports": 1,
|
|
"Group 1: Send ultraviolet reports": 1,
|
|
"Group 1: Send battery reports": 1,
|
|
},
|
|
endpoint=endpoint,
|
|
)
|
|
assert cmd_status.status == CommandStatus.QUEUED
|
|
assert len(ack_commands) == 4
|
|
assert ack_commands[3] == {
|
|
"command": "node.set_value",
|
|
"nodeId": node.node_id,
|
|
"valueId": {
|
|
"commandClass": CommandClass.CONFIGURATION.value,
|
|
"endpoint": endpoint,
|
|
"property": 101,
|
|
},
|
|
"value": 241,
|
|
"messageId": uuid4,
|
|
}
|
|
|
|
# Use an invalid property
|
|
with pytest.raises(NotFoundError):
|
|
await async_bulk_set_partial_config_parameters(node, 999, 99, endpoint=endpoint)
|
|
|
|
# use an invalid bitmask
|
|
with pytest.raises(NotFoundError):
|
|
await async_bulk_set_partial_config_parameters(
|
|
node, 101, {128: 1, 64: 1, 32: 1, 16: 1, 2: 1}, endpoint=endpoint
|
|
)
|
|
|
|
# use an invalid property name
|
|
with pytest.raises(NotFoundError):
|
|
await async_bulk_set_partial_config_parameters(
|
|
node, 101, {"Invalid property name": 1}, endpoint=endpoint
|
|
)
|
|
|
|
# use an invalid state value
|
|
with pytest.raises(BulkSetConfigParameterFailed):
|
|
await async_bulk_set_partial_config_parameters(
|
|
node, 101, {128: "fake state", 64: 1, 32: 1, 16: 1, 1: 1}, endpoint=endpoint
|
|
)
|
|
|
|
# Try to bulkset a property that isn't broken into partials with a dictionary
|
|
with pytest.raises(ValueTypeError):
|
|
await async_bulk_set_partial_config_parameters(
|
|
node, 252, {1: 1}, endpoint=endpoint
|
|
)
|
|
|
|
# Try to bulkset a property that isn't broken into partials, it should fall back to
|
|
# async_set_config_parameter
|
|
with patch("zwave_js_server.util.node.async_set_config_parameter") as mock_cmd:
|
|
mock_cmd.return_value = (None, None)
|
|
await async_bulk_set_partial_config_parameters(node, 252, 1, endpoint=endpoint)
|
|
mock_cmd.assert_called_once()
|
|
|
|
|
|
@pytest.mark.parametrize("endpoint", [0, 1])
|
|
async def test_bulk_set_with_full_and_partial_parameters(
|
|
endpoint, client, partial_and_full_parameter_state, uuid4, mock_command
|
|
):
|
|
"""Test bulk setting config parameters when state has full and partial values."""
|
|
node_state = copy.deepcopy(partial_and_full_parameter_state)
|
|
# Put all config parameters on endpoint we are testing
|
|
for value in node_state["values"]:
|
|
if (
|
|
value["commandClass"] == CommandClass.CONFIGURATION
|
|
and value["endpoint"] == 0
|
|
):
|
|
value["endpoint"] = endpoint
|
|
node: Node = Node(client, node_state)
|
|
ack_commands = mock_command(
|
|
{"command": "node.set_value", "nodeId": node.node_id},
|
|
{"result": {"status": 255}},
|
|
)
|
|
|
|
cmd_status = await async_bulk_set_partial_config_parameters(
|
|
node, 8, 34867929, endpoint=endpoint
|
|
)
|
|
|
|
assert cmd_status.status == CommandStatus.ACCEPTED
|
|
assert len(ack_commands) == 1
|
|
assert ack_commands[0] == {
|
|
"command": "node.set_value",
|
|
"nodeId": node.node_id,
|
|
"valueId": {
|
|
"commandClass": CommandClass.CONFIGURATION.value,
|
|
"endpoint": endpoint,
|
|
"property": 8,
|
|
},
|
|
"value": 34867929,
|
|
"messageId": uuid4,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("endpoint", [0, 1])
|
|
async def test_failures(endpoint, client, multisensor_6_state, mock_command):
|
|
"""Test setting config parameter failures."""
|
|
node_state = copy.deepcopy(multisensor_6_state)
|
|
# Put all config parameters on endpoint we are testing
|
|
for value in node_state["values"]:
|
|
if (
|
|
value["commandClass"] == CommandClass.CONFIGURATION
|
|
and value["endpoint"] == 0
|
|
):
|
|
value["endpoint"] = endpoint
|
|
node: Node = Node(client, node_state)
|
|
# We need the node to be alive so we wait for a response
|
|
node.handle_alive(node)
|
|
|
|
mock_command(
|
|
{"command": "node.set_value", "nodeId": node.node_id},
|
|
{"result": {"status": 0}},
|
|
)
|
|
|
|
with pytest.raises(SetValueFailed):
|
|
await async_bulk_set_partial_config_parameters(
|
|
node, 101, {64: 1, 32: 1, 16: 1, 1: 1}, endpoint=endpoint
|
|
)
|
|
|
|
with pytest.raises(SetValueFailed):
|
|
await async_set_config_parameter(node, 1, 101, 64, endpoint=endpoint)
|
|
|
|
|
|
@pytest.mark.parametrize("endpoint", [0, 1])
|
|
async def test_returned_values(endpoint, client, multisensor_6_state, mock_command):
|
|
"""Test returned values from setting config parameters."""
|
|
node_state = copy.deepcopy(multisensor_6_state)
|
|
# Put all config parameters on endpoint we are testing
|
|
for value in node_state["values"]:
|
|
if (
|
|
value["commandClass"] == CommandClass.CONFIGURATION
|
|
and value["endpoint"] == 0
|
|
):
|
|
value["endpoint"] = endpoint
|
|
node: Node = Node(client, node_state)
|
|
# We need the node to be alive so we wait for a response
|
|
node.handle_alive(node)
|
|
|
|
mock_command(
|
|
{"command": "node.set_value", "nodeId": node.node_id},
|
|
{"result": {"status": 255}},
|
|
)
|
|
|
|
cmd_status = await async_bulk_set_partial_config_parameters(
|
|
node, 101, {64: 1, 32: 1, 16: 1, 1: 1}, endpoint=endpoint
|
|
)
|
|
assert cmd_status.status == CommandStatus.ACCEPTED
|
|
|
|
zwave_value, cmd_status = await async_set_config_parameter(
|
|
node, 1, 101, 64, endpoint=endpoint
|
|
)
|
|
assert isinstance(zwave_value, ConfigurationValue)
|
|
assert cmd_status.status == CommandStatus.ACCEPTED
|
|
|
|
|
|
async def test_dump_node_state(inovelli_switch, inovelli_switch_state):
|
|
"""Test dumping the node state."""
|
|
assert inovelli_switch_state != inovelli_switch.data
|
|
assert "values" not in inovelli_switch.data
|
|
assert "endpoints" not in inovelli_switch.data
|
|
dump_state = dump_node_state(inovelli_switch)
|
|
for val in dump_state["values"].values():
|
|
assert val in inovelli_switch_state["values"]
|
|
for endpoint in dump_state["endpoints"].values():
|
|
assert endpoint in inovelli_switch_state["endpoints"]
|