Add zwave_js/get_any_firmware_update_progress WS cmd (#73905)

This commit is contained in:
Raman Gupta 2022-06-23 20:13:37 -04:00 committed by GitHub
parent dc0ea6fd55
commit 768e53ac2d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 98 additions and 0 deletions

View File

@ -421,6 +421,9 @@ def async_register_api(hass: HomeAssistant) -> None:
websocket_api.async_register_command(
hass, websocket_get_firmware_update_capabilities
)
websocket_api.async_register_command(
hass, websocket_get_any_firmware_update_progress
)
websocket_api.async_register_command(hass, websocket_check_for_config_updates)
websocket_api.async_register_command(hass, websocket_install_config_update)
websocket_api.async_register_command(
@ -1989,6 +1992,30 @@ async def websocket_get_firmware_update_capabilities(
connection.send_result(msg[ID], capabilities.to_dict())
@websocket_api.require_admin
@websocket_api.websocket_command(
{
vol.Required(TYPE): "zwave_js/get_any_firmware_update_progress",
vol.Required(ENTRY_ID): str,
}
)
@websocket_api.async_response
@async_handle_failed_command
@async_get_entry
async def websocket_get_any_firmware_update_progress(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
entry: ConfigEntry,
client: Client,
driver: Driver,
) -> None:
"""Get whether any firmware updates are in progress."""
connection.send_result(
msg[ID], await driver.controller.async_get_any_firmware_update_progress()
)
class FirmwareUploadView(HomeAssistantView):
"""View to upload firmware."""

View File

@ -3723,6 +3723,77 @@ async def test_get_firmware_update_capabilities(
assert msg["error"]["code"] == ERR_NOT_FOUND
async def test_get_any_firmware_update_progress(
hass, client, integration, hass_ws_client
):
"""Test that the get_any_firmware_update_progress WS API call works."""
entry = integration
ws_client = await hass_ws_client(hass)
client.async_send_command.return_value = {"progress": True}
await ws_client.send_json(
{
ID: 1,
TYPE: "zwave_js/get_any_firmware_update_progress",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert msg["success"]
assert msg["result"]
assert len(client.async_send_command.call_args_list) == 1
args = client.async_send_command.call_args[0][0]
assert args["command"] == "controller.get_any_firmware_update_progress"
# Test FailedZWaveCommand is caught
with patch(
"zwave_js_server.model.controller.Controller.async_get_any_firmware_update_progress",
side_effect=FailedZWaveCommand("failed_command", 1, "error message"),
):
await ws_client.send_json(
{
ID: 2,
TYPE: "zwave_js/get_any_firmware_update_progress",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == "zwave_error"
assert msg["error"]["message"] == "Z-Wave error 1: error message"
# 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: 3,
TYPE: "zwave_js/get_any_firmware_update_progress",
ENTRY_ID: entry.entry_id,
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_LOADED
# Test sending command with improper device ID fails
await ws_client.send_json(
{
ID: 4,
TYPE: "zwave_js/get_any_firmware_update_progress",
ENTRY_ID: "invalid_entry",
}
)
msg = await ws_client.receive_json()
assert not msg["success"]
assert msg["error"]["code"] == ERR_NOT_FOUND
async def test_check_for_config_updates(hass, client, integration, hass_ws_client):
"""Test that the check_for_config_updates WS API call works."""
entry = integration