From 2b30bda6c8371bbf546eeaa4b34393cbe58f9f86 Mon Sep 17 00:00:00 2001 From: Guido Schmitz Date: Mon, 9 May 2022 04:33:20 +0200 Subject: [PATCH] Add binary sensor platform to devolo Home Network (#60301) Co-authored-by: J. Nick Koston --- .../devolo_home_network/binary_sensor.py | 99 +++++++++++++++++++ .../components/devolo_home_network/const.py | 3 +- .../components/devolo_home_network/entity.py | 17 ++-- .../devolo_home_network/__init__.py | 1 + tests/components/devolo_home_network/const.py | 19 +++- .../devolo_home_network/test_binary_sensor.py | 83 ++++++++++++++++ 6 files changed, 210 insertions(+), 12 deletions(-) create mode 100644 homeassistant/components/devolo_home_network/binary_sensor.py create mode 100644 tests/components/devolo_home_network/test_binary_sensor.py diff --git a/homeassistant/components/devolo_home_network/binary_sensor.py b/homeassistant/components/devolo_home_network/binary_sensor.py new file mode 100644 index 00000000000..6bc02d802f5 --- /dev/null +++ b/homeassistant/components/devolo_home_network/binary_sensor.py @@ -0,0 +1,99 @@ +"""Platform for binary sensor integration.""" +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass + +from devolo_plc_api.device import Device + +from homeassistant.components.binary_sensor import ( + DEVICE_CLASS_PLUG, + BinarySensorEntity, + BinarySensorEntityDescription, +) +from homeassistant.config_entries import ConfigEntry +from homeassistant.core import HomeAssistant +from homeassistant.helpers.entity import EntityCategory +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 + + +def _is_connected_to_router(entity: DevoloBinarySensorEntity) -> bool: + """Check, if device is attached to the router.""" + return all( + device["attached_to_router"] + for device in entity.coordinator.data["network"]["devices"] + if device["mac_address"] == entity.device.mac + ) + + +@dataclass +class DevoloBinarySensorRequiredKeysMixin: + """Mixin for required keys.""" + + value_func: Callable[[DevoloBinarySensorEntity], bool] + + +@dataclass +class DevoloBinarySensorEntityDescription( + BinarySensorEntityDescription, DevoloBinarySensorRequiredKeysMixin +): + """Describes devolo sensor entity.""" + + +SENSOR_TYPES: dict[str, DevoloBinarySensorEntityDescription] = { + CONNECTED_TO_ROUTER: DevoloBinarySensorEntityDescription( + key=CONNECTED_TO_ROUTER, + device_class=DEVICE_CLASS_PLUG, + entity_category=EntityCategory.DIAGNOSTIC, + entity_registry_enabled_default=False, + icon="mdi:router-network", + name="Connected to router", + value_func=_is_connected_to_router, + ), +} + + +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] = hass.data[DOMAIN][entry.entry_id][ + "coordinators" + ] + + entities: list[BinarySensorEntity] = [] + if device.plcnet: + entities.append( + DevoloBinarySensorEntity( + coordinators[CONNECTED_PLC_DEVICES], + SENSOR_TYPES[CONNECTED_TO_ROUTER], + device, + entry.title, + ) + ) + async_add_entities(entities) + + +class DevoloBinarySensorEntity(DevoloEntity, BinarySensorEntity): + """Representation of a devolo binary sensor.""" + + def __init__( + self, + coordinator: DataUpdateCoordinator, + description: DevoloBinarySensorEntityDescription, + device: Device, + device_name: str, + ) -> None: + """Initialize entity.""" + self.entity_description: DevoloBinarySensorEntityDescription = description + super().__init__(coordinator, device, device_name) + + @property + def is_on(self) -> bool: + """State of the binary sensor.""" + return self.entity_description.value_func(self) diff --git a/homeassistant/components/devolo_home_network/const.py b/homeassistant/components/devolo_home_network/const.py index bd7170bfde5..2dfdd3c1d9a 100644 --- a/homeassistant/components/devolo_home_network/const.py +++ b/homeassistant/components/devolo_home_network/const.py @@ -5,7 +5,7 @@ from datetime import timedelta from homeassistant.const import Platform DOMAIN = "devolo_home_network" -PLATFORMS = [Platform.SENSOR] +PLATFORMS = [Platform.BINARY_SENSOR, Platform.SENSOR] PRODUCT = "product" SERIAL_NUMBER = "serial_number" @@ -15,5 +15,6 @@ LONG_UPDATE_INTERVAL = timedelta(minutes=5) SHORT_UPDATE_INTERVAL = timedelta(seconds=15) CONNECTED_PLC_DEVICES = "connected_plc_devices" +CONNECTED_TO_ROUTER = "connected_to_router" 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 index dbfe0e4035a..dd26324bc2c 100644 --- a/homeassistant/components/devolo_home_network/entity.py +++ b/homeassistant/components/devolo_home_network/entity.py @@ -21,17 +21,14 @@ class DevoloEntity(CoordinatorEntity): """Initialize a devolo home network device.""" super().__init__(coordinator) - self._device = device - self._device_name = device_name + self.device = device self._attr_device_info = DeviceInfo( - configuration_url=f"http://{self._device.ip}", - identifiers={(DOMAIN, str(self._device.serial_number))}, + configuration_url=f"http://{device.ip}", + identifiers={(DOMAIN, str(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}" + model=device.product, + name=device_name, + sw_version=device.firmware_version, ) + self._attr_unique_id = f"{device.serial_number}_{self.entity_description.key}" diff --git a/tests/components/devolo_home_network/__init__.py b/tests/components/devolo_home_network/__init__.py index 913193be3f7..1c10d7a59ef 100644 --- a/tests/components/devolo_home_network/__init__.py +++ b/tests/components/devolo_home_network/__init__.py @@ -28,5 +28,6 @@ def configure_integration(hass: HomeAssistant) -> MockConfigEntry: async def async_connect(self, session_instance: Any = None): """Give a mocked device the needed properties.""" + self.mac = DISCOVERY_INFO.properties["PlcMacAddress"] self.plcnet = PlcNetApi(IP, None, dataclasses.asdict(DISCOVERY_INFO)) self.device = DeviceApi(IP, None, dataclasses.asdict(DISCOVERY_INFO)) diff --git a/tests/components/devolo_home_network/const.py b/tests/components/devolo_home_network/const.py index 0e48833a78b..516a19f3421 100644 --- a/tests/components/devolo_home_network/const.py +++ b/tests/components/devolo_home_network/const.py @@ -62,6 +62,12 @@ NEIGHBOR_ACCESS_POINTS = { PLCNET = { "network": { + "devices": [ + { + "mac_address": "AA:BB:CC:DD:EE:FF", + "attached_to_router": False, + } + ], "data_rates": [ { "mac_address_from": "AA:BB:CC:DD:EE:FF", @@ -70,6 +76,17 @@ PLCNET = { "tx_rate": 0.0, }, ], - "devices": [], + } +} + +PLCNET_ATTACHED = { + "network": { + "devices": [ + { + "mac_address": "AA:BB:CC:DD:EE:FF", + "attached_to_router": True, + } + ], + "data_rates": [], } } diff --git a/tests/components/devolo_home_network/test_binary_sensor.py b/tests/components/devolo_home_network/test_binary_sensor.py new file mode 100644 index 00000000000..3d181da6106 --- /dev/null +++ b/tests/components/devolo_home_network/test_binary_sensor.py @@ -0,0 +1,83 @@ +"""Tests for the devolo Home Network sensors.""" +from unittest.mock import AsyncMock, patch + +from devolo_plc_api.exceptions.device import DeviceUnavailable +import pytest + +from homeassistant.components.binary_sensor import DOMAIN +from homeassistant.components.devolo_home_network.const import ( + CONNECTED_TO_ROUTER, + LONG_UPDATE_INTERVAL, +) +from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE +from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry +from homeassistant.helpers.entity import EntityCategory +from homeassistant.util import dt + +from . import configure_integration +from .const import PLCNET_ATTACHED + +from tests.common import async_fire_time_changed + + +@pytest.mark.usefixtures("mock_device", "mock_zeroconf") +async def test_binary_sensor_setup(hass: HomeAssistant): + """Test default setup of the binary 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_TO_ROUTER}") is None + + await hass.config_entries.async_unload(entry.entry_id) + + +@pytest.mark.usefixtures("mock_device", "mock_zeroconf") +async def test_update_attached_to_router(hass: HomeAssistant): + """Test state change of a attached_to_router binary sensor device.""" + state_key = f"{DOMAIN}.{CONNECTED_TO_ROUTER}" + entry = configure_integration(hass) + + er = entity_registry.async_get(hass) + + await hass.config_entries.async_setup(entry.entry_id) + await hass.async_block_till_done() + + # Enable entity + er.async_update_entity(state_key, disabled_by=None) + await hass.async_block_till_done() + 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_OFF + + assert er.async_get(state_key).entity_category == EntityCategory.DIAGNOSTIC + + # 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 + with patch( + "devolo_plc_api.plcnet_api.plcnetapi.PlcNetApi.async_get_network_overview", + new=AsyncMock(return_value=PLCNET_ATTACHED), + ): + 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_ON + + await hass.config_entries.async_unload(entry.entry_id)