Add button platform to devolo Home Network (#85834)

* Add Start WPS button

* Add remaining buttons

* Set correct entity categories

* Inherit from DevoloEntity

* Fix mypy

* Apply feedback

* Raise on DevicePasswordProtected

* Fix ruff

* Really fix ruff

* Adapt to recent development

* Change error message

homeassistant/components/devolo_home_network/__init__.py

@@ -185,7 +185,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 @callback
 def platforms(device: Device) -> set[Platform]:
     """Assemble supported platforms."""
-    supported_platforms = {Platform.SENSOR, Platform.SWITCH}
+    supported_platforms = {Platform.BUTTON, Platform.SENSOR, Platform.SWITCH}
     if device.plcnet:
         supported_platforms.add(Platform.BINARY_SENSOR)
     if device.device and "wifi1" in device.device.features:

homeassistant/components/devolo_home_network/binary_sensor.py

@@ -20,7 +20,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
 from .const import CONNECTED_PLC_DEVICES, CONNECTED_TO_ROUTER, DOMAIN
-from .entity import DevoloEntity
+from .entity import DevoloCoordinatorEntity
 
 
 def _is_connected_to_router(entity: DevoloBinarySensorEntity) -> bool:
@@ -79,7 +79,9 @@ async def async_setup_entry(
     async_add_entities(entities)
 
 
-class DevoloBinarySensorEntity(DevoloEntity[LogicalNetwork], BinarySensorEntity):
+class DevoloBinarySensorEntity(
+    DevoloCoordinatorEntity[LogicalNetwork], BinarySensorEntity
+):
     """Representation of a devolo binary sensor."""
 
     def __init__(

homeassistant/components/devolo_home_network/button.py

@@ -0,0 +1,133 @@
+"""Platform for button integration."""
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+
+from devolo_plc_api.device import Device
+from devolo_plc_api.exceptions.device import DevicePasswordProtected, DeviceUnavailable
+
+from homeassistant.components.button import (
+    ButtonDeviceClass,
+    ButtonEntity,
+    ButtonEntityDescription,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import EntityCategory
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import DOMAIN, IDENTIFY, PAIRING, RESTART, START_WPS
+from .entity import DevoloEntity
+
+
+@dataclass
+class DevoloButtonRequiredKeysMixin:
+    """Mixin for required keys."""
+
+    press_func: Callable[[Device], Awaitable[bool]]
+
+
+@dataclass
+class DevoloButtonEntityDescription(
+    ButtonEntityDescription, DevoloButtonRequiredKeysMixin
+):
+    """Describes devolo button entity."""
+
+
+BUTTON_TYPES: dict[str, DevoloButtonEntityDescription] = {
+    IDENTIFY: DevoloButtonEntityDescription(
+        key=IDENTIFY,
+        entity_category=EntityCategory.DIAGNOSTIC,
+        icon="mdi:led-on",
+        press_func=lambda device: device.plcnet.async_identify_device_start(),  # type: ignore[union-attr]
+    ),
+    PAIRING: DevoloButtonEntityDescription(
+        key=PAIRING,
+        icon="mdi:plus-network-outline",
+        press_func=lambda device: device.plcnet.async_pair_device(),  # type: ignore[union-attr]
+    ),
+    RESTART: DevoloButtonEntityDescription(
+        key=RESTART,
+        device_class=ButtonDeviceClass.RESTART,
+        entity_category=EntityCategory.CONFIG,
+        press_func=lambda device: device.device.async_restart(),  # type: ignore[union-attr]
+    ),
+    START_WPS: DevoloButtonEntityDescription(
+        key=START_WPS,
+        icon="mdi:wifi-plus",
+        press_func=lambda device: device.device.async_start_wps(),  # type: ignore[union-attr]
+    ),
+}
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
+    """Get all devices and buttons and setup them via config entry."""
+    device: Device = hass.data[DOMAIN][entry.entry_id]["device"]
+
+    entities: list[DevoloButtonEntity] = []
+    if device.plcnet:
+        entities.append(
+            DevoloButtonEntity(
+                entry,
+                BUTTON_TYPES[IDENTIFY],
+                device,
+            )
+        )
+        entities.append(
+            DevoloButtonEntity(
+                entry,
+                BUTTON_TYPES[PAIRING],
+                device,
+            )
+        )
+    if device.device and "restart" in device.device.features:
+        entities.append(
+            DevoloButtonEntity(
+                entry,
+                BUTTON_TYPES[RESTART],
+                device,
+            )
+        )
+    if device.device and "wifi1" in device.device.features:
+        entities.append(
+            DevoloButtonEntity(
+                entry,
+                BUTTON_TYPES[START_WPS],
+                device,
+            )
+        )
+    async_add_entities(entities)
+
+
+class DevoloButtonEntity(DevoloEntity, ButtonEntity):
+    """Representation of a devolo button."""
+
+    entity_description: DevoloButtonEntityDescription
+
+    def __init__(
+        self,
+        entry: ConfigEntry,
+        description: DevoloButtonEntityDescription,
+        device: Device,
+    ) -> None:
+        """Initialize entity."""
+        self.entity_description = description
+        super().__init__(entry, device)
+
+    async def async_press(self) -> None:
+        """Handle the button press."""
+        try:
+            await self.entity_description.press_func(self.device)
+        except DevicePasswordProtected as ex:
+            self.entry.async_start_reauth(self.hass)
+            raise HomeAssistantError(
+                f"Device {self.entry.title} require re-authenticatication to set or change the password"
+            ) from ex
+        except DeviceUnavailable as ex:
+            raise HomeAssistantError(
+                f"Device {self.entry.title} did not respond"
+            ) from ex

homeassistant/components/devolo_home_network/const.py

@@ -10,7 +10,6 @@ from devolo_plc_api.device_api import (
 )
 
 DOMAIN = "devolo_home_network"
-
 PRODUCT = "product"
 SERIAL_NUMBER = "serial_number"
 TITLE = "title"
@@ -21,7 +20,11 @@ SHORT_UPDATE_INTERVAL = timedelta(seconds=15)
 CONNECTED_PLC_DEVICES = "connected_plc_devices"
 CONNECTED_TO_ROUTER = "connected_to_router"
 CONNECTED_WIFI_CLIENTS = "connected_wifi_clients"
+IDENTIFY = "identify"
 NEIGHBORING_WIFI_NETWORKS = "neighboring_wifi_networks"
+PAIRING = "pairing"
+RESTART = "restart"
+START_WPS = "start_wps"
 SWITCH_GUEST_WIFI = "switch_guest_wifi"
 SWITCH_LEDS = "switch_leds"
 

homeassistant/components/devolo_home_network/entity.py

@@ -12,7 +12,7 @@ from devolo_plc_api.device_api import (
 from devolo_plc_api.plcnet_api import LogicalNetwork
 
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.entity import DeviceInfo, Entity
 from homeassistant.helpers.update_coordinator import (
     CoordinatorEntity,
     DataUpdateCoordinator,
@@ -32,7 +32,7 @@ _DataT = TypeVar(
 )
 
 
-class DevoloEntity(CoordinatorEntity[DataUpdateCoordinator[_DataT]]):
+class DevoloEntity(Entity):
     """Representation of a devolo home network device."""
 
     _attr_has_entity_name = True
@@ -40,12 +40,9 @@ class DevoloEntity(CoordinatorEntity[DataUpdateCoordinator[_DataT]]):
     def __init__(
         self,
         entry: ConfigEntry,
-        coordinator: DataUpdateCoordinator[_DataT],
         device: Device,
     ) -> None:
         """Initialize a devolo home network device."""
-        super().__init__(coordinator)
-
         self.device = device
         self.entry = entry
 
@@ -59,3 +56,19 @@ class DevoloEntity(CoordinatorEntity[DataUpdateCoordinator[_DataT]]):
         )
         self._attr_translation_key = self.entity_description.key
         self._attr_unique_id = f"{device.serial_number}_{self.entity_description.key}"
+
+
+class DevoloCoordinatorEntity(
+    CoordinatorEntity[DataUpdateCoordinator[_DataT]], DevoloEntity
+):
+    """Representation of a coordinated devolo home network device."""
+
+    def __init__(
+        self,
+        entry: ConfigEntry,
+        coordinator: DataUpdateCoordinator[_DataT],
+        device: Device,
+    ) -> None:
+        """Initialize a devolo home network device."""
+        super().__init__(coordinator)
+        DevoloEntity.__init__(self, entry, device)

homeassistant/components/devolo_home_network/sensor.py

@@ -26,7 +26,7 @@ from .const import (
     DOMAIN,
     NEIGHBORING_WIFI_NETWORKS,
 )
-from .entity import DevoloEntity
+from .entity import DevoloCoordinatorEntity
 
 _DataT = TypeVar(
     "_DataT",
@@ -113,7 +113,7 @@ async def async_setup_entry(
     async_add_entities(entities)
 
 
-class DevoloSensorEntity(DevoloEntity[_DataT], SensorEntity):
+class DevoloSensorEntity(DevoloCoordinatorEntity[_DataT], SensorEntity):
     """Representation of a devolo sensor."""
 
     entity_description: DevoloSensorEntityDescription[_DataT]

homeassistant/components/devolo_home_network/strings.json

@@ -34,6 +34,20 @@
         "name": "Connected to router"
       }
     },
+    "button": {
+      "identify": {
+        "name": "Identify device with a blinking LED"
+      },
+      "pairing": {
+        "name": "Start PLC pairing"
+      },
+      "restart": {
+        "name": "Restart device"
+      },
+      "start_wps": {
+        "name": "Start WPS"
+      }
+    },
     "sensor": {
       "connected_plc_devices": {
         "name": "Connected PLC devices"

homeassistant/components/devolo_home_network/switch.py

@@ -17,7 +17,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
 from .const import DOMAIN, SWITCH_GUEST_WIFI, SWITCH_LEDS
-from .entity import DevoloEntity
+from .entity import DevoloCoordinatorEntity
 
 _DataT = TypeVar("_DataT", bound=WifiGuestAccessGet | bool)
 
@@ -88,7 +88,7 @@ async def async_setup_entry(
     async_add_entities(entities)
 
 
-class DevoloSwitchEntity(DevoloEntity[_DataT], SwitchEntity):
+class DevoloSwitchEntity(DevoloCoordinatorEntity[_DataT], SwitchEntity):
     """Representation of a devolo switch."""
 
     entity_description: DevoloSwitchEntityDescription[_DataT]

tests/components/devolo_home_network/const.py

@@ -38,7 +38,7 @@ DISCOVERY_INFO = ZeroconfServiceInfo(
         "Path": "abcdefghijkl/deviceapi",
         "Version": "v0",
         "Product": "dLAN pro 1200+ WiFi ac",
-        "Features": "reset,update,led,intmtg,wifi1",
+        "Features": "intmtg1,led,reset,restart,update,wifi1",
         "MT": "2730",
         "SN": "1234567890",
         "FirmwareVersion": "5.6.1",

tests/components/devolo_home_network/mock.py

@@ -51,6 +51,8 @@ class MockDevice(Device):
         self.async_disconnect = AsyncMock()
         self.device = DeviceApi(IP, None, DISCOVERY_INFO)
         self.device.async_get_led_setting = AsyncMock(return_value=False)
+        self.device.async_restart = AsyncMock(return_value=True)
+        self.device.async_start_wps = AsyncMock(return_value=True)
         self.device.async_get_wifi_connected_station = AsyncMock(
             return_value=CONNECTED_STATIONS
         )
@@ -60,3 +62,5 @@ class MockDevice(Device):
         )
         self.plcnet = PlcNetApi(IP, None, DISCOVERY_INFO)
         self.plcnet.async_get_network_overview = AsyncMock(return_value=PLCNET)
+        self.plcnet.async_identify_device_start = AsyncMock(return_value=True)
+        self.plcnet.async_pair_device = AsyncMock(return_value=True)

tests/components/devolo_home_network/test_button.py

@@ -0,0 +1,242 @@
+"""Tests for the devolo Home Network buttons."""
+from unittest.mock import AsyncMock
+
+from devolo_plc_api.exceptions.device import DevicePasswordProtected, DeviceUnavailable
+import pytest
+
+from homeassistant.components.button import (
+    DOMAIN as PLATFORM,
+    SERVICE_PRESS,
+    ButtonDeviceClass,
+)
+from homeassistant.components.devolo_home_network.const import DOMAIN
+from homeassistant.config_entries import SOURCE_REAUTH
+from homeassistant.const import ATTR_ENTITY_ID, STATE_UNKNOWN
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.entity import EntityCategory
+
+from . import configure_integration
+from .mock import MockDevice
+
+
+@pytest.mark.usefixtures("mock_device")
+async def test_button_setup(hass: HomeAssistant) -> None:
+    """Test default setup of the button component."""
+    entry = configure_integration(hass)
+    device_name = entry.title.replace(" ", "_").lower()
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert (
+        hass.states.get(f"{PLATFORM}.{device_name}_identify_device_with_a_blinking_led")
+        is not None
+    )
+    assert hass.states.get(f"{PLATFORM}.{device_name}_start_plc_pairing") is not None
+    assert hass.states.get(f"{PLATFORM}.{device_name}_restart_device") is not None
+    assert hass.states.get(f"{PLATFORM}.{device_name}_start_wps") is not None
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+@pytest.mark.freeze_time("2023-01-13 12:00:00+00:00")
+async def test_identify_device(
+    hass: HomeAssistant, mock_device: MockDevice, entity_registry: er.EntityRegistry
+) -> None:
+    """Test start PLC pairing button."""
+    entry = configure_integration(hass)
+    device_name = entry.title.replace(" ", "_").lower()
+    state_key = f"{PLATFORM}.{device_name}_identify_device_with_a_blinking_led"
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(state_key)
+    assert state is not None
+    assert state.state == STATE_UNKNOWN
+    assert (
+        entity_registry.async_get(state_key).entity_category
+        is EntityCategory.DIAGNOSTIC
+    )
+
+    # Emulate button press
+    await hass.services.async_call(
+        PLATFORM,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: state_key},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get(state_key)
+    assert state.state == "2023-01-13T12:00:00+00:00"
+    assert mock_device.plcnet.async_identify_device_start.call_count == 1
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+@pytest.mark.freeze_time("2023-01-13 12:00:00+00:00")
+async def test_start_plc_pairing(hass: HomeAssistant, mock_device: MockDevice) -> None:
+    """Test start PLC pairing button."""
+    entry = configure_integration(hass)
+    device_name = entry.title.replace(" ", "_").lower()
+    state_key = f"{PLATFORM}.{device_name}_start_plc_pairing"
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(state_key)
+    assert state is not None
+    assert state.state == STATE_UNKNOWN
+
+    # Emulate button press
+    await hass.services.async_call(
+        PLATFORM,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: state_key},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get(state_key)
+    assert state.state == "2023-01-13T12:00:00+00:00"
+    assert mock_device.plcnet.async_pair_device.call_count == 1
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+@pytest.mark.freeze_time("2023-01-13 12:00:00+00:00")
+async def test_restart(
+    hass: HomeAssistant, mock_device: MockDevice, entity_registry: er.EntityRegistry
+) -> None:
+    """Test restart button."""
+    entry = configure_integration(hass)
+    device_name = entry.title.replace(" ", "_").lower()
+    state_key = f"{PLATFORM}.{device_name}_restart_device"
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(state_key)
+    assert state is not None
+    assert state.state == STATE_UNKNOWN
+    assert state.attributes["device_class"] == ButtonDeviceClass.RESTART
+    assert entity_registry.async_get(state_key).entity_category is EntityCategory.CONFIG
+
+    # Emulate button press
+    await hass.services.async_call(
+        PLATFORM,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: state_key},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get(state_key)
+    assert state.state == "2023-01-13T12:00:00+00:00"
+    assert mock_device.device.async_restart.call_count == 1
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+@pytest.mark.freeze_time("2023-01-13 12:00:00+00:00")
+async def test_start_wps(hass: HomeAssistant, mock_device: MockDevice) -> None:
+    """Test start WPS button."""
+    entry = configure_integration(hass)
+    device_name = entry.title.replace(" ", "_").lower()
+    state_key = f"{PLATFORM}.{device_name}_start_wps"
+
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    state = hass.states.get(state_key)
+    assert state is not None
+    assert state.state == STATE_UNKNOWN
+
+    # Emulate button press
+    await hass.services.async_call(
+        PLATFORM,
+        SERVICE_PRESS,
+        {ATTR_ENTITY_ID: state_key},
+        blocking=True,
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get(state_key)
+    assert state.state == "2023-01-13T12:00:00+00:00"
+    assert mock_device.device.async_start_wps.call_count == 1
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+@pytest.mark.parametrize(
+    ("name", "trigger_method"),
+    [
+        ["identify_device_with_a_blinking_led", "async_identify_device_start"],
+        ["start_plc_pairing", "async_pair_device"],
+        ["restart_device", "async_restart"],
+        ["start_wps", "async_start_wps"],
+    ],
+)
+async def test_device_failure(
+    hass: HomeAssistant,
+    mock_device: MockDevice,
+    name: str,
+    trigger_method: str,
+) -> None:
+    """Test device failure."""
+    entry = configure_integration(hass)
+    device_name = entry.title.replace(" ", "_").lower()
+    state_key = f"{PLATFORM}.{device_name}_{name}"
+
+    setattr(mock_device.device, trigger_method, AsyncMock())
+    api = getattr(mock_device.device, trigger_method)
+    api.side_effect = DeviceUnavailable
+    setattr(mock_device.plcnet, trigger_method, AsyncMock())
+    api = getattr(mock_device.plcnet, trigger_method)
+    api.side_effect = DeviceUnavailable
+
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    # Emulate button press
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            PLATFORM,
+            SERVICE_PRESS,
+            {ATTR_ENTITY_ID: state_key},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+
+    await hass.config_entries.async_unload(entry.entry_id)
+
+
+async def test_auth_failed(hass: HomeAssistant, mock_device: MockDevice) -> None:
+    """Test setting unautherized triggers the reauth flow."""
+    entry = configure_integration(hass)
+    device_name = entry.title.replace(" ", "_").lower()
+    state_key = f"{PLATFORM}.{device_name}_start_wps"
+
+    await hass.config_entries.async_setup(entry.entry_id)
+    await hass.async_block_till_done()
+
+    mock_device.device.async_start_wps.side_effect = DevicePasswordProtected
+
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            PLATFORM,
+            SERVICE_PRESS,
+            {ATTR_ENTITY_ID: state_key},
+            blocking=True,
+        )
+        await hass.async_block_till_done()
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+
+    flow = flows[0]
+    assert flow["step_id"] == "reauth_confirm"
+    assert flow["handler"] == DOMAIN
+    assert "context" in flow
+    assert flow["context"]["source"] == SOURCE_REAUTH
+    assert flow["context"]["entry_id"] == entry.entry_id
+
+    await hass.config_entries.async_unload(entry.entry_id)

tests/components/devolo_home_network/test_init.py

@@ -5,6 +5,7 @@ from devolo_plc_api.exceptions.device import DeviceNotFound
 import pytest
 
 from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR
+from homeassistant.components.button import DOMAIN as BUTTON
 from homeassistant.components.device_tracker import DOMAIN as DEVICE_TRACKER
 from homeassistant.components.devolo_home_network.const import DOMAIN
 from homeassistant.components.sensor import DOMAIN as SENSOR
@@ -83,9 +84,9 @@ async def test_hass_stop(hass: HomeAssistant, mock_device: MockDevice) -> None:
 @pytest.mark.parametrize(
     ("device", "expected_platforms"),
     [
-        ["mock_device", (BINARY_SENSOR, DEVICE_TRACKER, SENSOR, SWITCH)],
+        ["mock_device", (BINARY_SENSOR, BUTTON, DEVICE_TRACKER, SENSOR, SWITCH)],
-        ["mock_repeater_device", (DEVICE_TRACKER, SENSOR, SWITCH)],
+        ["mock_repeater_device", (BUTTON, DEVICE_TRACKER, SENSOR, SWITCH)],
-        ["mock_nonwifi_device", (BINARY_SENSOR, SENSOR, SWITCH)],
+        ["mock_nonwifi_device", (BINARY_SENSOR, BUTTON, SENSOR, SWITCH)],
     ],
 )
 async def test_platforms(
@@ -93,7 +94,7 @@ async def test_platforms(
     device: str,
     expected_platforms: set[str],
     request: pytest.FixtureRequest,
-):
+) -> None:
     """Test platform assembly."""
     request.getfixturevalue(device)
     entry = configure_integration(hass)