mirror of https://github.com/home-assistant/core
Add devolo home network integration (#45866)
Co-authored-by: Markus Bong <2Fake1987@gmail.com> Co-authored-by: Markus Bong <Markus.Bong@devolo.de>
This commit is contained in:
Guido Schmitz
GitHub
parent
3705f2f7f1
commit
f1884d34e9
|
|
@ -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.*
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -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},
|
||||
)
|
||||
|
|
@ -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"
|
||||
|
|
@ -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}"
|
||||
)
|
||||
|
|
@ -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"
|
||||
}
|
||||
|
|
@ -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)
|
||||
|
|
@ -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."
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -60,6 +60,7 @@ FLOWS = [
|
|||
"deconz",
|
||||
"denonavr",
|
||||
"devolo_home_control",
|
||||
"devolo_home_network",
|
||||
"dexcom",
|
||||
"dialogflow",
|
||||
"directv",
|
||||
|
|
|
|||
|
|
@ -58,6 +58,9 @@ ZEROCONF = {
|
|||
"_dvl-deviceapi._tcp.local.": [
|
||||
{
|
||||
"domain": "devolo_home_control"
|
||||
},
|
||||
{
|
||||
"domain": "devolo_home_network"
|
||||
}
|
||||
],
|
||||
"_easylink._tcp.local.": [
|
||||
|
|
|
|||
11
mypy.ini
11
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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -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
|
||||
|
|
@ -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": [],
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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)
|
||||
Loading…
Reference in New Issue