diff --git a/homeassistant/components/zwave_js/api.py b/homeassistant/components/zwave_js/api.py index 6b6286b78f4..ca25088a642 100644 --- a/homeassistant/components/zwave_js/api.py +++ b/homeassistant/components/zwave_js/api.py @@ -414,6 +414,7 @@ def async_register_api(hass: HomeAssistant) -> None: ) websocket_api.async_register_command(hass, websocket_data_collection_status) websocket_api.async_register_command(hass, websocket_abort_firmware_update) + websocket_api.async_register_command(hass, websocket_get_firmware_update_progress) websocket_api.async_register_command( hass, websocket_subscribe_firmware_update_status ) @@ -1867,6 +1868,26 @@ async def websocket_abort_firmware_update( connection.send_result(msg[ID]) +@websocket_api.require_admin +@websocket_api.websocket_command( + { + vol.Required(TYPE): "zwave_js/get_firmware_update_progress", + vol.Required(DEVICE_ID): str, + } +) +@websocket_api.async_response +@async_handle_failed_command +@async_get_node +async def websocket_get_firmware_update_progress( + hass: HomeAssistant, + connection: ActiveConnection, + msg: dict, + node: Node, +) -> None: + """Get whether firmware update is in progress.""" + connection.send_result(msg[ID], await node.async_get_firmware_update_progress()) + + def _get_firmware_update_progress_dict( progress: FirmwareUpdateProgress, ) -> dict[str, int]: diff --git a/tests/components/zwave_js/test_api.py b/tests/components/zwave_js/test_api.py index 337c74e955b..1ed125cc43a 100644 --- a/tests/components/zwave_js/test_api.py +++ b/tests/components/zwave_js/test_api.py @@ -3437,6 +3437,66 @@ async def test_abort_firmware_update( assert msg["error"]["code"] == ERR_NOT_FOUND +async def test_get_firmware_update_progress( + hass, client, multisensor_6, integration, hass_ws_client +): + """Test that the get_firmware_update_progress WS API call works.""" + entry = integration + ws_client = await hass_ws_client(hass) + device = get_device(hass, multisensor_6) + + client.async_send_command.return_value = {"progress": True} + await ws_client.send_json( + { + ID: 1, + TYPE: "zwave_js/get_firmware_update_progress", + DEVICE_ID: device.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"] == "node.get_firmware_update_progress" + assert args["nodeId"] == multisensor_6.node_id + + # Test FailedZWaveCommand is caught + with patch( + "zwave_js_server.model.node.Node.async_get_firmware_update_progress", + side_effect=FailedZWaveCommand("failed_command", 1, "error message"), + ): + await ws_client.send_json( + { + ID: 2, + TYPE: "zwave_js/get_firmware_update_progress", + DEVICE_ID: device.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_firmware_update_progress", + DEVICE_ID: device.id, + } + ) + msg = await ws_client.receive_json() + + assert not msg["success"] + assert msg["error"]["code"] == ERR_NOT_LOADED + + async def test_subscribe_firmware_update_status( hass, multisensor_6, integration, client, hass_ws_client ):