Add Airzone Cloud diagnostics (#93465)

* airzone_cloud: add diagnostics support

Signed-off-by: Álvaro Fernández Rojas <noltari@gmail.com>

* airzone_cloue: remove unused import

Signed-off-by: Álvaro Fernández Rojas <noltari@gmail.com>

* airzone_cloud: diagnostics: redact additional API keys

Signed-off-by: Álvaro Fernández Rojas <noltari@gmail.com>

---------

Signed-off-by: Álvaro Fernández Rojas <noltari@gmail.com>

homeassistant/components/airzone_cloud/diagnostics.py

@@ -0,0 +1,138 @@
+"""Support for the Airzone Cloud diagnostics."""
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from aioairzone_cloud.const import (
+    API_STAT_AP_MAC,
+    API_STAT_SSID,
+    AZD_WIFI_MAC,
+    RAW_DEVICES_STATUS,
+    RAW_INSTALLATIONS,
+    RAW_WEBSERVERS,
+)
+
+from homeassistant.components.diagnostics.util import async_redact_data
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+from .coordinator import AirzoneUpdateCoordinator
+
+TO_REDACT_API = [
+    "_id",
+    "city",
+    "group_id",
+    "location_id",
+    "pin",
+    "user_id",
+    API_STAT_AP_MAC,
+    API_STAT_SSID,
+]
+
+TO_REDACT_CONFIG = [
+    CONF_PASSWORD,
+    CONF_USERNAME,
+]
+
+TO_REDACT_COORD = [
+    AZD_WIFI_MAC,
+]
+
+
+def gather_ids(api_data: dict[str, Any]) -> dict[str, Any]:
+    """Return dict with IDs."""
+    ids: dict[str, Any] = {}
+
+    dev_idx = 1
+    for dev_id in api_data[RAW_DEVICES_STATUS]:
+        if dev_id not in ids:
+            ids[dev_id] = f"device{dev_idx}"
+            dev_idx += 1
+
+    inst_idx = 1
+    for inst_id in api_data[RAW_INSTALLATIONS]:
+        if inst_id not in ids:
+            ids[inst_id] = f"installation{inst_idx}"
+            inst_idx += 1
+
+    ws_idx = 1
+    for ws_id in api_data[RAW_WEBSERVERS]:
+        if ws_id not in ids:
+            ids[ws_id] = f"webserver{ws_idx}"
+            ws_idx += 1
+
+    return ids
+
+
+def redact_keys(data: Any, ids: dict[str, Any]) -> Any:
+    """Redact sensitive keys in a dict."""
+    if not isinstance(data, (Mapping, list)):
+        return data
+
+    if isinstance(data, list):
+        return [redact_keys(val, ids) for val in data]
+
+    redacted = {**data}
+
+    keys = list(redacted)
+    for key in keys:
+        if key in ids:
+            redacted[ids[key]] = redacted.pop(key)
+        elif isinstance(redacted[key], Mapping):
+            redacted[key] = redact_keys(redacted[key], ids)
+        elif isinstance(redacted[key], list):
+            redacted[key] = [redact_keys(item, ids) for item in redacted[key]]
+
+    return redacted
+
+
+def redact_values(data: Any, ids: dict[str, Any]) -> Any:
+    """Redact sensitive values in a dict."""
+    if not isinstance(data, (Mapping, list)):
+        if data in ids:
+            return ids[data]
+        return data
+
+    if isinstance(data, list):
+        return [redact_values(val, ids) for val in data]
+
+    redacted = {**data}
+
+    for key, value in redacted.items():
+        if value is None:
+            continue
+        if isinstance(value, Mapping):
+            redacted[key] = redact_values(value, ids)
+        elif isinstance(value, list):
+            redacted[key] = [redact_values(item, ids) for item in value]
+        elif value in ids:
+            redacted[key] = ids[value]
+
+    return redacted
+
+
+def redact_all(
+    data: dict[str, Any], ids: dict[str, Any], to_redact: list[str]
+) -> dict[str, Any]:
+    """Redact sensitive data."""
+    _data = redact_keys(data, ids)
+    _data = redact_values(_data, ids)
+    return async_redact_data(_data, to_redact)
+
+
+async def async_get_config_entry_diagnostics(
+    hass: HomeAssistant, config_entry: ConfigEntry
+) -> dict[str, Any]:
+    """Return diagnostics for a config entry."""
+    coordinator: AirzoneUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
+    raw_data = coordinator.airzone.raw_data()
+    ids = gather_ids(raw_data)
+
+    return {
+        "api_data": redact_all(raw_data, ids, TO_REDACT_API),
+        "config_entry": redact_all(config_entry.as_dict(), ids, TO_REDACT_CONFIG),
+        "coord_data": redact_all(coordinator.data, ids, TO_REDACT_COORD),
+    }

tests/components/airzone_cloud/test_diagnostics.py

@@ -0,0 +1,116 @@
+"""The diagnostics tests for the Airzone Cloud platform."""
+
+from unittest.mock import patch
+
+from aioairzone_cloud.const import (
+    API_DEVICE_ID,
+    API_DEVICES,
+    API_GROUPS,
+    API_WS_ID,
+    AZD_INSTALLATIONS,
+    AZD_SYSTEMS,
+    AZD_WEBSERVERS,
+    AZD_ZONES,
+    RAW_DEVICES_CONFIG,
+    RAW_DEVICES_STATUS,
+    RAW_INSTALLATIONS,
+    RAW_INSTALLATIONS_LIST,
+    RAW_WEBSERVERS,
+)
+
+from homeassistant.components.airzone_cloud.const import DOMAIN
+from homeassistant.components.diagnostics import REDACTED
+from homeassistant.const import CONF_ID, CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+
+from .util import CONFIG, WS_ID, async_init_integration
+
+from tests.components.diagnostics import get_diagnostics_for_config_entry
+from tests.typing import ClientSessionGenerator
+
+RAW_DATA_MOCK = {
+    RAW_DEVICES_CONFIG: {
+        "dev1": {},
+    },
+    RAW_DEVICES_STATUS: {
+        "dev1": {},
+    },
+    RAW_INSTALLATIONS: {
+        CONFIG[CONF_ID]: {
+            API_GROUPS: [
+                {
+                    API_DEVICES: [
+                        {
+                            API_DEVICE_ID: "device1",
+                            API_WS_ID: WS_ID,
+                        },
+                    ],
+                },
+            ],
+            "plugins": {
+                "schedules": {
+                    "calendar_ws_ids": [
+                        WS_ID,
+                    ],
+                },
+            },
+        },
+    },
+    RAW_INSTALLATIONS_LIST: {},
+    RAW_WEBSERVERS: {
+        WS_ID: {},
+    },
+    "test_cov": {
+        "1": None,
+        "2": ["foo", "bar"],
+        "3": [
+            [
+                "foo",
+                "bar",
+            ],
+        ],
+    },
+}
+
+
+async def test_config_entry_diagnostics(
+    hass: HomeAssistant, hass_client: ClientSessionGenerator
+) -> None:
+    """Test config entry diagnostics."""
+    await async_init_integration(hass)
+    assert hass.data[DOMAIN]
+
+    config_entry = hass.config_entries.async_entries(DOMAIN)[0]
+    with patch(
+        "homeassistant.components.airzone_cloud.AirzoneCloudApi.raw_data",
+        return_value=RAW_DATA_MOCK,
+    ):
+        diag = await get_diagnostics_for_config_entry(hass, hass_client, config_entry)
+
+    assert list(diag["api_data"]) >= list(RAW_DATA_MOCK)
+    assert "dev1" not in diag["api_data"][RAW_DEVICES_CONFIG]
+    assert "device1" in diag["api_data"][RAW_DEVICES_CONFIG]
+    assert "inst1" not in diag["api_data"][RAW_INSTALLATIONS]
+    assert "installation1" in diag["api_data"][RAW_INSTALLATIONS]
+    assert WS_ID not in diag["api_data"][RAW_WEBSERVERS]
+    assert "webserver1" in diag["api_data"][RAW_WEBSERVERS]
+
+    assert (
+        diag["config_entry"].items()
+        >= {
+            "data": {
+                CONF_ID: "installation1",
+                CONF_PASSWORD: REDACTED,
+                CONF_USERNAME: REDACTED,
+            },
+            "domain": DOMAIN,
+            "unique_id": "installation1",
+        }.items()
+    )
+
+    assert list(diag["coord_data"]) >= [
+        AZD_INSTALLATIONS,
+        AZD_SYSTEMS,
+        AZD_WEBSERVERS,
+        AZD_ZONES,
+    ]

tests/components/airzone_cloud/util.py

@@ -151,7 +151,7 @@ async def async_init_integration(
     config_entry = MockConfigEntry(
         data=CONFIG,
         domain=DOMAIN,
-        unique_id="airzone_cloud_unique_id",
+        unique_id=CONFIG[CONF_ID],
     )
     config_entry.add_to_hass(hass)