Add zwave_js/get_firmware_update_capabilties WS command (#73691)

* Add zwave_js/get_firmware_update_capabilties WS command

* Fix test

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."""
 

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