Add WS API commands to check for and install zwave_js config updates (#51106)

homeassistant/components/zwave_js/api.py

@@ -168,6 +168,8 @@ 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_check_for_config_updates)
+    websocket_api.async_register_command(hass, websocket_install_config_update)
     hass.http.register_view(DumpView())
     hass.http.register_view(FirmwareUploadView())
 
@@ -1312,3 +1314,51 @@ class FirmwareUploadView(HomeAssistantView):
             raise web_exceptions.HTTPBadRequest from err
 
         return self.json(None)
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command(
+    {
+        vol.Required(TYPE): "zwave_js/check_for_config_updates",
+        vol.Required(ENTRY_ID): str,
+    }
+)
+@websocket_api.async_response
+@async_get_entry
+async def websocket_check_for_config_updates(
+    hass: HomeAssistant,
+    connection: ActiveConnection,
+    msg: dict,
+    entry: ConfigEntry,
+    client: Client,
+) -> None:
+    """Check for config updates."""
+    config_update = await client.driver.async_check_for_config_updates()
+    connection.send_result(
+        msg[ID],
+        {
+            "update_available": config_update.update_available,
+            "new_version": config_update.new_version,
+        },
+    )
+
+
+@websocket_api.require_admin
+@websocket_api.websocket_command(
+    {
+        vol.Required(TYPE): "zwave_js/install_config_update",
+        vol.Required(ENTRY_ID): str,
+    }
+)
+@websocket_api.async_response
+@async_get_entry
+async def websocket_install_config_update(
+    hass: HomeAssistant,
+    connection: ActiveConnection,
+    msg: dict,
+    entry: ConfigEntry,
+    client: Client,
+) -> None:
+    """Check for config updates."""
+    success = await client.driver.async_install_config_update()
+    connection.send_result(msg[ID], success)

tests/components/zwave_js/test_api.py

@@ -1899,3 +1899,104 @@ async def test_subscribe_firmware_update_status_failures(
 
     assert not msg["success"]
     assert msg["error"]["code"] == ERR_NOT_LOADED
+
+
+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
+    ws_client = await hass_ws_client(hass)
+
+    # Test we can get log configuration
+    client.async_send_command.return_value = {
+        "updateAvailable": True,
+        "newVersion": "test",
+    }
+    await ws_client.send_json(
+        {
+            ID: 1,
+            TYPE: "zwave_js/check_for_config_updates",
+            ENTRY_ID: entry.entry_id,
+        }
+    )
+    msg = await ws_client.receive_json()
+    assert msg["result"]
+    assert msg["success"]
+
+    config_update = msg["result"]
+    assert config_update["update_available"]
+    assert config_update["new_version"] == "test"
+
+    # 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: 2,
+            TYPE: "zwave_js/check_for_config_updates",
+            ENTRY_ID: entry.entry_id,
+        }
+    )
+    msg = await ws_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == ERR_NOT_LOADED
+
+    await ws_client.send_json(
+        {
+            ID: 3,
+            TYPE: "zwave_js/check_for_config_updates",
+            ENTRY_ID: "INVALID",
+        }
+    )
+    msg = await ws_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == ERR_NOT_FOUND
+
+
+async def test_install_config_update(hass, client, integration, hass_ws_client):
+    """Test that the install_config_update WS API call works."""
+    entry = integration
+    ws_client = await hass_ws_client(hass)
+
+    # Test we can get log configuration
+    client.async_send_command.return_value = {"success": True}
+    await ws_client.send_json(
+        {
+            ID: 1,
+            TYPE: "zwave_js/install_config_update",
+            ENTRY_ID: entry.entry_id,
+        }
+    )
+    msg = await ws_client.receive_json()
+    assert msg["result"]
+    assert msg["success"]
+
+    # 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: 2,
+            TYPE: "zwave_js/install_config_update",
+            ENTRY_ID: entry.entry_id,
+        }
+    )
+    msg = await ws_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == ERR_NOT_LOADED
+
+    await ws_client.send_json(
+        {
+            ID: 3,
+            TYPE: "zwave_js/install_config_update",
+            ENTRY_ID: "INVALID",
+        }
+    )
+    msg = await ws_client.receive_json()
+
+    assert not msg["success"]
+    assert msg["error"]["code"] == ERR_NOT_FOUND