Add update platform to devolo Home Network (#86003)

* Add update platform

* Take care of progress

* Adapt to recent development

* Only add platform if supported

* Avoid unneeded line change

* Fix ruff in tests

* Handle update failures like in button platform

* Apply suggestions

* Fix tests

* Remove unused logger
This commit is contained in:
Guido Schmitz 2023-08-21 20:59:58 +02:00 committed by GitHub
parent c39fc0766e
commit 365dc47740
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 341 additions and 4 deletions

View File

@ -9,6 +9,7 @@ from devolo_plc_api import Device
from devolo_plc_api.device_api import (
ConnectedStationInfo,
NeighborAPInfo,
UpdateFirmwareCheck,
WifiGuestAccessGet,
)
from devolo_plc_api.exceptions.device import (
@ -37,6 +38,7 @@ from .const import (
DOMAIN,
LONG_UPDATE_INTERVAL,
NEIGHBORING_WIFI_NETWORKS,
REGULAR_FIRMWARE,
SHORT_UPDATE_INTERVAL,
SWITCH_GUEST_WIFI,
SWITCH_LEDS,
@ -45,7 +47,9 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry( # noqa: C901
hass: HomeAssistant, entry: ConfigEntry
) -> bool:
"""Set up devolo Home Network from a config entry."""
hass.data.setdefault(DOMAIN, {})
zeroconf_instance = await zeroconf.async_get_async_instance(hass)
@ -66,6 +70,15 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
hass.data[DOMAIN][entry.entry_id] = {"device": device}
async def async_update_firmware_available() -> UpdateFirmwareCheck:
"""Fetch data from API endpoint."""
assert device.device
try:
async with asyncio.timeout(10):
return await device.device.async_check_firmware_available()
except DeviceUnavailable as err:
raise UpdateFailed(err) from err
async def async_update_connected_plc_devices() -> LogicalNetwork:
"""Fetch data from API endpoint."""
assert device.plcnet
@ -134,6 +147,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
update_method=async_update_led_status,
update_interval=SHORT_UPDATE_INTERVAL,
)
if device.device and "update" in device.device.features:
coordinators[REGULAR_FIRMWARE] = DataUpdateCoordinator(
hass,
_LOGGER,
name=REGULAR_FIRMWARE,
update_method=async_update_firmware_available,
update_interval=LONG_UPDATE_INTERVAL,
)
if device.device and "wifi1" in device.device.features:
coordinators[CONNECTED_WIFI_CLIENTS] = DataUpdateCoordinator(
hass,
@ -192,4 +213,6 @@ def platforms(device: Device) -> set[Platform]:
supported_platforms.add(Platform.BINARY_SENSOR)
if device.device and "wifi1" in device.device.features:
supported_platforms.add(Platform.DEVICE_TRACKER)
if device.device and "update" in device.device.features:
supported_platforms.add(Platform.UPDATE)
return supported_platforms

View File

@ -23,6 +23,7 @@ CONNECTED_WIFI_CLIENTS = "connected_wifi_clients"
IDENTIFY = "identify"
NEIGHBORING_WIFI_NETWORKS = "neighboring_wifi_networks"
PAIRING = "pairing"
REGULAR_FIRMWARE = "regular_firmware"
RESTART = "restart"
START_WPS = "start_wps"
SWITCH_GUEST_WIFI = "switch_guest_wifi"

View File

@ -0,0 +1,132 @@
"""Platform for update integration."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any
from devolo_plc_api.device import Device
from devolo_plc_api.device_api import UpdateFirmwareCheck
from devolo_plc_api.exceptions.device import DevicePasswordProtected, DeviceUnavailable
from homeassistant.components.update import (
UpdateDeviceClass,
UpdateEntity,
UpdateEntityDescription,
UpdateEntityFeature,
)
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 homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN, REGULAR_FIRMWARE
from .entity import DevoloCoordinatorEntity
@dataclass
class DevoloUpdateRequiredKeysMixin:
"""Mixin for required keys."""
latest_version: Callable[[UpdateFirmwareCheck], str]
update_func: Callable[[Device], Awaitable[bool]]
@dataclass
class DevoloUpdateEntityDescription(
UpdateEntityDescription, DevoloUpdateRequiredKeysMixin
):
"""Describes devolo update entity."""
UPDATE_TYPES: dict[str, DevoloUpdateEntityDescription] = {
REGULAR_FIRMWARE: DevoloUpdateEntityDescription(
key=REGULAR_FIRMWARE,
device_class=UpdateDeviceClass.FIRMWARE,
entity_category=EntityCategory.CONFIG,
latest_version=lambda data: data.new_firmware_version.split("_")[0],
update_func=lambda device: device.device.async_start_firmware_update(), # type: ignore[union-attr]
),
}
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Get all devices and sensors and setup them via config entry."""
device: Device = hass.data[DOMAIN][entry.entry_id]["device"]
coordinators: dict[str, DataUpdateCoordinator[Any]] = hass.data[DOMAIN][
entry.entry_id
]["coordinators"]
async_add_entities(
[
DevoloUpdateEntity(
entry,
coordinators[REGULAR_FIRMWARE],
UPDATE_TYPES[REGULAR_FIRMWARE],
device,
)
]
)
class DevoloUpdateEntity(DevoloCoordinatorEntity, UpdateEntity):
"""Representation of a devolo update."""
_attr_supported_features = (
UpdateEntityFeature.INSTALL | UpdateEntityFeature.PROGRESS
)
entity_description: DevoloUpdateEntityDescription
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator,
description: DevoloUpdateEntityDescription,
device: Device,
) -> None:
"""Initialize entity."""
self.entity_description = description
super().__init__(entry, coordinator, device)
self._attr_translation_key = None
self._in_progress_old_version: str | None = None
@property
def installed_version(self) -> str:
"""Version currently in use."""
return self.device.firmware_version
@property
def latest_version(self) -> str:
"""Latest version available for install."""
if latest_version := self.entity_description.latest_version(
self.coordinator.data
):
return latest_version
return self.device.firmware_version
@property
def in_progress(self) -> bool:
"""Update installation in progress."""
return self._in_progress_old_version == self.installed_version
async def async_install(
self, version: str | None, backup: bool, **kwargs: Any
) -> None:
"""Turn the entity on."""
self._in_progress_old_version = self.installed_version
try:
await self.entity_description.update_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

View File

@ -1,11 +1,13 @@
"""Constants used for mocking data."""
from devolo_plc_api.device_api import (
UPDATE_AVAILABLE,
WIFI_BAND_2G,
WIFI_BAND_5G,
WIFI_VAP_MAIN_AP,
ConnectedStationInfo,
NeighborAPInfo,
UpdateFirmwareCheck,
WifiGuestAccessGet,
)
from devolo_plc_api.plcnet_api import LogicalNetwork
@ -79,6 +81,10 @@ DISCOVERY_INFO_WRONG_DEVICE = ZeroconfServiceInfo(
type="mock_type",
)
FIRMWARE_UPDATE_AVAILABLE = UpdateFirmwareCheck(
result=UPDATE_AVAILABLE, new_firmware_version="5.6.2_2023-01-15"
)
GUEST_WIFI = WifiGuestAccessGet(
ssid="devolo-guest-930",
key="HMANPGBA",

View File

@ -13,6 +13,7 @@ from zeroconf.asyncio import AsyncZeroconf
from .const import (
CONNECTED_STATIONS,
DISCOVERY_INFO,
FIRMWARE_UPDATE_AVAILABLE,
GUEST_WIFI,
IP,
NEIGHBOR_ACCESS_POINTS,
@ -50,6 +51,9 @@ class MockDevice(Device):
"""Reset mock to starting point."""
self.async_disconnect = AsyncMock()
self.device = DeviceApi(IP, None, DISCOVERY_INFO)
self.device.async_check_firmware_available = AsyncMock(
return_value=FIRMWARE_UPDATE_AVAILABLE
)
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)
@ -60,6 +64,7 @@ class MockDevice(Device):
self.device.async_get_wifi_neighbor_access_points = AsyncMock(
return_value=NEIGHBOR_ACCESS_POINTS
)
self.device.async_start_firmware_update = AsyncMock(return_value=True)
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)

View File

@ -10,6 +10,7 @@ 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
from homeassistant.components.switch import DOMAIN as SWITCH
from homeassistant.components.update import DOMAIN as UPDATE
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_IP_ADDRESS, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
@ -84,9 +85,12 @@ async def test_hass_stop(hass: HomeAssistant, mock_device: MockDevice) -> None:
@pytest.mark.parametrize(
("device", "expected_platforms"),
[
["mock_device", (BINARY_SENSOR, BUTTON, DEVICE_TRACKER, SENSOR, SWITCH)],
["mock_repeater_device", (BUTTON, DEVICE_TRACKER, SENSOR, SWITCH)],
["mock_nonwifi_device", (BINARY_SENSOR, BUTTON, SENSOR, SWITCH)],
[
"mock_device",
(BINARY_SENSOR, BUTTON, DEVICE_TRACKER, SENSOR, SWITCH, UPDATE),
],
["mock_repeater_device", (BUTTON, DEVICE_TRACKER, SENSOR, SWITCH, UPDATE)],
["mock_nonwifi_device", (BINARY_SENSOR, BUTTON, SENSOR, SWITCH, UPDATE)],
],
)
async def test_platforms(

View File

@ -0,0 +1,166 @@
"""Tests for the devolo Home Network update."""
from devolo_plc_api.device_api import UPDATE_NOT_AVAILABLE, UpdateFirmwareCheck
from devolo_plc_api.exceptions.device import DevicePasswordProtected, DeviceUnavailable
import pytest
from homeassistant.components.devolo_home_network.const import (
DOMAIN,
LONG_UPDATE_INTERVAL,
)
from homeassistant.components.update import (
DOMAIN as PLATFORM,
SERVICE_INSTALL,
UpdateDeviceClass,
)
from homeassistant.config_entries import SOURCE_REAUTH
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON, STATE_UNAVAILABLE
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 homeassistant.util import dt as dt_util
from . import configure_integration
from .const import FIRMWARE_UPDATE_AVAILABLE
from .mock import MockDevice
from tests.common import async_fire_time_changed
@pytest.mark.usefixtures("mock_device")
async def test_update_setup(hass: HomeAssistant) -> None:
"""Test default setup of the update 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}_firmware") is not None
await hass.config_entries.async_unload(entry.entry_id)
async def test_update_firmware(
hass: HomeAssistant, mock_device: MockDevice, entity_registry: er.EntityRegistry
) -> None:
"""Test updating a device."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
state_key = f"{PLATFORM}.{device_name}_firmware"
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_ON
assert state.attributes["device_class"] == UpdateDeviceClass.FIRMWARE
assert state.attributes["installed_version"] == mock_device.firmware_version
assert (
state.attributes["latest_version"]
== FIRMWARE_UPDATE_AVAILABLE.new_firmware_version.split("_")[0]
)
assert entity_registry.async_get(state_key).entity_category == EntityCategory.CONFIG
await hass.services.async_call(
PLATFORM,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: state_key},
blocking=True,
)
assert mock_device.device.async_start_firmware_update.call_count == 1
# Emulate state change
mock_device.device.async_check_firmware_available.return_value = (
UpdateFirmwareCheck(result=UPDATE_NOT_AVAILABLE)
)
async_fire_time_changed(hass, dt_util.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_OFF
await hass.config_entries.async_unload(entry.entry_id)
async def test_device_failure_check(
hass: HomeAssistant, mock_device: MockDevice
) -> None:
"""Test device failure during check."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
state_key = f"{PLATFORM}.{device_name}_firmware"
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
mock_device.device.async_check_firmware_available.side_effect = DeviceUnavailable
async_fire_time_changed(hass, dt_util.utcnow() + LONG_UPDATE_INTERVAL)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
async def test_device_failure_update(
hass: HomeAssistant,
mock_device: MockDevice,
) -> None:
"""Test device failure when starting update."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
state_key = f"{PLATFORM}.{device_name}_firmware"
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
mock_device.device.async_start_firmware_update.side_effect = DeviceUnavailable
# Emulate update start
with pytest.raises(HomeAssistantError):
await hass.services.async_call(
PLATFORM,
SERVICE_INSTALL,
{ATTR_ENTITY_ID: state_key},
blocking=True,
)
await hass.config_entries.async_unload(entry.entry_id)
async def test_auth_failed(hass: HomeAssistant, mock_device: MockDevice) -> None:
"""Test updating unautherized triggers the reauth flow."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
state_key = f"{PLATFORM}.{device_name}_firmware"
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
mock_device.device.async_start_firmware_update.side_effect = DevicePasswordProtected
with pytest.raises(HomeAssistantError):
assert await hass.services.async_call(
PLATFORM,
SERVICE_INSTALL,
{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)