evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add zwave_js WS API cmds to get node state and version info (#51396)

Browse source 
* Add zwave_js view to retrieve a node's state

* remove typehints

* Make dump views require admin

* Add version info to node level dump

* Add back typehints

* switch from list to dict

* switch from dump node view to two WS API commands

* switch to snake
This commit is contained in:
Raman Gupta 2021-06-14 16:43:51 -04:00 • committed by GitHub
parent f00f2b4ae4
commit 8705168fe6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 189 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

57
homeassistant/components/zwave_js/api.py
View file

@ -138,6 +138,7 @@ def async_register_api(hass: HomeAssistant) -> None:
"""Register all of our api endpoints."""
websocket_api.async_register_command(hass, websocket_network_status)
websocket_api.async_register_command(hass, websocket_node_status)
websocket_api.async_register_command(hass, websocket_node_state)
websocket_api.async_register_command(hass, websocket_node_metadata)
websocket_api.async_register_command(hass, websocket_ping_node)
websocket_api.async_register_command(hass, websocket_add_node)
@ -164,6 +165,7 @@ def async_register_api(hass: HomeAssistant) -> None:
hass, websocket_update_data_collection_preference
)
websocket_api.async_register_command(hass, websocket_data_collection_status)
websocket_api.async_register_command(hass, websocket_version_info)
websocket_api.async_register_command(hass, websocket_abort_firmware_update)
websocket_api.async_register_command(
hass, websocket_subscribe_firmware_update_status
@ -253,6 +255,28 @@ async def websocket_node_status(
)
@websocket_api.websocket_command(
{
vol.Required(TYPE): "zwave_js/node_state",
vol.Required(ENTRY_ID): str,
vol.Required(NODE_ID): int,
}
)
@websocket_api.async_response
@async_get_node
async def websocket_node_state(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
node: Node,
) -> None:
"""Get the state data of a Z-Wave JS node."""
connection.send_result(
msg[ID],
node.data,
)
@websocket_api.websocket_command(
{
vol.Required(TYPE): "zwave_js/node_metadata",
@ -1170,6 +1194,8 @@ class DumpView(HomeAssistantView):
async def get(self, request: web.Request, config_entry_id: str) -> web.Response:
"""Dump the state of Z-Wave."""
if not request["hass_user"].is_admin:
raise Unauthorized()
hass = request.app["hass"]
if config_entry_id not in hass.data[DOMAIN]:
@ -1188,6 +1214,35 @@ class DumpView(HomeAssistantView):
)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required(TYPE): "zwave_js/version_info",
vol.Required(ENTRY_ID): str,
},
)
@websocket_api.async_response
@async_get_entry
async def websocket_version_info(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
entry: ConfigEntry,
client: Client,
) -> None:
"""Get version info from the Z-Wave JS server."""
version_info = {
"driver_version": client.version.driver_version,
"server_version": client.version.server_version,
"min_schema_version": client.version.min_schema_version,
"max_schema_version": client.version.max_schema_version,
}
connection.send_result(
msg[ID],
version_info,
)
@websocket_api.require_admin
@websocket_api.websocket_command(
{
@ -1287,7 +1342,7 @@ class FirmwareUploadView(HomeAssistantView):
raise web_exceptions.HTTPBadRequest
entry = hass.config_entries.async_get_entry(config_entry_id)
client = hass.data[DOMAIN][config_entry_id][DATA_CLIENT]
client: Client = hass.data[DOMAIN][config_entry_id][DATA_CLIENT]
node = client.driver.controller.nodes.get(int(node_id))
if not node:
raise web_exceptions.HTTPNotFound

134
tests/components/zwave_js/test_api.py
View file

@ -119,6 +119,54 @@ async def test_node_status(hass, multisensor_6, integration, hass_ws_client):
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_node_state(hass, multisensor_6, integration, hass_ws_client):
"""Test the node_state websocket command."""
entry = integration
ws_client = await hass_ws_client(hass)
node = multisensor_6
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/node_state",
ENTRY_ID: entry.entry_id,
NODE_ID: node.node_id,
}
)
msg = await ws_client.receive_json()
assert msg["result"] == node.data
# Test getting non-existent node fails
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/node_state",
ENTRY_ID: entry.entry_id,
NODE_ID: 99999,
}
)
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(
{
ID: 5,
TYPE: "zwave_js/node_state",
ENTRY_ID: entry.entry_id,
NODE_ID: node.node_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_node_metadata(hass, wallmote_central_scene, integration, hass_ws_client):
"""Test the node metadata websocket command."""
entry = integration
@ -1304,6 +1352,57 @@ async def test_dump_view(integration, hass_client):
assert json.loads(await resp.text()) == [{"hello": "world"}, {"second": "msg"}]
async def test_version_info(hass, integration, hass_ws_client, version_state):
"""Test the HTTP dump node view."""
entry = integration
ws_client = await hass_ws_client(hass)
version_info = {
"driver_version": version_state["driverVersion"],
"server_version": version_state["serverVersion"],
"min_schema_version": 0,
"max_schema_version": 0,
}
await ws_client.send_json(
{
ID: 3,
TYPE: "zwave_js/version_info",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["result"] == version_info
# Test getting non-existent entry fails
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/version_info",
ENTRY_ID: "INVALID",
}
)
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(
{
ID: 5,
TYPE: "zwave_js/version_info",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
async def test_firmware_upload_view(
hass, multisensor_6, integration, hass_client, firmware_file
):
@ -1348,6 +1447,38 @@ async def test_firmware_upload_view_invalid_payload(
assert resp.status == 400
@pytest.mark.parametrize(
"method, url",
[("get", "/api/zwave_js/dump/{}")],
)
async def test_view_non_admin_user(
integration, hass_client, hass_admin_user, method, url
):
"""Test config entry level views for non-admin users."""
client = await hass_client()
# Verify we require admin user
hass_admin_user.groups = []
resp = await client.request(method, url.format(integration.entry_id))
assert resp.status == 401
@pytest.mark.parametrize(
"method, url",
[("post", "/api/zwave_js/firmware/upload/{}/{}")],
)
async def test_node_view_non_admin_user(
multisensor_6, integration, hass_client, hass_admin_user, method, url
):
"""Test node level views for non-admin users."""
client = await hass_client()
# Verify we require admin user
hass_admin_user.groups = []
resp = await client.request(
method, url.format(integration.entry_id, multisensor_6.node_id)
)
assert resp.status == 401
@pytest.mark.parametrize(
"method, url",
[
@ -1363,7 +1494,8 @@ async def test_view_invalid_entry_id(integration, hass_client, method, url):
@pytest.mark.parametrize(
"method, url", [("post", "/api/zwave_js/firmware/upload/{}/111")]
"method, url",
[("post", "/api/zwave_js/firmware/upload/{}/111")],
)
async def test_view_invalid_node_id(integration, hass_client, method, url):
"""Test an invalid config entry id parameter."""