Add diagnostics to Velux integration (#163896)

homeassistant/components/velux/diagnostics.py

@@ -0,0 +1,86 @@
+"""Diagnostics support for Velux."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from homeassistant.components.diagnostics import async_redact_data
+from homeassistant.const import CONF_MAC, CONF_PASSWORD
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr, entity_registry as er
+
+from . import VeluxConfigEntry
+
+TO_REDACT = {CONF_MAC, CONF_PASSWORD}
+
+
+async def async_get_config_entry_diagnostics(
+    hass: HomeAssistant, entry: VeluxConfigEntry
+) -> dict[str, Any]:
+    """Return diagnostics for a config entry, includes nodes, devices, and entities."""
+
+    pyvlx = entry.runtime_data
+
+    nodes: list[dict[str, Any]] = [
+        {
+            "node_id": node.node_id,
+            "name": node.name,
+            "serial_number": node.serial_number,
+            "type": type(node).__name__,
+            "device_updated_callbacks": node.device_updated_cbs,
+        }
+        for node in pyvlx.nodes
+    ]
+
+    device_registry = dr.async_get(hass)
+    entity_registry = er.async_get(hass)
+
+    devices: list[dict[str, Any]] = []
+    for device in dr.async_entries_for_config_entry(device_registry, entry.entry_id):
+        entities: list[dict[str, Any]] = []
+        for entity_entry in er.async_entries_for_device(
+            entity_registry,
+            device_id=device.id,
+            include_disabled_entities=True,
+        ):
+            state_dict = None
+            if state := hass.states.get(entity_entry.entity_id):
+                state_dict = dict(state.as_dict())
+                state_dict.pop("context", None)
+
+            entities.append(
+                {
+                    "entity_id": entity_entry.entity_id,
+                    "unique_id": entity_entry.unique_id,
+                    "state": state_dict,
+                }
+            )
+
+        devices.append(
+            {
+                "name": device.name,
+                "entities": entities,
+            }
+        )
+
+    return {
+        "config_entry": async_redact_data(entry.data, TO_REDACT),
+        "connection": {
+            "connected": pyvlx.connection.connected,
+            "connection_count": pyvlx.connection.connection_counter,
+            "frame_received_cbs": pyvlx.connection.frame_received_cbs,
+            "connection_opened_cbs": pyvlx.connection.connection_opened_cbs,
+            "connection_closed_cbs": pyvlx.connection.connection_closed_cbs,
+        },
+        "gateway": {
+            "state": str(pyvlx.klf200.state) if pyvlx.klf200.state else None,
+            "version": str(pyvlx.klf200.version) if pyvlx.klf200.version else None,
+            "protocol_version": (
+                str(pyvlx.klf200.protocol_version)
+                if pyvlx.klf200.protocol_version
+                else None
+            ),
+        },
+        "nodes": nodes,
+        "devices": devices,
+    }

homeassistant/components/velux/quality_scale.yaml

@@ -33,7 +33,7 @@ rules:
 
   # Gold
   devices: done
-  diagnostics: todo
+  diagnostics: done
   discovery-update-info: todo
   discovery: done
   docs-data-update: todo

tests/components/velux/conftest.py

@@ -71,6 +71,7 @@ def mock_window() -> AsyncMock:
     window.rain_sensor = True
     window.serial_number = "123456789"
     window.get_limitation.return_value = MagicMock(min_value=0)
+    window.device_updated_cbs = []
     window.is_opening = False
     window.is_closing = False
     window.position = MagicMock(position_percent=30, closed=False)

tests/components/velux/snapshots/test_diagnostics.ambr

@@ -0,0 +1,63 @@
+# serializer version: 1
+# name: test_diagnostics[mock_window]
+  dict({
+    'config_entry': dict({
+      'host': '127.0.0.1',
+      'password': '**REDACTED**',
+    }),
+    'connection': dict({
+      'connected': True,
+      'connection_closed_cbs': list([
+      ]),
+      'connection_count': 3,
+      'connection_opened_cbs': list([
+      ]),
+      'frame_received_cbs': list([
+      ]),
+    }),
+    'devices': list([
+      dict({
+        'entities': list([
+        ]),
+        'name': 'KLF 200 Gateway',
+      }),
+      dict({
+        'entities': list([
+          dict({
+            'entity_id': 'cover.test_window',
+            'state': dict({
+              'attributes': dict({
+                'current_position': 70,
+                'device_class': 'window',
+                'friendly_name': 'Test Window',
+                'supported_features': 15,
+              }),
+              'entity_id': 'cover.test_window',
+              'last_changed': '2025-01-01T00:00:00+00:00',
+              'last_reported': '2025-01-01T00:00:00+00:00',
+              'last_updated': '2025-01-01T00:00:00+00:00',
+              'state': 'open',
+            }),
+            'unique_id': '123456789',
+          }),
+        ]),
+        'name': 'Test Window',
+      }),
+    ]),
+    'gateway': dict({
+      'protocol_version': None,
+      'state': '<DtoState gateway_state="GatewayState.GATEWAY_MODE_WITH_ACTUATORS" gateway_sub_state="GatewaySubState.IDLE"/>',
+      'version': None,
+    }),
+    'nodes': list([
+      dict({
+        'device_updated_callbacks': list([
+        ]),
+        'name': 'Test Window',
+        'node_id': 1,
+        'serial_number': '123456789',
+        'type': 'AsyncMock',
+      }),
+    ]),
+  })
+# ---

tests/components/velux/test_diagnostics.py

@@ -0,0 +1,50 @@
+"""Tests for the diagnostics data provided by the Velux integration."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from pyvlx.const import GatewayState, GatewaySubState
+from pyvlx.dataobjects import DtoState
+from syrupy.assertion import SnapshotAssertion
+
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+from tests.components.diagnostics import get_diagnostics_for_config_entry
+from tests.typing import ClientSessionGenerator
+
+
+@pytest.mark.freeze_time("2025-01-01T00:00:00+00:00")
+@pytest.mark.parametrize("mock_pyvlx", ["mock_window"], indirect=True)
+async def test_diagnostics(
+    hass: HomeAssistant,
+    hass_client: ClientSessionGenerator,
+    mock_config_entry: MockConfigEntry,
+    mock_pyvlx: MagicMock,
+    mock_window: MagicMock,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test diagnostics for Velux config entry."""
+    mock_window.node_id = 1
+    mock_pyvlx.connection.connected = True
+    mock_pyvlx.connection.connection_counter = 3
+    mock_pyvlx.connection.frame_received_cbs = []
+    mock_pyvlx.connection.connection_opened_cbs = []
+    mock_pyvlx.connection.connection_closed_cbs = []
+    mock_pyvlx.klf200.state = DtoState(
+        GatewayState.GATEWAY_MODE_WITH_ACTUATORS, GatewaySubState.IDLE
+    )
+    mock_pyvlx.klf200.version = None
+    mock_pyvlx.klf200.protocol_version = None
+
+    mock_config_entry.add_to_hass(hass)
+
+    with patch("homeassistant.components.velux.PLATFORMS", [Platform.COVER]):
+        await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert (
+        await get_diagnostics_for_config_entry(hass, hass_client, mock_config_entry)
+        == snapshot
+    )