From f1884d34e9a68993d0461e99c612cbf1030a48f2 Mon Sep 17 00:00:00 2001
From: Guido Schmitz <Shutgun@users.noreply.github.com>
Date: Thu, 28 Oct 2021 22:42:10 +0200
Subject: [PATCH] Add devolo home network integration (#45866)
Co-authored-by: Markus Bong <2Fake1987@gmail.com>
Co-authored-by: Markus Bong <Markus.Bong@devolo.de>
---
.strict-typing | 1 +
CODEOWNERS | 1 +
.../devolo_home_network/__init__.py | 122 +++++++++++++
.../devolo_home_network/config_flow.py | 108 +++++++++++
.../components/devolo_home_network/const.py | 17 ++
.../components/devolo_home_network/entity.py | 37 ++++
.../devolo_home_network/manifest.json | 11 ++
.../components/devolo_home_network/sensor.py | 122 +++++++++++++
.../devolo_home_network/strings.json | 25 +++
.../devolo_home_network/translations/en.json | 25 +++
homeassistant/generated/config_flows.py | 1 +
homeassistant/generated/zeroconf.py | 3 +
mypy.ini | 11 ++
requirements_all.txt | 3 +
requirements_test_all.txt | 3 +
.../devolo_home_network/__init__.py | 31 ++++
.../devolo_home_network/conftest.py | 41 +++++
tests/components/devolo_home_network/const.py | 64 +++++++
.../devolo_home_network/test_config_flow.py | 172 ++++++++++++++++++
.../devolo_home_network/test_init.py | 61 +++++++
.../devolo_home_network/test_sensor.py | 148 +++++++++++++++
21 files changed, 1007 insertions(+)
create mode 100644 homeassistant/components/devolo_home_network/__init__.py
create mode 100644 homeassistant/components/devolo_home_network/config_flow.py
create mode 100644 homeassistant/components/devolo_home_network/const.py
create mode 100644 homeassistant/components/devolo_home_network/entity.py
create mode 100644 homeassistant/components/devolo_home_network/manifest.json
create mode 100644 homeassistant/components/devolo_home_network/sensor.py
create mode 100644 homeassistant/components/devolo_home_network/strings.json
create mode 100644 homeassistant/components/devolo_home_network/translations/en.json
create mode 100644 tests/components/devolo_home_network/__init__.py
create mode 100644 tests/components/devolo_home_network/conftest.py
create mode 100644 tests/components/devolo_home_network/const.py
create mode 100644 tests/components/devolo_home_network/test_config_flow.py
create mode 100644 tests/components/devolo_home_network/test_init.py
create mode 100644 tests/components/devolo_home_network/test_sensor.py
diff --git a/.strict-typing b/.strict-typing
index 685c87aa094..8977c9f68c8 100644
--- a/.strict-typing
+++ b/.strict-typing
@@ -32,6 +32,7 @@ homeassistant.components.crownstone.*
homeassistant.components.device_automation.*
homeassistant.components.device_tracker.*
homeassistant.components.devolo_home_control.*
+homeassistant.components.devolo_home_network.*
homeassistant.components.dlna_dmr.*
homeassistant.components.dnsip.*
homeassistant.components.dsmr.*
diff --git a/CODEOWNERS b/CODEOWNERS
index 2612cc3fd18..680424fbdea 100644
--- a/CODEOWNERS
+++ b/CODEOWNERS
@@ -118,6 +118,7 @@ homeassistant/components/denonavr/* @ol-iver @starkillerOG
homeassistant/components/derivative/* @afaucogney
homeassistant/components/device_automation/* @home-assistant/core
homeassistant/components/devolo_home_control/* @2Fake @Shutgun
+homeassistant/components/devolo_home_network/* @2Fake @Shutgun
homeassistant/components/dexcom/* @gagebenne
homeassistant/components/dhcp/* @bdraco
homeassistant/components/dht/* @thegardenmonkey
diff --git a/homeassistant/components/devolo_home_network/__init__.py b/homeassistant/components/devolo_home_network/__init__.py
new file mode 100644
index 00000000000..f427e5acbfc
--- /dev/null
+++ b/homeassistant/components/devolo_home_network/__init__.py
@@ -0,0 +1,122 @@
+"""The devolo Home Network integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import async_timeout
+from devolo_plc_api.device import Device
+from devolo_plc_api.exceptions.device import DeviceNotFound, DeviceUnavailable
+
+from homeassistant.components import zeroconf
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_IP_ADDRESS, EVENT_HOMEASSISTANT_STOP
+from homeassistant.core import Event, HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.httpx_client import get_async_client
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import (
+ CONNECTED_PLC_DEVICES,
+ CONNECTED_WIFI_CLIENTS,
+ DOMAIN,
+ LONG_UPDATE_INTERVAL,
+ NEIGHBORING_WIFI_NETWORKS,
+ PLATFORMS,
+ SHORT_UPDATE_INTERVAL,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(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)
+ async_client = get_async_client(hass)
+
+ try:
+ device = Device(
+ ip=entry.data[CONF_IP_ADDRESS], zeroconf_instance=zeroconf_instance
+ )
+ await device.async_connect(session_instance=async_client)
+ except DeviceNotFound as err:
+ raise ConfigEntryNotReady(
+ f"Unable to connect to {entry.data[CONF_IP_ADDRESS]}"
+ ) from err
+
+ async def async_update_connected_plc_devices() -> dict[str, Any]:
+ """Fetch data from API endpoint."""
+ try:
+ async with async_timeout.timeout(10):
+ return await device.plcnet.async_get_network_overview() # type: ignore[no-any-return, union-attr]
+ except DeviceUnavailable as err:
+ raise UpdateFailed(err) from err
+
+ async def async_update_wifi_connected_station() -> dict[str, Any]:
+ """Fetch data from API endpoint."""
+ try:
+ async with async_timeout.timeout(10):
+ return await device.device.async_get_wifi_connected_station() # type: ignore[no-any-return, union-attr]
+ except DeviceUnavailable as err:
+ raise UpdateFailed(err) from err
+
+ async def async_update_wifi_neighbor_access_points() -> dict[str, Any]:
+ """Fetch data from API endpoint."""
+ try:
+ async with async_timeout.timeout(30):
+ return await device.device.async_get_wifi_neighbor_access_points() # type: ignore[no-any-return, union-attr]
+ except DeviceUnavailable as err:
+ raise UpdateFailed(err) from err
+
+ async def disconnect(event: Event) -> None:
+ """Disconnect from device."""
+ await device.async_disconnect()
+
+ coordinators: dict[str, DataUpdateCoordinator] = {}
+ if device.plcnet:
+ coordinators[CONNECTED_PLC_DEVICES] = DataUpdateCoordinator(
+ hass,
+ _LOGGER,
+ name=CONNECTED_PLC_DEVICES,
+ update_method=async_update_connected_plc_devices,
+ update_interval=LONG_UPDATE_INTERVAL,
+ )
+ if device.device and "wifi1" in device.device.features:
+ coordinators[CONNECTED_WIFI_CLIENTS] = DataUpdateCoordinator(
+ hass,
+ _LOGGER,
+ name=CONNECTED_WIFI_CLIENTS,
+ update_method=async_update_wifi_connected_station,
+ update_interval=SHORT_UPDATE_INTERVAL,
+ )
+ coordinators[NEIGHBORING_WIFI_NETWORKS] = DataUpdateCoordinator(
+ hass,
+ _LOGGER,
+ name=NEIGHBORING_WIFI_NETWORKS,
+ update_method=async_update_wifi_neighbor_access_points,
+ update_interval=LONG_UPDATE_INTERVAL,
+ )
+
+ hass.data[DOMAIN][entry.entry_id] = {"device": device, "coordinators": coordinators}
+
+ for coordinator in coordinators.values():
+ await coordinator.async_config_entry_first_refresh()
+
+ hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+
+ entry.async_on_unload(
+ hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, disconnect)
+ )
+
+ return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+ """Unload a config entry."""
+ unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+ if unload_ok:
+ await hass.data[DOMAIN][entry.entry_id]["device"].async_disconnect()
+ hass.data[DOMAIN].pop(entry.entry_id)
+
+ return unload_ok
diff --git a/homeassistant/components/devolo_home_network/config_flow.py b/homeassistant/components/devolo_home_network/config_flow.py
new file mode 100644
index 00000000000..fa0ee983b69
--- /dev/null
+++ b/homeassistant/components/devolo_home_network/config_flow.py
@@ -0,0 +1,108 @@
+"""Config flow for devolo Home Network integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from devolo_plc_api.device import Device
+from devolo_plc_api.exceptions.device import DeviceNotFound
+import voluptuous as vol
+
+from homeassistant import config_entries, core
+from homeassistant.components import zeroconf
+from homeassistant.const import CONF_HOST, CONF_IP_ADDRESS, CONF_NAME
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers.httpx_client import get_async_client
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+
+from .const import DOMAIN, PRODUCT, SERIAL_NUMBER, TITLE
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema({vol.Required(CONF_IP_ADDRESS): str})
+
+
+async def validate_input(
+ hass: core.HomeAssistant, data: dict[str, Any]
+) -> dict[str, str]:
+ """Validate the user input allows us to connect.
+
+ Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
+ """
+ zeroconf_instance = await zeroconf.async_get_instance(hass)
+ async_client = get_async_client(hass)
+
+ device = Device(data[CONF_IP_ADDRESS], zeroconf_instance=zeroconf_instance)
+
+ await device.async_connect(session_instance=async_client)
+ await device.async_disconnect()
+
+ return {
+ SERIAL_NUMBER: str(device.serial_number),
+ TITLE: device.hostname.split(".")[0],
+ }
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+ """Handle a config flow for devolo Home Network."""
+
+ VERSION = 1
+
+ async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
+ """Handle the initial step."""
+ errors: dict = {}
+
+ if user_input is None:
+ return self.async_show_form(
+ step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+ )
+
+ try:
+ info = await validate_input(self.hass, user_input)
+ except DeviceNotFound:
+ errors["base"] = "cannot_connect"
+ except Exception: # pylint: disable=broad-except
+ _LOGGER.exception("Unexpected exception")
+ errors["base"] = "unknown"
+ else:
+ await self.async_set_unique_id(info[SERIAL_NUMBER], raise_on_progress=False)
+ self._abort_if_unique_id_configured()
+ return self.async_create_entry(title=info[TITLE], data=user_input)
+
+ return self.async_show_form(
+ step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+ )
+
+ async def async_step_zeroconf(
+ self, discovery_info: DiscoveryInfoType
+ ) -> FlowResult:
+ """Handle zerooconf discovery."""
+ if discovery_info["properties"]["MT"] in ["2600", "2601"]:
+ return self.async_abort(reason="home_control")
+
+ await self.async_set_unique_id(discovery_info["properties"]["SN"])
+ self._abort_if_unique_id_configured()
+
+ # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
+ self.context[CONF_HOST] = discovery_info["host"]
+ self.context["title_placeholders"] = {
+ PRODUCT: discovery_info["properties"]["Product"],
+ CONF_NAME: discovery_info["hostname"].split(".")[0],
+ }
+
+ return await self.async_step_zeroconf_confirm()
+
+ async def async_step_zeroconf_confirm(
+ self, user_input: ConfigType | None = None
+ ) -> FlowResult:
+ """Handle a flow initiated by zeroconf."""
+ title = self.context["title_placeholders"][CONF_NAME]
+ if user_input is not None:
+ data = {
+ CONF_IP_ADDRESS: self.context[CONF_HOST],
+ }
+ return self.async_create_entry(title=title, data=data)
+ return self.async_show_form(
+ step_id="zeroconf_confirm",
+ description_placeholders={"host_name": title},
+ )
diff --git a/homeassistant/components/devolo_home_network/const.py b/homeassistant/components/devolo_home_network/const.py
new file mode 100644
index 00000000000..9276acebe41
--- /dev/null
+++ b/homeassistant/components/devolo_home_network/const.py
@@ -0,0 +1,17 @@
+"""Constants for the devolo Home Network integration."""
+
+from datetime import timedelta
+
+DOMAIN = "devolo_home_network"
+PLATFORMS = ["sensor"]
+
+PRODUCT = "product"
+SERIAL_NUMBER = "serial_number"
+TITLE = "title"
+
+LONG_UPDATE_INTERVAL = timedelta(minutes=5)
+SHORT_UPDATE_INTERVAL = timedelta(seconds=15)
+
+CONNECTED_PLC_DEVICES = "connected_plc_devices"
+CONNECTED_WIFI_CLIENTS = "connected_wifi_clients"
+NEIGHBORING_WIFI_NETWORKS = "neighboring_wifi_networks"
diff --git a/homeassistant/components/devolo_home_network/entity.py b/homeassistant/components/devolo_home_network/entity.py
new file mode 100644
index 00000000000..dbfe0e4035a
--- /dev/null
+++ b/homeassistant/components/devolo_home_network/entity.py
@@ -0,0 +1,37 @@
+"""Generic platform."""
+from __future__ import annotations
+
+from devolo_plc_api.device import Device
+
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.helpers.update_coordinator import (
+ CoordinatorEntity,
+ DataUpdateCoordinator,
+)
+
+from .const import DOMAIN
+
+
+class DevoloEntity(CoordinatorEntity):
+ """Representation of a devolo home network device."""
+
+ def __init__(
+ self, coordinator: DataUpdateCoordinator, device: Device, device_name: str
+ ) -> None:
+ """Initialize a devolo home network device."""
+ super().__init__(coordinator)
+
+ self._device = device
+ self._device_name = device_name
+
+ self._attr_device_info = DeviceInfo(
+ configuration_url=f"http://{self._device.ip}",
+ identifiers={(DOMAIN, str(self._device.serial_number))},
+ manufacturer="devolo",
+ model=self._device.product,
+ name=self._device_name,
+ sw_version=self._device.firmware_version,
+ )
+ self._attr_unique_id = (
+ f"{self._device.serial_number}_{self.entity_description.key}"
+ )
diff --git a/homeassistant/components/devolo_home_network/manifest.json b/homeassistant/components/devolo_home_network/manifest.json
new file mode 100644
index 00000000000..987211ca631
--- /dev/null
+++ b/homeassistant/components/devolo_home_network/manifest.json
@@ -0,0 +1,11 @@
+{
+ "domain": "devolo_home_network",
+ "name": "devolo Home Network",
+ "config_flow": true,
+ "documentation": "https://www.home-assistant.io/integrations/devolo_home_network",
+ "requirements": ["devolo-plc-api==0.6.2"],
+ "zeroconf": ["_dvl-deviceapi._tcp.local."],
+ "codeowners": ["@2Fake", "@Shutgun"],
+ "quality_scale": "platinum",
+ "iot_class": "local_polling"
+}
diff --git a/homeassistant/components/devolo_home_network/sensor.py b/homeassistant/components/devolo_home_network/sensor.py
new file mode 100644
index 00000000000..3b0175d8c31
--- /dev/null
+++ b/homeassistant/components/devolo_home_network/sensor.py
@@ -0,0 +1,122 @@
+"""Platform for sensor integration."""
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Any
+
+from devolo_plc_api.device import Device
+
+from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import HomeAssistantType
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import (
+ CONNECTED_PLC_DEVICES,
+ CONNECTED_WIFI_CLIENTS,
+ DOMAIN,
+ NEIGHBORING_WIFI_NETWORKS,
+)
+from .entity import DevoloEntity
+
+
+@dataclass
+class DevoloSensorRequiredKeysMixin:
+ """Mixin for required keys."""
+
+ value_func: Callable[[dict[str, Any]], int]
+
+
+@dataclass
+class DevoloSensorEntityDescription(
+ SensorEntityDescription, DevoloSensorRequiredKeysMixin
+):
+ """Describes devolo sensor entity."""
+
+
+SENSOR_TYPES: dict[str, DevoloSensorEntityDescription] = {
+ CONNECTED_PLC_DEVICES: DevoloSensorEntityDescription(
+ key=CONNECTED_PLC_DEVICES,
+ entity_registry_enabled_default=False,
+ icon="mdi:lan",
+ name="Connected PLC devices",
+ value_func=lambda data: len(
+ {device["mac_address_from"] for device in data["network"]["data_rates"]}
+ ),
+ ),
+ CONNECTED_WIFI_CLIENTS: DevoloSensorEntityDescription(
+ key=CONNECTED_WIFI_CLIENTS,
+ entity_registry_enabled_default=True,
+ icon="mdi:wifi",
+ name="Connected Wifi clients",
+ value_func=lambda data: len(data["connected_stations"]),
+ ),
+ NEIGHBORING_WIFI_NETWORKS: DevoloSensorEntityDescription(
+ key=NEIGHBORING_WIFI_NETWORKS,
+ entity_registry_enabled_default=False,
+ icon="mdi:wifi-marker",
+ name="Neighboring Wifi networks",
+ value_func=lambda data: len(data["neighbor_aps"]),
+ ),
+}
+
+
+async def async_setup_entry(
+ hass: HomeAssistantType, 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] = hass.data[DOMAIN][entry.entry_id][
+ "coordinators"
+ ]
+
+ entities: list[DevoloSensorEntity] = []
+ if device.plcnet:
+ entities.append(
+ DevoloSensorEntity(
+ coordinators[CONNECTED_PLC_DEVICES],
+ SENSOR_TYPES[CONNECTED_PLC_DEVICES],
+ device,
+ entry.title,
+ )
+ )
+ if device.device and "wifi1" in device.device.features:
+ entities.append(
+ DevoloSensorEntity(
+ coordinators[CONNECTED_WIFI_CLIENTS],
+ SENSOR_TYPES[CONNECTED_WIFI_CLIENTS],
+ device,
+ entry.title,
+ )
+ )
+ entities.append(
+ DevoloSensorEntity(
+ coordinators[NEIGHBORING_WIFI_NETWORKS],
+ SENSOR_TYPES[NEIGHBORING_WIFI_NETWORKS],
+ device,
+ entry.title,
+ )
+ )
+ async_add_entities(entities)
+
+
+class DevoloSensorEntity(DevoloEntity, SensorEntity):
+ """Representation of a devolo sensor."""
+
+ def __init__(
+ self,
+ coordinator: DataUpdateCoordinator,
+ description: DevoloSensorEntityDescription,
+ device: Device,
+ device_name: str,
+ ) -> None:
+ """Initialize entity."""
+ self.entity_description: DevoloSensorEntityDescription = description
+ super().__init__(coordinator, device, device_name)
+
+ @property
+ def native_value(self) -> int:
+ """State of the sensor."""
+ return self.entity_description.value_func(self.coordinator.data)
diff --git a/homeassistant/components/devolo_home_network/strings.json b/homeassistant/components/devolo_home_network/strings.json
new file mode 100644
index 00000000000..685e139d2b8
--- /dev/null
+++ b/homeassistant/components/devolo_home_network/strings.json
@@ -0,0 +1,25 @@
+{
+ "config": {
+ "flow_title": "{product} ({name})",
+ "step": {
+ "user": {
+ "description": "[%key:common::config_flow::description::confirm_setup%]",
+ "data": {
+ "ip_address": "[%key:common::config_flow::data::ip%]"
+ }
+ },
+ "zeroconf_confirm": {
+ "description": "Do you want to add the devolo home network device with the hostname `{host_name}` to Home Assistant?",
+ "title": "Discovered devolo home network device"
+ }
+ },
+ "error": {
+ "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+ "unknown": "[%key:common::config_flow::error::unknown%]"
+ },
+ "abort": {
+ "already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
+ "home_control": "The devolo Home Control Central Unit does not work with this integration."
+ }
+ }
+}
\ No newline at end of file
diff --git a/homeassistant/components/devolo_home_network/translations/en.json b/homeassistant/components/devolo_home_network/translations/en.json
new file mode 100644
index 00000000000..52e51d953c1
--- /dev/null
+++ b/homeassistant/components/devolo_home_network/translations/en.json
@@ -0,0 +1,25 @@
+{
+ "config": {
+ "abort": {
+ "already_configured": "Device is already configured"
+ },
+ "error": {
+ "cannot_connect": "Failed to connect",
+ "invalid_auth": "Invalid authentication",
+ "unknown": "Unexpected error"
+ },
+ "flow_title": "{product} ({name})",
+ "step": {
+ "user": {
+ "data": {
+ "ip_address": "IP Address"
+ },
+ "description": "Do you want to start set up?"
+ },
+ "zeroconf_confirm": {
+ "description": "Do you want to add the devolo home network device with the hostname `{host_name}` to Home Assistant?",
+ "title": "Discovered devolo home network device"
+ }
+ }
+ }
+}
\ No newline at end of file
diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py
index aef37105170..b45c06b10b5 100644
--- a/homeassistant/generated/config_flows.py
+++ b/homeassistant/generated/config_flows.py
@@ -60,6 +60,7 @@ FLOWS = [
"deconz",
"denonavr",
"devolo_home_control",
+ "devolo_home_network",
"dexcom",
"dialogflow",
"directv",
diff --git a/homeassistant/generated/zeroconf.py b/homeassistant/generated/zeroconf.py
index 8ab7a7f8e31..aec93dd36c9 100644
--- a/homeassistant/generated/zeroconf.py
+++ b/homeassistant/generated/zeroconf.py
@@ -58,6 +58,9 @@ ZEROCONF = {
"_dvl-deviceapi._tcp.local.": [
{
"domain": "devolo_home_control"
+ },
+ {
+ "domain": "devolo_home_network"
}
],
"_easylink._tcp.local.": [
diff --git a/mypy.ini b/mypy.ini
index 5e9501d5407..4192e1a10ea 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -363,6 +363,17 @@ no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
+[mypy-homeassistant.components.devolo_home_network.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+no_implicit_optional = true
+warn_return_any = true
+warn_unreachable = true
+
[mypy-homeassistant.components.dlna_dmr.*]
check_untyped_defs = true
disallow_incomplete_defs = true
diff --git a/requirements_all.txt b/requirements_all.txt
index dbbd93deb1b..866ebdc8d29 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -536,6 +536,9 @@ denonavr==0.10.9
# homeassistant.components.devolo_home_control
devolo-home-control-api==0.17.4
+# homeassistant.components.devolo_home_network
+devolo-plc-api==0.6.2
+
# homeassistant.components.directv
directv==0.4.0
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index 7dff0f66554..a11eb5c9fff 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -335,6 +335,9 @@ denonavr==0.10.9
# homeassistant.components.devolo_home_control
devolo-home-control-api==0.17.4
+# homeassistant.components.devolo_home_network
+devolo-plc-api==0.6.2
+
# homeassistant.components.directv
directv==0.4.0
diff --git a/tests/components/devolo_home_network/__init__.py b/tests/components/devolo_home_network/__init__.py
new file mode 100644
index 00000000000..b9026d8453b
--- /dev/null
+++ b/tests/components/devolo_home_network/__init__.py
@@ -0,0 +1,31 @@
+"""Tests for the devolo Home Network integration."""
+
+from typing import Any
+
+from devolo_plc_api.device_api.deviceapi import DeviceApi
+from devolo_plc_api.plcnet_api.plcnetapi import PlcNetApi
+
+from homeassistant.components.devolo_home_network.const import DOMAIN
+from homeassistant.const import CONF_IP_ADDRESS
+from homeassistant.core import HomeAssistant
+
+from .const import DISCOVERY_INFO, IP
+
+from tests.common import MockConfigEntry
+
+
+def configure_integration(hass: HomeAssistant) -> MockConfigEntry:
+ """Configure the integration."""
+ config = {
+ CONF_IP_ADDRESS: IP,
+ }
+ entry = MockConfigEntry(domain=DOMAIN, data=config)
+ entry.add_to_hass(hass)
+
+ return entry
+
+
+async def async_connect(self, session_instance: Any = None):
+ """Give a mocked device the needed properties."""
+ self.plcnet = PlcNetApi(IP, None, DISCOVERY_INFO)
+ self.device = DeviceApi(IP, None, DISCOVERY_INFO)
diff --git a/tests/components/devolo_home_network/conftest.py b/tests/components/devolo_home_network/conftest.py
new file mode 100644
index 00000000000..ab798cd5cfd
--- /dev/null
+++ b/tests/components/devolo_home_network/conftest.py
@@ -0,0 +1,41 @@
+"""Fixtures for tests."""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from . import async_connect
+from .const import CONNECTED_STATIONS, DISCOVERY_INFO, NEIGHBOR_ACCESS_POINTS, PLCNET
+
+
+@pytest.fixture()
+def mock_device():
+ """Mock connecting to a devolo home network device."""
+ with patch("devolo_plc_api.device.Device.async_connect", async_connect), patch(
+ "devolo_plc_api.device.Device.async_disconnect"
+ ), patch(
+ "devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_connected_station",
+ new=AsyncMock(return_value=CONNECTED_STATIONS),
+ ), patch(
+ "devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_neighbor_access_points",
+ new=AsyncMock(return_value=NEIGHBOR_ACCESS_POINTS),
+ ), patch(
+ "devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi.async_get_network_overview",
+ new=AsyncMock(return_value=PLCNET),
+ ):
+ yield
+
+
+@pytest.fixture(name="info")
+def mock_validate_input():
+ """Mock setup entry and user input."""
+ info = {
+ "serial_number": DISCOVERY_INFO["properties"]["SN"],
+ "title": DISCOVERY_INFO["properties"]["Product"],
+ }
+
+ with patch(
+ "homeassistant.components.devolo_home_network.config_flow.validate_input",
+ return_value=info,
+ ):
+ yield info
diff --git a/tests/components/devolo_home_network/const.py b/tests/components/devolo_home_network/const.py
new file mode 100644
index 00000000000..e9b2113c4a6
--- /dev/null
+++ b/tests/components/devolo_home_network/const.py
@@ -0,0 +1,64 @@
+"""Constants used for mocking data."""
+
+IP = "1.1.1.1"
+
+CONNECTED_STATIONS = {
+ "connected_stations": [
+ {
+ "mac_address": "AA:BB:CC:DD:EE:FF",
+ "vap_type": "WIFI_VAP_MAIN_AP",
+ "band": "WIFI_BAND_5G",
+ "rx_rate": 87800,
+ "tx_rate": 87800,
+ }
+ ],
+}
+
+DISCOVERY_INFO = {
+ "host": IP,
+ "port": 14791,
+ "hostname": "test.local.",
+ "type": "_dvl-deviceapi._tcp.local.",
+ "name": "dLAN pro 1200+ WiFi ac._dvl-deviceapi._tcp.local.",
+ "properties": {
+ "Path": "abcdefghijkl/deviceapi",
+ "Version": "v0",
+ "Product": "dLAN pro 1200+ WiFi ac",
+ "Features": "reset,update,led,intmtg,wifi1",
+ "MT": "2730",
+ "SN": "1234567890",
+ "FirmwareVersion": "5.6.1",
+ "FirmwareDate": "2020-10-23",
+ "PS": "",
+ "PlcMacAddress": "AA:BB:CC:DD:EE:FF",
+ },
+}
+
+DISCOVERY_INFO_WRONG_DEVICE = {"properties": {"MT": "2600"}}
+
+NEIGHBOR_ACCESS_POINTS = {
+ "neighbor_aps": [
+ {
+ "mac_address": "AA:BB:CC:DD:EE:FF",
+ "ssid": "wifi",
+ "band": "WIFI_BAND_2G",
+ "channel": 1,
+ "signal": -73,
+ "signal_bars": 1,
+ }
+ ]
+}
+
+PLCNET = {
+ "network": {
+ "data_rates": [
+ {
+ "mac_address_from": "AA:BB:CC:DD:EE:FF",
+ "mac_address_to": "11:22:33:44:55:66",
+ "rx_rate": 0.0,
+ "tx_rate": 0.0,
+ },
+ ],
+ "devices": [],
+ }
+}
diff --git a/tests/components/devolo_home_network/test_config_flow.py b/tests/components/devolo_home_network/test_config_flow.py
new file mode 100644
index 00000000000..0be07be9a00
--- /dev/null
+++ b/tests/components/devolo_home_network/test_config_flow.py
@@ -0,0 +1,172 @@
+"""Test the devolo Home Network config flow."""
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import patch
+
+from devolo_plc_api.exceptions.device import DeviceNotFound
+import pytest
+
+from homeassistant import config_entries
+from homeassistant.components.devolo_home_network import config_flow
+from homeassistant.components.devolo_home_network.const import (
+ DOMAIN,
+ SERIAL_NUMBER,
+ TITLE,
+)
+from homeassistant.const import CONF_BASE, CONF_IP_ADDRESS, CONF_NAME
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import (
+ RESULT_TYPE_ABORT,
+ RESULT_TYPE_CREATE_ENTRY,
+ RESULT_TYPE_FORM,
+)
+
+from .const import DISCOVERY_INFO, DISCOVERY_INFO_WRONG_DEVICE, IP
+
+
+async def test_form(hass: HomeAssistant, info: dict[str, Any]):
+ """Test we get the form."""
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": config_entries.SOURCE_USER}
+ )
+ assert result["type"] == RESULT_TYPE_FORM
+ assert result["errors"] == {}
+
+ with patch(
+ "homeassistant.components.devolo_home_network.async_setup_entry",
+ return_value=True,
+ ) as mock_setup_entry:
+ result2 = await hass.config_entries.flow.async_configure(
+ result["flow_id"],
+ {
+ CONF_IP_ADDRESS: IP,
+ },
+ )
+ await hass.async_block_till_done()
+
+ assert result2["type"] == RESULT_TYPE_CREATE_ENTRY
+ assert result2["result"].unique_id == info["serial_number"]
+ assert result2["title"] == info["title"]
+ assert result2["data"] == {
+ CONF_IP_ADDRESS: IP,
+ }
+ assert len(mock_setup_entry.mock_calls) == 1
+
+
+@pytest.mark.parametrize(
+ "exception_type, expected_error",
+ [[DeviceNotFound, "cannot_connect"], [Exception, "unknown"]],
+)
+async def test_form_error(hass: HomeAssistant, exception_type, expected_error):
+ """Test we handle errors."""
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": config_entries.SOURCE_USER}
+ )
+
+ with patch(
+ "homeassistant.components.devolo_home_network.config_flow.validate_input",
+ side_effect=exception_type,
+ ):
+ result2 = await hass.config_entries.flow.async_configure(
+ result["flow_id"],
+ {
+ CONF_IP_ADDRESS: IP,
+ },
+ )
+
+ assert result2["type"] == RESULT_TYPE_FORM
+ assert result2["errors"] == {CONF_BASE: expected_error}
+
+
+async def test_zeroconf(hass: HomeAssistant):
+ """Test that the zeroconf form is served."""
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN,
+ context={"source": config_entries.SOURCE_ZEROCONF},
+ data=DISCOVERY_INFO,
+ )
+
+ assert result["step_id"] == "zeroconf_confirm"
+ assert result["type"] == RESULT_TYPE_FORM
+ assert result["description_placeholders"] == {"host_name": "test"}
+
+ context = next(
+ flow["context"]
+ for flow in hass.config_entries.flow.async_progress()
+ if flow["flow_id"] == result["flow_id"]
+ )
+
+ assert (
+ context["title_placeholders"][CONF_NAME]
+ == DISCOVERY_INFO["hostname"].split(".", maxsplit=1)[0]
+ )
+
+ result2 = await hass.config_entries.flow.async_configure(
+ result["flow_id"],
+ {},
+ )
+ await hass.async_block_till_done()
+
+ assert result2["title"] == "test"
+ assert result2["data"] == {
+ CONF_IP_ADDRESS: IP,
+ }
+
+
+async def test_abort_zeroconf_wrong_device(hass: HomeAssistant):
+ """Test we abort zeroconf for wrong devices."""
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN,
+ context={"source": config_entries.SOURCE_ZEROCONF},
+ data=DISCOVERY_INFO_WRONG_DEVICE,
+ )
+ assert result["type"] == RESULT_TYPE_ABORT
+ assert result["reason"] == "home_control"
+
+
+@pytest.mark.usefixtures("info")
+async def test_abort_if_configued(hass: HomeAssistant):
+ """Test we abort config flow if already configured."""
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": config_entries.SOURCE_USER}
+ )
+ await hass.config_entries.flow.async_configure(
+ result["flow_id"],
+ {
+ CONF_IP_ADDRESS: IP,
+ },
+ )
+ await hass.async_block_till_done()
+
+ # Abort on concurrent user flow
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": config_entries.SOURCE_USER}
+ )
+ result2 = await hass.config_entries.flow.async_configure(
+ result["flow_id"],
+ {
+ CONF_IP_ADDRESS: IP,
+ },
+ )
+ await hass.async_block_till_done()
+ assert result2["type"] == RESULT_TYPE_ABORT
+ assert result2["reason"] == "already_configured"
+
+ # Abort on concurrent zeroconf discovery flow
+ result3 = await hass.config_entries.flow.async_init(
+ DOMAIN,
+ context={"source": config_entries.SOURCE_ZEROCONF},
+ data=DISCOVERY_INFO,
+ )
+ assert result3["type"] == RESULT_TYPE_ABORT
+ assert result3["reason"] == "already_configured"
+
+
+@pytest.mark.usefixtures("mock_device")
+@pytest.mark.usefixtures("mock_zeroconf")
+async def test_validate_input(hass: HomeAssistant):
+ """Test input validaton."""
+ info = await config_flow.validate_input(hass, {CONF_IP_ADDRESS: IP})
+ assert SERIAL_NUMBER in info
+ assert TITLE in info
diff --git a/tests/components/devolo_home_network/test_init.py b/tests/components/devolo_home_network/test_init.py
new file mode 100644
index 00000000000..66d32e8974d
--- /dev/null
+++ b/tests/components/devolo_home_network/test_init.py
@@ -0,0 +1,61 @@
+"""Test the devolo Home Network integration setup."""
+from unittest.mock import patch
+
+from devolo_plc_api.exceptions.device import DeviceNotFound
+import pytest
+
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.const import EVENT_HOMEASSISTANT_STOP
+from homeassistant.core import HomeAssistant
+
+from . import configure_integration
+
+
+@pytest.mark.usefixtures("mock_device")
+@pytest.mark.usefixtures("mock_zeroconf")
+async def test_setup_entry(hass: HomeAssistant):
+ """Test setup entry."""
+ entry = configure_integration(hass)
+ with patch(
+ "homeassistant.config_entries.ConfigEntries.async_forward_entry_setup",
+ return_value=True,
+ ), patch("homeassistant.core.EventBus.async_listen_once"):
+ assert await hass.config_entries.async_setup(entry.entry_id)
+ assert entry.state is ConfigEntryState.LOADED
+
+
+@pytest.mark.usefixtures("mock_zeroconf")
+async def test_setup_device_not_found(hass: HomeAssistant):
+ """Test setup entry."""
+ entry = configure_integration(hass)
+ with patch(
+ "homeassistant.components.devolo_home_network.Device.async_connect",
+ side_effect=DeviceNotFound,
+ ):
+ await hass.config_entries.async_setup(entry.entry_id)
+ assert entry.state is ConfigEntryState.SETUP_RETRY
+
+
+@pytest.mark.usefixtures("mock_device")
+@pytest.mark.usefixtures("mock_zeroconf")
+async def test_unload_entry(hass: HomeAssistant):
+ """Test unload entry."""
+ entry = configure_integration(hass)
+ await hass.config_entries.async_setup(entry.entry_id)
+ await hass.async_block_till_done()
+ await hass.config_entries.async_unload(entry.entry_id)
+ assert entry.state is ConfigEntryState.NOT_LOADED
+
+
+@pytest.mark.usefixtures("mock_device")
+@pytest.mark.usefixtures("mock_zeroconf")
+async def test_hass_stop(hass: HomeAssistant):
+ """Test homeassistant stop event."""
+ entry = configure_integration(hass)
+ with patch(
+ "homeassistant.components.devolo_home_network.Device.async_disconnect"
+ ) as async_disconnect:
+ await hass.config_entries.async_setup(entry.entry_id)
+ await hass.async_block_till_done()
+ hass.bus.async_fire(EVENT_HOMEASSISTANT_STOP)
+ assert async_disconnect.assert_called_once
diff --git a/tests/components/devolo_home_network/test_sensor.py b/tests/components/devolo_home_network/test_sensor.py
new file mode 100644
index 00000000000..100db9005aa
--- /dev/null
+++ b/tests/components/devolo_home_network/test_sensor.py
@@ -0,0 +1,148 @@
+"""Tests for the devolo Home Network sensors."""
+from unittest.mock import patch
+
+from devolo_plc_api.exceptions.device import DeviceUnavailable
+import pytest
+
+from homeassistant.components.devolo_home_network.const import (
+ LONG_UPDATE_INTERVAL,
+ SHORT_UPDATE_INTERVAL,
+)
+from homeassistant.components.sensor import DOMAIN
+from homeassistant.const import STATE_UNAVAILABLE
+from homeassistant.core import HomeAssistant
+from homeassistant.util import dt
+
+from . import configure_integration
+
+from tests.common import async_fire_time_changed
+
+
+@pytest.mark.usefixtures("mock_device")
+@pytest.mark.usefixtures("mock_zeroconf")
+async def test_sensor_setup(hass: HomeAssistant):
+ """Test default setup of the sensor component."""
+ entry = configure_integration(hass)
+ await hass.config_entries.async_setup(entry.entry_id)
+ await hass.async_block_till_done()
+
+ assert hass.states.get(f"{DOMAIN}.connected_wifi_clients") is not None
+ assert hass.states.get(f"{DOMAIN}.connected_plc_devices") is None
+ assert hass.states.get(f"{DOMAIN}.neighboring_wifi_networks") is None
+
+ await hass.config_entries.async_unload(entry.entry_id)
+
+
+@pytest.mark.usefixtures("mock_device")
+@pytest.mark.usefixtures("mock_zeroconf")
+async def test_update_connected_wifi_clients(hass: HomeAssistant):
+ """Test state change of a connected_wifi_clients sensor device."""
+ state_key = f"{DOMAIN}.connected_wifi_clients"
+
+ entry = configure_integration(hass)
+ 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 == "1"
+
+ # Emulate device failure
+ with patch(
+ "devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_connected_station",
+ side_effect=DeviceUnavailable,
+ ):
+ async_fire_time_changed(hass, dt.utcnow() + SHORT_UPDATE_INTERVAL)
+ await hass.async_block_till_done()
+
+ state = hass.states.get(state_key)
+ assert state is not None
+ assert state.state == STATE_UNAVAILABLE
+
+ # Emulate state change
+ async_fire_time_changed(hass, dt.utcnow() + SHORT_UPDATE_INTERVAL)
+ await hass.async_block_till_done()
+
+ state = hass.states.get(state_key)
+ assert state is not None
+ assert state.state == "1"
+
+ await hass.config_entries.async_unload(entry.entry_id)
+
+
+@pytest.mark.usefixtures("mock_device")
+@pytest.mark.usefixtures("mock_zeroconf")
+async def test_update_neighboring_wifi_networks(hass: HomeAssistant):
+ """Test state change of a neighboring_wifi_networks sensor device."""
+ state_key = f"{DOMAIN}.neighboring_wifi_networks"
+ entry = configure_integration(hass)
+ with patch(
+ "homeassistant.helpers.entity.Entity.entity_registry_enabled_default",
+ return_value=True,
+ ):
+ 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 == "1"
+
+ # Emulate device failure
+ with patch(
+ "devolo_plc_api.device_api.deviceapi.DeviceApi.async_get_wifi_neighbor_access_points",
+ side_effect=DeviceUnavailable,
+ ):
+ async_fire_time_changed(hass, dt.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
+
+ # Emulate state change
+ async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
+ await hass.async_block_till_done()
+
+ state = hass.states.get(state_key)
+ assert state is not None
+ assert state.state == "1"
+
+ await hass.config_entries.async_unload(entry.entry_id)
+
+
+@pytest.mark.usefixtures("mock_device")
+@pytest.mark.usefixtures("mock_zeroconf")
+async def test_update_connected_plc_devices(hass: HomeAssistant):
+ """Test state change of a connected_plc_devices sensor device."""
+ state_key = f"{DOMAIN}.connected_plc_devices"
+ entry = configure_integration(hass)
+ with patch(
+ "homeassistant.helpers.entity.Entity.entity_registry_enabled_default",
+ return_value=True,
+ ):
+ 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 == "1"
+
+ # Emulate device failure
+ with patch(
+ "devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi.async_get_network_overview",
+ side_effect=DeviceUnavailable,
+ ):
+ async_fire_time_changed(hass, dt.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
+
+ # Emulate state change
+ async_fire_time_changed(hass, dt.utcnow() + LONG_UPDATE_INTERVAL)
+ await hass.async_block_till_done()
+
+ state = hass.states.get(state_key)
+ assert state is not None
+ assert state.state == "1"
+
+ await hass.config_entries.async_unload(entry.entry_id)
--
GitLab