Add zwave_js support for diagnostics (#64336)

homeassistant/components/zwave_js/api.py

@@ -4,12 +4,10 @@ from __future__ import annotations
 from collections.abc import Callable
 import dataclasses
 from functools import partial, wraps
-import json
 from typing import Any
 
-from aiohttp import hdrs, web, web_exceptions, web_request
+from aiohttp import web, web_exceptions, web_request
 import voluptuous as vol
-from zwave_js_server import dump
 from zwave_js_server.client import Client
 from zwave_js_server.const import (
     CommandClass,
@@ -350,7 +348,6 @@ def async_register_api(hass: HomeAssistant) -> None:
     websocket_api.async_register_command(hass, websocket_subscribe_node_statistics)
     websocket_api.async_register_command(hass, websocket_node_ready)
     websocket_api.async_register_command(hass, websocket_migrate_zwave)
-    hass.http.register_view(DumpView())
     hass.http.register_view(FirmwareUploadView())
 
 
@@ -1779,35 +1776,6 @@ async def websocket_data_collection_status(
     connection.send_result(msg[ID], result)
 
 
-class DumpView(HomeAssistantView):
-    """View to dump the state of the Z-Wave JS server."""
-
-    url = "/api/zwave_js/dump/{config_entry_id}"
-    name = "api:zwave_js:dump"
-
-    async def get(self, request: web.Request, config_entry_id: str) -> web.Response:
-        """Dump the state of Z-Wave."""
-        # pylint: disable=no-self-use
-        if not request["hass_user"].is_admin:
-            raise Unauthorized()
-        hass = request.app["hass"]
-
-        if config_entry_id not in hass.data[DOMAIN]:
-            raise web_exceptions.HTTPBadRequest
-
-        entry = hass.config_entries.async_get_entry(config_entry_id)
-
-        msgs = await dump.dump_msgs(entry.data[CONF_URL], async_get_clientsession(hass))
-
-        return web.Response(
-            body=json.dumps(msgs, indent=2) + "\n",
-            headers={
-                hdrs.CONTENT_TYPE: "application/json",
-                hdrs.CONTENT_DISPOSITION: 'attachment; filename="zwave_js_dump.json"',
-            },
-        )
-
-
 @websocket_api.require_admin
 @websocket_api.websocket_command(
     {

homeassistant/components/zwave_js/diagnostics.py

@@ -0,0 +1,19 @@
+"""Provides diagnostics for Z-Wave JS."""
+from __future__ import annotations
+
+from zwave_js_server.dump import dump_msgs
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_URL
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+
+async def async_get_config_entry_diagnostics(
+    hass: HomeAssistant, config_entry: ConfigEntry
+) -> list[dict]:
+    """Return diagnostics for a config entry."""
+    msgs: list[dict] = await dump_msgs(
+        config_entry.data[CONF_URL], async_get_clientsession(hass)
+    )
+    return msgs

tests/components/zwave_js/test_api.py

@@ -2799,20 +2799,8 @@ async def test_get_config_parameters(hass, multisensor_6, integration, hass_ws_c
     assert msg["error"]["code"] == ERR_NOT_LOADED
 
 
-async def test_dump_view(integration, hass_client):
-    """Test the HTTP dump view."""
-    client = await hass_client()
-    with patch(
-        "zwave_js_server.dump.dump_msgs",
-        return_value=[{"hello": "world"}, {"second": "msg"}],
-    ):
-        resp = await client.get(f"/api/zwave_js/dump/{integration.entry_id}")
-    assert resp.status == HTTPStatus.OK
-    assert json.loads(await resp.text()) == [{"hello": "world"}, {"second": "msg"}]
-
-
 async def test_version_info(hass, integration, hass_ws_client, version_state):
-    """Test the HTTP dump node view."""
+    """Test version info API request."""
     entry = integration
     ws_client = await hass_ws_client(hass)
 
@@ -2906,21 +2894,6 @@ async def test_firmware_upload_view_invalid_payload(
     assert resp.status == HTTPStatus.BAD_REQUEST
 
 
-@pytest.mark.parametrize(
-    "method, url",
-    [("get", "/api/zwave_js/dump/{}")],
-)
-async def test_view_non_admin_user(
-    integration, hass_client, hass_admin_user, method, url
-):
-    """Test config entry level views for non-admin users."""
-    client = await hass_client()
-    # Verify we require admin user
-    hass_admin_user.groups = []
-    resp = await client.request(method, url.format(integration.entry_id))
-    assert resp.status == HTTPStatus.UNAUTHORIZED
-
-
 @pytest.mark.parametrize(
     "method, url",
     [("post", "/api/zwave_js/firmware/upload/{}/{}")],
@@ -2941,7 +2914,6 @@ async def test_node_view_non_admin_user(
 @pytest.mark.parametrize(
     "method, url",
     [
-        ("get", "/api/zwave_js/dump/INVALID"),
         ("post", "/api/zwave_js/firmware/upload/INVALID/1"),
     ],
 )

tests/components/zwave_js/test_diagnostics.py

@@ -0,0 +1,16 @@
+"""Test the Z-Wave JS diagnostics."""
+from unittest.mock import patch
+
+from homeassistant.components.zwave_js.diagnostics import (
+    async_get_config_entry_diagnostics,
+)
+
+
+async def test_config_entry_diagnostics(hass, integration):
+    """Test the config entry level diagnostics data dump."""
+    with patch(
+        "homeassistant.components.zwave_js.diagnostics.dump_msgs",
+        return_value=[{"hello": "world"}, {"second": "msg"}],
+    ):
+        result = await async_get_config_entry_diagnostics(hass, integration)
+        assert result == [{"hello": "world"}, {"second": "msg"}]