Add diagnostics per device to Home Connect (#131010)

* Add diagnostics per device to Home Connect

* Include programs at device diagnostics

* Applied suggestions from epenet

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Applied more suggestions from epenet

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>

* Test naming consistency

Co-authored-by: abmantis <amfcalt@gmail.com>

* Add return type to `_generate_entry_diagnostics`

---------

Co-authored-by: epenet <6771947+epenet@users.noreply.github.com>
Co-authored-by: abmantis <amfcalt@gmail.com>

homeassistant/components/home_connect/__init__.py

@@ -6,6 +6,7 @@ from datetime import timedelta
 import logging
 from typing import Any
 
+from homeconnect.api import HomeConnectAppliance
 from requests import HTTPError
 import voluptuous as vol
 
@@ -91,7 +92,7 @@ PLATFORMS = [
 
 def _get_appliance_by_device_id(
     hass: HomeAssistant, device_id: str
-) -> api.HomeConnectDevice:
+) -> HomeConnectAppliance:
     """Return a Home Connect appliance instance given an device_id."""
     for hc_api in hass.data[DOMAIN].values():
         for device in hc_api.devices:

homeassistant/components/home_connect/diagnostics.py

@@ -4,22 +4,45 @@ from __future__ import annotations
 
 from typing import Any
 
+from homeconnect.api import HomeConnectAppliance
+
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers.device_registry import DeviceEntry
 
+from . import _get_appliance_by_device_id
+from .api import HomeConnectDevice
 from .const import DOMAIN
 
 
+def _generate_appliance_diagnostics(appliance: HomeConnectAppliance) -> dict[str, Any]:
+    return {
+        "status": appliance.status,
+        "programs": appliance.get_programs_available(),
+    }
+
+
+def _generate_entry_diagnostics(
+    devices: list[HomeConnectDevice],
+) -> dict[str, dict[str, Any]]:
+    return {
+        device.appliance.haId: _generate_appliance_diagnostics(device.appliance)
+        for device in devices
+    }
+
+
 async def async_get_config_entry_diagnostics(
     hass: HomeAssistant, config_entry: ConfigEntry
 ) -> dict[str, Any]:
     """Return diagnostics for a config entry."""
-    return {
+    return await hass.async_add_executor_job(
-        device.appliance.haId: {
+        _generate_entry_diagnostics, hass.data[DOMAIN][config_entry.entry_id].devices
-            "status": device.appliance.status,
+    )
-            "programs": await hass.async_add_executor_job(
+
-                device.appliance.get_programs_available
+
-            ),
+async def async_get_device_diagnostics(
-        }
+    hass: HomeAssistant, config_entry: ConfigEntry, device: DeviceEntry
-        for device in hass.data[DOMAIN][config_entry.entry_id].devices
+) -> dict[str, Any]:
-    }
+    """Return diagnostics for a device."""
+    appliance = _get_appliance_by_device_id(hass, device.id)
+    return await hass.async_add_executor_job(_generate_appliance_diagnostics, appliance)

tests/components/home_connect/snapshots/test_diagnostics.ambr

@@ -413,3 +413,56 @@
     }),
   })
 # ---
+# name: test_async_get_device_diagnostics
+  dict({
+    'programs': list([
+      'Dishcare.Dishwasher.Program.Auto1',
+      'Dishcare.Dishwasher.Program.Auto2',
+      'Dishcare.Dishwasher.Program.Auto3',
+      'Dishcare.Dishwasher.Program.Eco50',
+      'Dishcare.Dishwasher.Program.Quick45',
+    ]),
+    'status': dict({
+      'BSH.Common.Setting.AmbientLightBrightness': dict({
+        'type': 'Double',
+        'unit': '%',
+        'value': 70,
+      }),
+      'BSH.Common.Setting.AmbientLightColor': dict({
+        'type': 'BSH.Common.EnumType.AmbientLightColor',
+        'value': 'BSH.Common.EnumType.AmbientLightColor.Color43',
+      }),
+      'BSH.Common.Setting.AmbientLightCustomColor': dict({
+        'type': 'String',
+        'value': '#4a88f8',
+      }),
+      'BSH.Common.Setting.AmbientLightEnabled': dict({
+        'type': 'Boolean',
+        'value': True,
+      }),
+      'BSH.Common.Setting.ChildLock': dict({
+        'type': 'Boolean',
+        'value': False,
+      }),
+      'BSH.Common.Setting.PowerState': dict({
+        'type': 'BSH.Common.EnumType.PowerState',
+        'value': 'BSH.Common.EnumType.PowerState.On',
+      }),
+      'BSH.Common.Status.DoorState': dict({
+        'value': 'BSH.Common.EnumType.DoorState.Closed',
+      }),
+      'BSH.Common.Status.OperationState': dict({
+        'value': 'BSH.Common.EnumType.OperationState.Ready',
+      }),
+      'BSH.Common.Status.RemoteControlActive': dict({
+        'value': True,
+      }),
+      'BSH.Common.Status.RemoteControlStartAllowed': dict({
+        'value': True,
+      }),
+      'Refrigeration.Common.Status.Door.Refrigerator': dict({
+        'value': 'BSH.Common.EnumType.DoorState.Open',
+      }),
+    }),
+  })
+# ---

tests/components/home_connect/test_diagnostics.py

@@ -6,11 +6,14 @@ from unittest.mock import MagicMock
 import pytest
 from syrupy import SnapshotAssertion
 
+from homeassistant.components.home_connect.const import DOMAIN
 from homeassistant.components.home_connect.diagnostics import (
     async_get_config_entry_diagnostics,
+    async_get_device_diagnostics,
 )
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers import device_registry as dr
 
 from .conftest import get_all_appliances
 
@@ -26,10 +29,33 @@ async def test_async_get_config_entry_diagnostics(
     get_appliances: MagicMock,
     snapshot: SnapshotAssertion,
 ) -> None:
-    """Test setup and unload."""
+    """Test config entry diagnostics."""
     get_appliances.side_effect = get_all_appliances
     assert config_entry.state == ConfigEntryState.NOT_LOADED
     assert await integration_setup()
     assert config_entry.state == ConfigEntryState.LOADED
 
     assert await async_get_config_entry_diagnostics(hass, config_entry) == snapshot
+
+
+@pytest.mark.usefixtures("bypass_throttle")
+async def test_async_get_device_diagnostics(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    integration_setup: Callable[[], Awaitable[bool]],
+    setup_credentials: None,
+    get_appliances: MagicMock,
+    device_registry: dr.DeviceRegistry,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test device config entry diagnostics."""
+    get_appliances.side_effect = get_all_appliances
+    assert config_entry.state == ConfigEntryState.NOT_LOADED
+    assert await integration_setup()
+    assert config_entry.state == ConfigEntryState.LOADED
+
+    device = device_registry.async_get_device(
+        identifiers={(DOMAIN, "SIEMENS-HCS02DWH1-6BE58C26DCC1")}
+    )
+
+    assert await async_get_device_diagnostics(hass, config_entry, device) == snapshot