Add matter diagnostics (#86091)

* Add matter diagnostics

* Complete test typing

* Rename redact attributes helper

* Adjust device lookup after identifier addition

homeassistant/components/matter/diagnostics.py

@@ -0,0 +1,80 @@
+"""Provide diagnostics for Matter."""
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any
+
+from matter_server.common.helpers.util import dataclass_to_dict
+
+from homeassistant.components.diagnostics import REDACTED
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+
+from .const import DOMAIN, ID_TYPE_DEVICE_ID
+from .helpers import get_device_id, get_matter
+
+ATTRIBUTES_TO_REDACT = {"chip.clusters.Objects.Basic.Attributes.Location"}
+
+
+def redact_matter_attributes(node_data: dict[str, Any]) -> dict[str, Any]:
+    """Redact Matter cluster attribute."""
+    redacted = deepcopy(node_data)
+    for attribute_to_redact in ATTRIBUTES_TO_REDACT:
+        for value in redacted["attributes"].values():
+            if value["attribute_type"] == attribute_to_redact:
+                value["value"] = REDACTED
+
+    return redacted
+
+
+def remove_serialization_type(data: dict[str, Any]) -> dict[str, Any]:
+    """Remove serialization type from data."""
+    if "_type" in data:
+        data.pop("_type")
+    return data
+
+
+async def async_get_config_entry_diagnostics(
+    hass: HomeAssistant, config_entry: ConfigEntry
+) -> dict[str, Any]:
+    """Return diagnostics for a config entry."""
+    matter = get_matter(hass)
+    server_diagnostics = await matter.matter_client.get_diagnostics()
+    data = remove_serialization_type(dataclass_to_dict(server_diagnostics))
+    nodes = [redact_matter_attributes(node_data) for node_data in data["nodes"]]
+    data["nodes"] = nodes
+
+    return {"server": data}
+
+
+async def async_get_device_diagnostics(
+    hass: HomeAssistant, config_entry: ConfigEntry, device: dr.DeviceEntry
+) -> dict[str, Any]:
+    """Return diagnostics for a device."""
+    matter = get_matter(hass)
+    device_id_type_prefix = f"{ID_TYPE_DEVICE_ID}_"
+    device_id_full = next(
+        identifier[1]
+        for identifier in device.identifiers
+        if identifier[0] == DOMAIN and identifier[1].startswith(device_id_type_prefix)
+    )
+    device_id = device_id_full.lstrip(device_id_type_prefix)
+
+    server_diagnostics = await matter.matter_client.get_diagnostics()
+
+    node = next(
+        node
+        for node in await matter.matter_client.get_nodes()
+        for node_device in node.node_devices
+        if get_device_id(server_diagnostics.info, node_device) == device_id
+    )
+
+    return {
+        "server_info": remove_serialization_type(
+            dataclass_to_dict(server_diagnostics.info)
+        ),
+        "node": redact_matter_attributes(
+            remove_serialization_type(dataclass_to_dict(node))
+        ),
+    }

tests/components/matter/test_diagnostics.py

@@ -0,0 +1,112 @@
+"""Test the Matter diagnostics platform."""
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+import json
+from typing import Any
+from unittest.mock import MagicMock
+
+from aiohttp import ClientSession
+from matter_server.common.helpers.util import dataclass_from_dict
+from matter_server.common.models.server_information import ServerDiagnostics
+import pytest
+
+from homeassistant.components.matter.const import DOMAIN
+from homeassistant.components.matter.diagnostics import redact_matter_attributes
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
+
+from .common import setup_integration_with_node_fixture
+
+from tests.common import MockConfigEntry, load_fixture
+from tests.components.diagnostics import (
+    get_diagnostics_for_config_entry,
+    get_diagnostics_for_device,
+)
+
+
+@pytest.fixture(name="config_entry_diagnostics")
+def config_entry_diagnostics_fixture() -> dict[str, Any]:
+    """Fixture for config entry diagnostics."""
+    return json.loads(load_fixture("config_entry_diagnostics.json", DOMAIN))
+
+
+@pytest.fixture(name="config_entry_diagnostics_redacted")
+def config_entry_diagnostics_redacted_fixture() -> dict[str, Any]:
+    """Fixture for redacted config entry diagnostics."""
+    return json.loads(load_fixture("config_entry_diagnostics_redacted.json", DOMAIN))
+
+
+@pytest.fixture(name="device_diagnostics")
+def device_diagnostics_fixture() -> dict[str, Any]:
+    """Fixture for device diagnostics."""
+    return json.loads(load_fixture("nodes/device_diagnostics.json", DOMAIN))
+
+
+async def test_matter_attribute_redact(device_diagnostics: dict[str, Any]) -> None:
+    """Test the matter attribute redact helper."""
+    assert device_diagnostics["attributes"]["0/40/6"]["value"] == "XX"
+
+    redacted_device_diagnostics = redact_matter_attributes(device_diagnostics)
+
+    # Check that the correct attribute value is redacted.
+    assert (
+        redacted_device_diagnostics["attributes"]["0/40/6"]["value"] == "**REDACTED**"
+    )
+
+    # Check that the other attribute values are not redacted.
+    redacted_device_diagnostics["attributes"]["0/40/6"]["value"] = "XX"
+    assert redacted_device_diagnostics == device_diagnostics
+
+
+async def test_config_entry_diagnostics(
+    hass: HomeAssistant,
+    hass_client: Callable[..., Awaitable[ClientSession]],
+    matter_client: MagicMock,
+    integration: MockConfigEntry,
+    config_entry_diagnostics: dict[str, Any],
+    config_entry_diagnostics_redacted: dict[str, Any],
+) -> None:
+    """Test the config entry level diagnostics."""
+    matter_client.get_diagnostics.return_value = dataclass_from_dict(
+        ServerDiagnostics, config_entry_diagnostics
+    )
+
+    diagnostics = await get_diagnostics_for_config_entry(hass, hass_client, integration)
+
+    assert diagnostics == config_entry_diagnostics_redacted
+
+
+async def test_device_diagnostics(
+    hass: HomeAssistant,
+    hass_client: Callable[..., Awaitable[ClientSession]],
+    matter_client: MagicMock,
+    config_entry_diagnostics: dict[str, Any],
+    device_diagnostics: dict[str, Any],
+) -> None:
+    """Test the device diagnostics."""
+    await setup_integration_with_node_fixture(hass, "device_diagnostics", matter_client)
+    system_info_dict = config_entry_diagnostics["info"]
+    device_diagnostics_redacted = {
+        "server_info": system_info_dict,
+        "node": redact_matter_attributes(device_diagnostics),
+    }
+    server_diagnostics_response = {
+        "info": system_info_dict,
+        "nodes": [device_diagnostics],
+        "events": [],
+    }
+    server_diagnostics = dataclass_from_dict(
+        ServerDiagnostics, server_diagnostics_response
+    )
+    matter_client.get_diagnostics.return_value = server_diagnostics
+    config_entry = hass.config_entries.async_entries(DOMAIN)[0]
+    dev_reg = dr.async_get(hass)
+    device = dr.async_entries_for_config_entry(dev_reg, config_entry.entry_id)[0]
+    assert device
+
+    diagnostics = await get_diagnostics_for_device(
+        hass, hass_client, config_entry, device
+    )
+
+    assert diagnostics == device_diagnostics_redacted

tests/components/matter/test_helpers.py

@@ -0,0 +1,22 @@
+"""Test the Matter helpers."""
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from homeassistant.components.matter.helpers import get_device_id
+from homeassistant.core import HomeAssistant
+
+from .common import setup_integration_with_node_fixture
+
+
+async def test_get_device_id(
+    hass: HomeAssistant,
+    matter_client: MagicMock,
+) -> None:
+    """Test get_device_id."""
+    node = await setup_integration_with_node_fixture(
+        hass, "device_diagnostics", matter_client
+    )
+    device_id = get_device_id(matter_client.server_info, node.node_devices[0])
+
+    assert device_id == "00000000000004D2-0000000000000005-MatterNodeDevice"