Get/Set custom config parameter for zwave_js node (#129332)
* Get/Set custom config parameter for zwave_js node * add tests * handle errors on set * test FailedCommandpull/130133/head^2
Petar Petrov
GitHub
parent
1f32e02ba2
commit
3eab0b704e
|
|
@ -56,6 +56,7 @@ from zwave_js_server.model.utils import (
|
|||
async_parse_qr_code_string,
|
||||
async_try_parse_dsk_from_qr_code_string,
|
||||
)
|
||||
from zwave_js_server.model.value import ConfigurationValueFormat
|
||||
from zwave_js_server.util.node import async_set_config_parameter
|
||||
|
||||
from homeassistant.components import websocket_api
|
||||
|
|
@ -106,6 +107,8 @@ PROPERTY = "property"
|
|||
PROPERTY_KEY = "property_key"
|
||||
ENDPOINT = "endpoint"
|
||||
VALUE = "value"
|
||||
VALUE_SIZE = "value_size"
|
||||
VALUE_FORMAT = "value_format"
|
||||
|
||||
# constants for log config commands
|
||||
CONFIG = "config"
|
||||
|
|
@ -416,6 +419,8 @@ def async_register_api(hass: HomeAssistant) -> None:
|
|||
websocket_api.async_register_command(hass, websocket_rebuild_node_routes)
|
||||
websocket_api.async_register_command(hass, websocket_set_config_parameter)
|
||||
websocket_api.async_register_command(hass, websocket_get_config_parameters)
|
||||
websocket_api.async_register_command(hass, websocket_get_raw_config_parameter)
|
||||
websocket_api.async_register_command(hass, websocket_set_raw_config_parameter)
|
||||
websocket_api.async_register_command(hass, websocket_subscribe_log_updates)
|
||||
websocket_api.async_register_command(hass, websocket_update_log_config)
|
||||
websocket_api.async_register_command(hass, websocket_get_log_config)
|
||||
|
|
@ -1760,6 +1765,72 @@ async def websocket_get_config_parameters(
|
|||
)
|
||||
|
||||
|
||||
@websocket_api.require_admin
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required(TYPE): "zwave_js/set_raw_config_parameter",
|
||||
vol.Required(DEVICE_ID): str,
|
||||
vol.Required(PROPERTY): int,
|
||||
vol.Required(VALUE): int,
|
||||
vol.Required(VALUE_SIZE): vol.All(vol.Coerce(int), vol.Range(min=1, max=4)),
|
||||
vol.Required(VALUE_FORMAT): vol.Coerce(ConfigurationValueFormat),
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
@async_handle_failed_command
|
||||
@async_get_node
|
||||
async def websocket_set_raw_config_parameter(
|
||||
hass: HomeAssistant,
|
||||
connection: ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
node: Node,
|
||||
) -> None:
|
||||
"""Set a custom config parameter value for a Z-Wave node."""
|
||||
result = await node.async_set_raw_config_parameter_value(
|
||||
msg[VALUE],
|
||||
msg[PROPERTY],
|
||||
value_size=msg[VALUE_SIZE],
|
||||
value_format=msg[VALUE_FORMAT],
|
||||
)
|
||||
|
||||
connection.send_result(
|
||||
msg[ID],
|
||||
{
|
||||
STATUS: result.status,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@websocket_api.require_admin
|
||||
@websocket_api.websocket_command(
|
||||
{
|
||||
vol.Required(TYPE): "zwave_js/get_raw_config_parameter",
|
||||
vol.Required(DEVICE_ID): str,
|
||||
vol.Required(PROPERTY): int,
|
||||
}
|
||||
)
|
||||
@websocket_api.async_response
|
||||
@async_handle_failed_command
|
||||
@async_get_node
|
||||
async def websocket_get_raw_config_parameter(
|
||||
hass: HomeAssistant,
|
||||
connection: ActiveConnection,
|
||||
msg: dict[str, Any],
|
||||
node: Node,
|
||||
) -> None:
|
||||
"""Get a custom config parameter value for a Z-Wave node."""
|
||||
value = await node.async_get_raw_config_parameter_value(
|
||||
msg[PROPERTY],
|
||||
)
|
||||
|
||||
connection.send_result(
|
||||
msg[ID],
|
||||
{
|
||||
VALUE: value,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def filename_is_present_if_logging_to_file(obj: dict) -> dict:
|
||||
"""Validate that filename is provided if log_to_file is True."""
|
||||
if obj.get(LOG_TO_FILE, False) and FILENAME not in obj:
|
||||
|
|
|
|||
|
|
@ -78,6 +78,8 @@ from homeassistant.components.zwave_js.api import (
|
|||
TYPE,
|
||||
UUID,
|
||||
VALUE,
|
||||
VALUE_FORMAT,
|
||||
VALUE_SIZE,
|
||||
VERSION,
|
||||
)
|
||||
from homeassistant.components.zwave_js.const import (
|
||||
|
|
@ -3137,6 +3139,180 @@ async def test_get_config_parameters(
|
|||
assert msg["error"]["code"] == ERR_NOT_LOADED
|
||||
|
||||
|
||||
async def test_set_raw_config_parameter(
|
||||
hass: HomeAssistant,
|
||||
client,
|
||||
multisensor_6,
|
||||
integration,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
) -> None:
|
||||
"""Test that the set_raw_config_parameter WS API call works."""
|
||||
entry = integration
|
||||
ws_client = await hass_ws_client(hass)
|
||||
device = get_device(hass, multisensor_6)
|
||||
|
||||
# Change from async_send_command to async_send_command_no_wait
|
||||
client.async_send_command_no_wait.return_value = None
|
||||
|
||||
# Test setting a raw config parameter value
|
||||
await ws_client.send_json_auto_id(
|
||||
{
|
||||
TYPE: "zwave_js/set_raw_config_parameter",
|
||||
DEVICE_ID: device.id,
|
||||
PROPERTY: 102,
|
||||
VALUE: 1,
|
||||
VALUE_SIZE: 2,
|
||||
VALUE_FORMAT: 1,
|
||||
}
|
||||
)
|
||||
|
||||
msg = await ws_client.receive_json()
|
||||
assert msg["success"]
|
||||
assert msg["result"]["status"] == "queued"
|
||||
|
||||
assert len(client.async_send_command_no_wait.call_args_list) == 1
|
||||
args = client.async_send_command_no_wait.call_args[0][0]
|
||||
assert args["command"] == "endpoint.set_raw_config_parameter_value"
|
||||
assert args["nodeId"] == multisensor_6.node_id
|
||||
assert args["options"]["parameter"] == 102
|
||||
assert args["options"]["value"] == 1
|
||||
assert args["options"]["valueSize"] == 2
|
||||
assert args["options"]["valueFormat"] == 1
|
||||
|
||||
# Reset the mock for async_send_command_no_wait instead
|
||||
client.async_send_command_no_wait.reset_mock()
|
||||
|
||||
# Test getting non-existent node fails
|
||||
await ws_client.send_json_auto_id(
|
||||
{
|
||||
TYPE: "zwave_js/set_raw_config_parameter",
|
||||
DEVICE_ID: "fake_device",
|
||||
PROPERTY: 102,
|
||||
VALUE: 1,
|
||||
VALUE_SIZE: 2,
|
||||
VALUE_FORMAT: 1,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == ERR_NOT_FOUND
|
||||
|
||||
# Test sending command with not loaded entry fails
|
||||
await hass.config_entries.async_unload(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await ws_client.send_json_auto_id(
|
||||
{
|
||||
TYPE: "zwave_js/set_raw_config_parameter",
|
||||
DEVICE_ID: device.id,
|
||||
PROPERTY: 102,
|
||||
VALUE: 1,
|
||||
VALUE_SIZE: 2,
|
||||
VALUE_FORMAT: 1,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == ERR_NOT_LOADED
|
||||
|
||||
|
||||
async def test_get_raw_config_parameter(
|
||||
hass: HomeAssistant,
|
||||
multisensor_6,
|
||||
integration,
|
||||
client,
|
||||
hass_ws_client: WebSocketGenerator,
|
||||
) -> None:
|
||||
"""Test the get_raw_config_parameter websocket command."""
|
||||
entry = integration
|
||||
ws_client = await hass_ws_client(hass)
|
||||
device = get_device(hass, multisensor_6)
|
||||
|
||||
client.async_send_command.return_value = {"value": 1}
|
||||
|
||||
# Test getting a raw config parameter value
|
||||
await ws_client.send_json_auto_id(
|
||||
{
|
||||
TYPE: "zwave_js/get_raw_config_parameter",
|
||||
DEVICE_ID: device.id,
|
||||
PROPERTY: 102,
|
||||
}
|
||||
)
|
||||
|
||||
msg = await ws_client.receive_json()
|
||||
assert msg["success"]
|
||||
assert msg["result"]["value"] == 1
|
||||
|
||||
assert len(client.async_send_command.call_args_list) == 1
|
||||
args = client.async_send_command.call_args[0][0]
|
||||
assert args["command"] == "endpoint.get_raw_config_parameter_value"
|
||||
assert args["nodeId"] == multisensor_6.node_id
|
||||
assert args["options"]["parameter"] == 102
|
||||
|
||||
client.async_send_command.reset_mock()
|
||||
|
||||
# Test FailedZWaveCommand is caught
|
||||
with patch(
|
||||
"zwave_js_server.model.node.Node.async_get_raw_config_parameter_value",
|
||||
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
|
||||
):
|
||||
await ws_client.send_json_auto_id(
|
||||
{
|
||||
TYPE: "zwave_js/get_raw_config_parameter",
|
||||
DEVICE_ID: device.id,
|
||||
PROPERTY: 102,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == "zwave_error"
|
||||
assert msg["error"]["message"] == "zwave_error: Z-Wave error 1 - error message"
|
||||
|
||||
# Test getting non-existent node fails
|
||||
await ws_client.send_json_auto_id(
|
||||
{
|
||||
TYPE: "zwave_js/get_raw_config_parameter",
|
||||
DEVICE_ID: "fake_device",
|
||||
PROPERTY: 102,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == ERR_NOT_FOUND
|
||||
|
||||
# Test FailedCommand exception
|
||||
client.async_send_command.side_effect = FailedCommand("test", "test")
|
||||
await ws_client.send_json_auto_id(
|
||||
{
|
||||
TYPE: "zwave_js/get_raw_config_parameter",
|
||||
DEVICE_ID: device.id,
|
||||
PROPERTY: 102,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == "test"
|
||||
assert msg["error"]["message"] == "Command failed: test"
|
||||
|
||||
# Test sending command with not loaded entry fails
|
||||
await hass.config_entries.async_unload(entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
await ws_client.send_json_auto_id(
|
||||
{
|
||||
TYPE: "zwave_js/get_raw_config_parameter",
|
||||
DEVICE_ID: device.id,
|
||||
PROPERTY: 102,
|
||||
}
|
||||
)
|
||||
msg = await ws_client.receive_json()
|
||||
|
||||
assert not msg["success"]
|
||||
assert msg["error"]["code"] == ERR_NOT_LOADED
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("firmware_data", "expected_data"),
|
||||
[({"target": "1"}, {"firmware_target": 1}), ({}, {})],
|
||||
|
|
|
|||
Loading…
Reference in New Issue