diff --git a/homeassistant/components/zwave_js/api.py b/homeassistant/components/zwave_js/api.py index 75180cfa84f..6b6286b78f4 100644 --- a/homeassistant/components/zwave_js/api.py +++ b/homeassistant/components/zwave_js/api.py @@ -417,6 +417,9 @@ def async_register_api(hass: HomeAssistant) -> None: websocket_api.async_register_command( hass, websocket_subscribe_firmware_update_status ) + websocket_api.async_register_command( + hass, websocket_get_firmware_update_capabilities + ) 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( @@ -1944,6 +1947,27 @@ async def websocket_subscribe_firmware_update_status( ) +@websocket_api.require_admin +@websocket_api.websocket_command( + { + vol.Required(TYPE): "zwave_js/get_firmware_update_capabilities", + vol.Required(DEVICE_ID): str, + } +) +@websocket_api.async_response +@async_handle_failed_command +@async_get_node +async def websocket_get_firmware_update_capabilities( + hass: HomeAssistant, + connection: ActiveConnection, + msg: dict, + node: Node, +) -> None: + """Abort a firmware update.""" + capabilities = await node.async_get_firmware_update_capabilities() + connection.send_result(msg[ID], capabilities.to_dict()) + + class FirmwareUploadView(HomeAssistantView): """View to upload firmware.""" diff --git a/tests/components/zwave_js/test_api.py b/tests/components/zwave_js/test_api.py index 149fe394a6d..337c74e955b 100644 --- a/tests/components/zwave_js/test_api.py +++ b/tests/components/zwave_js/test_api.py @@ -3423,19 +3423,10 @@ async def test_abort_firmware_update( assert not msg["success"] assert msg["error"]["code"] == ERR_NOT_LOADED - -async def test_abort_firmware_update_failures( - hass, multisensor_6, client, integration, hass_ws_client -): - """Test failures for the abort_firmware_update websocket command.""" - entry = integration - ws_client = await hass_ws_client(hass) - device = get_device(hass, multisensor_6) - # Test sending command with improper device ID fails await ws_client.send_json( { - ID: 2, + ID: 4, TYPE: "zwave_js/abort_firmware_update", DEVICE_ID: "fake_device", } @@ -3445,22 +3436,6 @@ async def test_abort_firmware_update_failures( 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: 3, - TYPE: "zwave_js/abort_firmware_update", - 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 @@ -3603,6 +3578,91 @@ async def test_subscribe_firmware_update_status_failures( assert msg["error"]["code"] == ERR_NOT_LOADED +async def test_get_firmware_update_capabilities( + hass, client, multisensor_6, integration, hass_ws_client +): + """Test that the get_firmware_update_capabilities 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 = { + "capabilities": { + "firmwareUpgradable": True, + "firmwareTargets": [0], + "continuesToFunction": True, + "supportsActivation": True, + } + } + await ws_client.send_json( + { + ID: 1, + TYPE: "zwave_js/get_firmware_update_capabilities", + DEVICE_ID: device.id, + } + ) + msg = await ws_client.receive_json() + assert msg["success"] + assert msg["result"] == { + "firmware_upgradable": True, + "firmware_targets": [0], + "continues_to_function": True, + "supports_activation": True, + } + + 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_capabilities" + assert args["nodeId"] == multisensor_6.node_id + + # Test FailedZWaveCommand is caught + with patch( + "zwave_js_server.model.node.Node.async_get_firmware_update_capabilities", + side_effect=FailedZWaveCommand("failed_command", 1, "error message"), + ): + await ws_client.send_json( + { + ID: 2, + TYPE: "zwave_js/get_firmware_update_capabilities", + 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_capabilities", + DEVICE_ID: device.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_firmware_update_capabilities", + DEVICE_ID: "fake_device", + } + ) + 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