Add diagnostics platform to Shelly (#64660)

* add config entry diagnostics

* add test

* make pylint happy

* add some device settings

* adjust tests

* add hass system info

* make sw_version be a property of wrapper classes

* remove hass sys info, since is added centraly

* check if device is initialized

* device_info is always available

* add device status data

homeassistant/components/shelly/__init__.py

@@ -428,17 +428,21 @@ class BlockDeviceWrapper(update_coordinator.DataUpdateCoordinator):
         """Mac address of the device."""
         return cast(str, self.entry.unique_id)
 
+    @property
+    def sw_version(self) -> str:
+        """Firmware version of the device."""
+        return self.device.firmware_version if self.device.initialized else ""
+
     def async_setup(self) -> None:
         """Set up the wrapper."""
         dev_reg = device_registry.async_get(self.hass)
-        sw_version = self.device.firmware_version if self.device.initialized else ""
         entry = dev_reg.async_get_or_create(
             config_entry_id=self.entry.entry_id,
             name=self.name,
             connections={(device_registry.CONNECTION_NETWORK_MAC, self.mac)},
             manufacturer="Shelly",
             model=aioshelly.const.MODEL_NAMES.get(self.model, self.model),
-            sw_version=sw_version,
+            sw_version=self.sw_version,
             hw_version=f"gen{self.device.gen} ({self.model})",
             configuration_url=f"http://{self.entry.data[CONF_HOST]}",
         )
@@ -713,17 +717,21 @@ class RpcDeviceWrapper(update_coordinator.DataUpdateCoordinator):
         """Mac address of the device."""
         return cast(str, self.entry.unique_id)
 
+    @property
+    def sw_version(self) -> str:
+        """Firmware version of the device."""
+        return self.device.firmware_version if self.device.initialized else ""
+
     def async_setup(self) -> None:
         """Set up the wrapper."""
         dev_reg = device_registry.async_get(self.hass)
-        sw_version = self.device.firmware_version if self.device.initialized else ""
         entry = dev_reg.async_get_or_create(
             config_entry_id=self.entry.entry_id,
             name=self.name,
             connections={(device_registry.CONNECTION_NETWORK_MAC, self.mac)},
             manufacturer="Shelly",
             model=aioshelly.const.MODEL_NAMES.get(self.model, self.model),
-            sw_version=sw_version,
+            sw_version=self.sw_version,
             hw_version=f"gen{self.device.gen} ({self.model})",
             configuration_url=f"http://{self.entry.data[CONF_HOST]}",
         )

homeassistant/components/shelly/diagnostics.py

@@ -0,0 +1,78 @@
+"""Diagnostics support for Shelly."""
+from __future__ import annotations
+
+from homeassistant.components.diagnostics import async_redact_data
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
+from homeassistant.core import HomeAssistant
+
+from . import BlockDeviceWrapper, RpcDeviceWrapper
+from .const import BLOCK, DATA_CONFIG_ENTRY, DOMAIN, RPC
+
+TO_REDACT = {CONF_USERNAME, CONF_PASSWORD}
+
+
+async def async_get_config_entry_diagnostics(
+    hass: HomeAssistant, entry: ConfigEntry
+) -> dict:
+    """Return diagnostics for a config entry."""
+    data: dict = hass.data[DOMAIN][DATA_CONFIG_ENTRY][entry.entry_id]
+
+    device_settings: str | dict = "not initialized"
+    device_status: str | dict = "not initialized"
+    if BLOCK in data:
+        block_wrapper: BlockDeviceWrapper = data[BLOCK]
+        device_info = {
+            "name": block_wrapper.name,
+            "model": block_wrapper.model,
+            "sw_version": block_wrapper.sw_version,
+        }
+        if block_wrapper.device.initialized:
+            device_settings = {
+                k: v
+                for k, v in block_wrapper.device.settings.items()
+                if k in ["cloud", "coiot"]
+            }
+            device_status = {
+                k: v
+                for k, v in block_wrapper.device.status.items()
+                if k
+                in [
+                    "update",
+                    "wifi_sta",
+                    "time",
+                    "has_update",
+                    "ram_total",
+                    "ram_free",
+                    "ram_lwm",
+                    "fs_size",
+                    "fs_free",
+                    "uptime",
+                ]
+            }
+    else:
+        rpc_wrapper: RpcDeviceWrapper = data[RPC]
+        device_info = {
+            "name": rpc_wrapper.name,
+            "model": rpc_wrapper.model,
+            "sw_version": rpc_wrapper.sw_version,
+        }
+        if rpc_wrapper.device.initialized:
+            device_settings = {
+                k: v for k, v in rpc_wrapper.device.config.items() if k in ["cloud"]
+            }
+            device_status = {
+                k: v
+                for k, v in rpc_wrapper.device.status.items()
+                if k in ["sys", "wifi"]
+            }
+
+    if isinstance(device_status, dict):
+        device_status = async_redact_data(device_status, ["ssid"])
+
+    return {
+        "entry": async_redact_data(entry.as_dict(), TO_REDACT),
+        "device_info": device_info,
+        "device_settings": device_settings,
+        "device_status": device_status,
+    }

tests/components/shelly/test_diagnostics.py

@@ -0,0 +1,80 @@
+"""The scene tests for the myq platform."""
+from aiohttp import ClientSession
+
+from homeassistant.components.diagnostics import REDACTED
+from homeassistant.components.shelly.const import DOMAIN
+from homeassistant.components.shelly.diagnostics import TO_REDACT
+from homeassistant.core import HomeAssistant
+
+from tests.components.diagnostics import get_diagnostics_for_config_entry
+
+RELAY_BLOCK_ID = 0
+
+
+async def test_block_config_entry_diagnostics(
+    hass: HomeAssistant, hass_client: ClientSession, coap_wrapper
+):
+    """Test config entry diagnostics for block device."""
+    assert coap_wrapper
+
+    entry = hass.config_entries.async_entries(DOMAIN)[0]
+    entry_dict = entry.as_dict()
+    entry_dict["data"].update(
+        {key: REDACTED for key in TO_REDACT if key in entry_dict["data"]}
+    )
+
+    result = await get_diagnostics_for_config_entry(hass, hass_client, entry)
+
+    assert result == {
+        "entry": entry_dict,
+        "device_info": {
+            "name": coap_wrapper.name,
+            "model": coap_wrapper.model,
+            "sw_version": coap_wrapper.sw_version,
+        },
+        "device_settings": {"coiot": {"update_period": 15}},
+        "device_status": {
+            "update": {
+                "beta_version": "some_beta_version",
+                "has_update": True,
+                "new_version": "some_new_version",
+                "old_version": "some_old_version",
+                "status": "pending",
+            }
+        },
+    }
+
+
+async def test_rpc_config_entry_diagnostics(
+    hass: HomeAssistant,
+    hass_client: ClientSession,
+    rpc_wrapper,
+):
+    """Test config entry diagnostics for rpc device."""
+    assert rpc_wrapper
+
+    entry = hass.config_entries.async_entries(DOMAIN)[0]
+    entry_dict = entry.as_dict()
+    entry_dict["data"].update(
+        {key: REDACTED for key in TO_REDACT if key in entry_dict["data"]}
+    )
+
+    result = await get_diagnostics_for_config_entry(hass, hass_client, entry)
+
+    assert result == {
+        "entry": entry_dict,
+        "device_info": {
+            "name": rpc_wrapper.name,
+            "model": rpc_wrapper.model,
+            "sw_version": rpc_wrapper.sw_version,
+        },
+        "device_settings": {},
+        "device_status": {
+            "sys": {
+                "available_updates": {
+                    "beta": {"version": "some_beta_version"},
+                    "stable": {"version": "some_beta_version"},
+                }
+            }
+        },
+    }