Add switch platform to devolo_home_network (#72494)

This commit is contained in:
Guido Schmitz 2023-01-10 10:05:59 +01:00 committed by GitHub
parent 05c32c51fd
commit d40a4aa970
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 577 additions and 18 deletions

View File

@ -6,15 +6,23 @@ from typing import Any
import async_timeout
from devolo_plc_api import Device
from devolo_plc_api.device_api import ConnectedStationInfo, NeighborAPInfo
from devolo_plc_api.exceptions.device import DeviceNotFound, DeviceUnavailable
from devolo_plc_api.device_api import (
ConnectedStationInfo,
NeighborAPInfo,
WifiGuestAccessGet,
)
from devolo_plc_api.exceptions.device import (
DeviceNotFound,
DevicePasswordProtected,
DeviceUnavailable,
)
from devolo_plc_api.plcnet_api import LogicalNetwork
from homeassistant.components import zeroconf
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_IP_ADDRESS, CONF_PASSWORD, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -26,6 +34,8 @@ from .const import (
NEIGHBORING_WIFI_NETWORKS,
PLATFORMS,
SHORT_UPDATE_INTERVAL,
SWITCH_GUEST_WIFI,
SWITCH_LEDS,
)
_LOGGER = logging.getLogger(__name__)
@ -59,6 +69,28 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
except DeviceUnavailable as err:
raise UpdateFailed(err) from err
async def async_update_guest_wifi_status() -> WifiGuestAccessGet:
"""Fetch data from API endpoint."""
assert device.device
try:
async with async_timeout.timeout(10):
return await device.device.async_get_wifi_guest_access()
except DeviceUnavailable as err:
raise UpdateFailed(err) from err
except DevicePasswordProtected as err:
raise ConfigEntryAuthFailed(err) from err
async def async_update_led_status() -> bool:
"""Fetch data from API endpoint."""
assert device.device
try:
async with async_timeout.timeout(10):
return await device.device.async_get_led_setting()
except DeviceUnavailable as err:
raise UpdateFailed(err) from err
except DevicePasswordProtected as err:
raise ConfigEntryAuthFailed(err) from err
async def async_update_wifi_connected_station() -> list[ConnectedStationInfo]:
"""Fetch data from API endpoint."""
assert device.device
@ -90,6 +122,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
update_method=async_update_connected_plc_devices,
update_interval=LONG_UPDATE_INTERVAL,
)
if device.device and "led" in device.device.features:
coordinators[SWITCH_LEDS] = DataUpdateCoordinator(
hass,
_LOGGER,
name=SWITCH_LEDS,
update_method=async_update_led_status,
update_interval=SHORT_UPDATE_INTERVAL,
)
if device.device and "wifi1" in device.device.features:
coordinators[CONNECTED_WIFI_CLIENTS] = DataUpdateCoordinator(
hass,
@ -105,6 +145,13 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
update_method=async_update_wifi_neighbor_access_points,
update_interval=LONG_UPDATE_INTERVAL,
)
coordinators[SWITCH_GUEST_WIFI] = DataUpdateCoordinator(
hass,
_LOGGER,
name=SWITCH_GUEST_WIFI,
update_method=async_update_guest_wifi_status,
update_interval=SHORT_UPDATE_INTERVAL,
)
hass.data[DOMAIN][entry.entry_id] = {"device": device, "coordinators": coordinators}

View File

@ -72,10 +72,10 @@ async def async_setup_entry(
if device.plcnet:
entities.append(
DevoloBinarySensorEntity(
entry,
coordinators[CONNECTED_PLC_DEVICES],
SENSOR_TYPES[CONNECTED_TO_ROUTER],
device,
entry.title,
)
)
async_add_entities(entities)
@ -86,14 +86,14 @@ class DevoloBinarySensorEntity(DevoloEntity[LogicalNetwork], BinarySensorEntity)
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator[LogicalNetwork],
description: DevoloBinarySensorEntityDescription,
device: Device,
device_name: str,
) -> None:
"""Initialize entity."""
self.entity_description: DevoloBinarySensorEntityDescription = description
super().__init__(coordinator, device, device_name)
super().__init__(entry, coordinator, device)
@property
def is_on(self) -> bool:

View File

@ -12,7 +12,12 @@ from devolo_plc_api.device_api import (
from homeassistant.const import Platform
DOMAIN = "devolo_home_network"
PLATFORMS = [Platform.BINARY_SENSOR, Platform.DEVICE_TRACKER, Platform.SENSOR]
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.DEVICE_TRACKER,
Platform.SENSOR,
Platform.SWITCH,
]
PRODUCT = "product"
SERIAL_NUMBER = "serial_number"
@ -25,6 +30,8 @@ CONNECTED_PLC_DEVICES = "connected_plc_devices"
CONNECTED_TO_ROUTER = "connected_to_router"
CONNECTED_WIFI_CLIENTS = "connected_wifi_clients"
NEIGHBORING_WIFI_NETWORKS = "neighboring_wifi_networks"
SWITCH_GUEST_WIFI = "switch_guest_wifi"
SWITCH_LEDS = "switch_leds"
WIFI_APTYPE = {
WIFI_VAP_MAIN_AP: "Main",

View File

@ -4,9 +4,14 @@ from __future__ import annotations
from typing import TypeVar, Union
from devolo_plc_api.device import Device
from devolo_plc_api.device_api import ConnectedStationInfo, NeighborAPInfo
from devolo_plc_api.device_api import (
ConnectedStationInfo,
NeighborAPInfo,
WifiGuestAccessGet,
)
from devolo_plc_api.plcnet_api import LogicalNetwork
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
@ -21,6 +26,8 @@ _DataT = TypeVar(
LogicalNetwork,
list[ConnectedStationInfo],
list[NeighborAPInfo],
WifiGuestAccessGet,
bool,
],
)
@ -32,21 +39,22 @@ class DevoloEntity(CoordinatorEntity[DataUpdateCoordinator[_DataT]]):
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator[_DataT],
device: Device,
device_name: str,
) -> None:
"""Initialize a devolo home network device."""
super().__init__(coordinator)
self.device = device
self.entry = entry
self._attr_device_info = DeviceInfo(
configuration_url=f"http://{device.ip}",
identifiers={(DOMAIN, str(device.serial_number))},
manufacturer="devolo",
model=device.product,
name=device_name,
name=entry.title,
sw_version=device.firmware_version,
)
self._attr_unique_id = f"{device.serial_number}_{self.entity_description.key}"

View File

@ -65,7 +65,6 @@ SENSOR_TYPES: dict[str, DevoloSensorEntityDescription[Any]] = {
),
CONNECTED_WIFI_CLIENTS: DevoloSensorEntityDescription[list[ConnectedStationInfo]](
key=CONNECTED_WIFI_CLIENTS,
entity_registry_enabled_default=True,
icon="mdi:wifi",
name="Connected Wifi clients",
state_class=SensorStateClass.MEASUREMENT,
@ -95,27 +94,27 @@ async def async_setup_entry(
if device.plcnet:
entities.append(
DevoloSensorEntity(
entry,
coordinators[CONNECTED_PLC_DEVICES],
SENSOR_TYPES[CONNECTED_PLC_DEVICES],
device,
entry.title,
)
)
if device.device and "wifi1" in device.device.features:
entities.append(
DevoloSensorEntity(
entry,
coordinators[CONNECTED_WIFI_CLIENTS],
SENSOR_TYPES[CONNECTED_WIFI_CLIENTS],
device,
entry.title,
)
)
entities.append(
DevoloSensorEntity(
entry,
coordinators[NEIGHBORING_WIFI_NETWORKS],
SENSOR_TYPES[NEIGHBORING_WIFI_NETWORKS],
device,
entry.title,
)
)
async_add_entities(entities)
@ -128,14 +127,14 @@ class DevoloSensorEntity(DevoloEntity[_DataT], SensorEntity):
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator[_DataT],
description: DevoloSensorEntityDescription[_DataT],
device: Device,
device_name: str,
) -> None:
"""Initialize entity."""
self.entity_description = description
super().__init__(coordinator, device, device_name)
super().__init__(entry, coordinator, device)
@property
def native_value(self) -> int:

View File

@ -0,0 +1,138 @@
"""Platform for switch integration."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any, Generic, TypeVar, Union
from devolo_plc_api.device import Device
from devolo_plc_api.device_api import WifiGuestAccessGet
from devolo_plc_api.exceptions.device import DevicePasswordProtected, DeviceUnavailable
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
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 DOMAIN, SWITCH_GUEST_WIFI, SWITCH_LEDS
from .entity import DevoloEntity
_DataT = TypeVar(
"_DataT",
bound=Union[
WifiGuestAccessGet,
bool,
],
)
@dataclass
class DevoloSwitchRequiredKeysMixin(Generic[_DataT]):
"""Mixin for required keys."""
is_on_func: Callable[[_DataT], bool]
turn_on_func: Callable[[Device], Awaitable[bool]]
turn_off_func: Callable[[Device], Awaitable[bool]]
@dataclass
class DevoloSwitchEntityDescription(
SwitchEntityDescription, DevoloSwitchRequiredKeysMixin[_DataT]
):
"""Describes devolo switch entity."""
SWITCH_TYPES: dict[str, DevoloSwitchEntityDescription[Any]] = {
SWITCH_GUEST_WIFI: DevoloSwitchEntityDescription[WifiGuestAccessGet](
key=SWITCH_GUEST_WIFI,
icon="mdi:wifi",
name="Enable guest Wifi",
is_on_func=lambda data: data.enabled is True,
turn_on_func=lambda device: device.device.async_set_wifi_guest_access(True), # type: ignore[union-attr]
turn_off_func=lambda device: device.device.async_set_wifi_guest_access(False), # type: ignore[union-attr]
),
SWITCH_LEDS: DevoloSwitchEntityDescription[bool](
key=SWITCH_LEDS,
entity_category=EntityCategory.CONFIG,
icon="mdi:led-off",
name="Enable LEDs",
is_on_func=bool,
turn_on_func=lambda device: device.device.async_set_led_setting(True), # type: ignore[union-attr]
turn_off_func=lambda device: device.device.async_set_led_setting(False), # 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"]
entities: list[DevoloSwitchEntity[Any]] = []
if device.device and "led" in device.device.features:
entities.append(
DevoloSwitchEntity(
entry,
coordinators[SWITCH_LEDS],
SWITCH_TYPES[SWITCH_LEDS],
device,
)
)
if device.device and "wifi1" in device.device.features:
entities.append(
DevoloSwitchEntity(
entry,
coordinators[SWITCH_GUEST_WIFI],
SWITCH_TYPES[SWITCH_GUEST_WIFI],
device,
)
)
async_add_entities(entities)
class DevoloSwitchEntity(DevoloEntity[_DataT], SwitchEntity):
"""Representation of a devolo switch."""
entity_description: DevoloSwitchEntityDescription[_DataT]
def __init__(
self,
entry: ConfigEntry,
coordinator: DataUpdateCoordinator[_DataT],
description: DevoloSwitchEntityDescription[_DataT],
device: Device,
) -> None:
"""Initialize entity."""
self.entity_description = description
super().__init__(entry, coordinator, device)
@property
def is_on(self) -> bool:
"""State of the switch."""
return self.entity_description.is_on_func(self.coordinator.data)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
try:
await self.entity_description.turn_on_func(self.device)
except DevicePasswordProtected:
self.entry.async_start_reauth(self.hass)
except DeviceUnavailable:
pass # The coordinator will handle this
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
try:
await self.entity_description.turn_off_func(self.device)
except DevicePasswordProtected:
self.entry.async_start_reauth(self.hass)
except DeviceUnavailable:
pass # The coordinator will handle this
await self.coordinator.async_request_refresh()

View File

@ -6,6 +6,7 @@ from devolo_plc_api.device_api import (
WIFI_VAP_MAIN_AP,
ConnectedStationInfo,
NeighborAPInfo,
WifiGuestAccessGet,
)
from devolo_plc_api.plcnet_api import LogicalNetwork
@ -56,6 +57,13 @@ DISCOVERY_INFO_WRONG_DEVICE = ZeroconfServiceInfo(
type="mock_type",
)
GUEST_WIFI = WifiGuestAccessGet(
ssid="devolo-guest-930",
key="HMANPGBA",
enabled=False,
remaining_duration=0,
)
NEIGHBOR_ACCESS_POINTS = [
NeighborAPInfo(
mac_address="AA:BB:CC:DD:EE:FF",
@ -67,7 +75,6 @@ NEIGHBOR_ACCESS_POINTS = [
)
]
PLCNET = LogicalNetwork(
devices=[
{
@ -85,7 +92,6 @@ PLCNET = LogicalNetwork(
],
)
PLCNET_ATTACHED = LogicalNetwork(
devices=[
{

View File

@ -13,6 +13,7 @@ from zeroconf.asyncio import AsyncZeroconf
from .const import (
CONNECTED_STATIONS,
DISCOVERY_INFO,
GUEST_WIFI,
IP,
NEIGHBOR_ACCESS_POINTS,
PLCNET,
@ -43,9 +44,11 @@ class MockDevice(Device):
"""Reset mock to starting point."""
self.async_disconnect = AsyncMock()
self.device = DeviceApi(IP, None, DISCOVERY_INFO)
self.device.async_get_led_setting = AsyncMock(return_value=False)
self.device.async_get_wifi_connected_station = AsyncMock(
return_value=CONNECTED_STATIONS
)
self.device.async_get_wifi_guest_access = AsyncMock(return_value=GUEST_WIFI)
self.device.async_get_wifi_neighbor_access_points = AsyncMock(
return_value=NEIGHBOR_ACCESS_POINTS
)

View File

@ -0,0 +1,351 @@
"""Tests for the devolo Home Network switch."""
from datetime import timedelta
from unittest.mock import AsyncMock, patch
from devolo_plc_api.device_api import WifiGuestAccessGet
from devolo_plc_api.exceptions.device import DevicePasswordProtected, DeviceUnavailable
import pytest
from homeassistant.components.devolo_home_network.const import (
DOMAIN,
SHORT_UPDATE_INTERVAL,
)
from homeassistant.components.switch import DOMAIN as PLATFORM
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import (
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
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.helpers.update_coordinator import REQUEST_REFRESH_DEFAULT_COOLDOWN
from homeassistant.util import dt
from . import configure_integration
from .mock import MockDevice
from tests.common import async_fire_time_changed
@pytest.mark.usefixtures("mock_device")
async def test_switch_setup(hass: HomeAssistant):
"""Test default setup of the switch 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}_enable_guest_wifi") is not None
assert hass.states.get(f"{PLATFORM}.{device_name}_enable_leds") is not None
await hass.config_entries.async_unload(entry.entry_id)
async def test_update_guest_wifi_status_auth_failed(
hass: HomeAssistant, mock_device: MockDevice
):
"""Test getting the wifi_status with wrong password triggers the reauth flow."""
entry = configure_integration(hass)
mock_device.device.async_get_wifi_guest_access.side_effect = DevicePasswordProtected
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_ERROR
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)
async def test_update_led_status_auth_failed(
hass: HomeAssistant, mock_device: MockDevice
):
"""Test getting the led status with wrong password triggers the reauth flow."""
entry = configure_integration(hass)
mock_device.device.async_get_led_setting.side_effect = DevicePasswordProtected
await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done()
assert entry.state is ConfigEntryState.SETUP_ERROR
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)
async def test_update_enable_guest_wifi(hass: HomeAssistant, mock_device: MockDevice):
"""Test state change of a enable_guest_wifi switch device."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
state_key = f"{PLATFORM}.{device_name}_enable_guest_wifi"
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_OFF
# Emulate state change
mock_device.device.async_get_wifi_guest_access.return_value = WifiGuestAccessGet(
enabled=True
)
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_ON
# Switch off
mock_device.device.async_get_wifi_guest_access.return_value = WifiGuestAccessGet(
enabled=False
)
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_set_wifi_guest_access",
new=AsyncMock(),
) as turn_off:
await hass.services.async_call(
PLATFORM, SERVICE_TURN_OFF, {"entity_id": state_key}, blocking=True
)
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_OFF
turn_off.assert_called_once_with(False)
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REQUEST_REFRESH_DEFAULT_COOLDOWN)
)
await hass.async_block_till_done()
# Switch on
mock_device.device.async_get_wifi_guest_access.return_value = WifiGuestAccessGet(
enabled=True
)
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_set_wifi_guest_access",
new=AsyncMock(),
) as turn_on:
await hass.services.async_call(
PLATFORM, SERVICE_TURN_ON, {"entity_id": state_key}, blocking=True
)
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_ON
turn_on.assert_called_once_with(True)
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REQUEST_REFRESH_DEFAULT_COOLDOWN)
)
await hass.async_block_till_done()
# Device unavailable
mock_device.device.async_get_wifi_guest_access.side_effect = DeviceUnavailable()
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_set_wifi_guest_access",
side_effect=DeviceUnavailable,
):
await hass.services.async_call(
PLATFORM, SERVICE_TURN_ON, {"entity_id": state_key}, blocking=True
)
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
await hass.config_entries.async_unload(entry.entry_id)
async def test_update_enable_leds(hass: HomeAssistant, mock_device: MockDevice):
"""Test state change of a enable_leds switch device."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
state_key = f"{PLATFORM}.{device_name}_enable_leds"
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_OFF
er = entity_registry.async_get(hass)
assert er.async_get(state_key).entity_category == EntityCategory.CONFIG
# Emulate state change
mock_device.device.async_get_led_setting.return_value = True
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_ON
# Switch off
mock_device.device.async_get_led_setting.return_value = False
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_set_led_setting",
new=AsyncMock(),
) as turn_off:
await hass.services.async_call(
PLATFORM, SERVICE_TURN_OFF, {"entity_id": state_key}, blocking=True
)
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_OFF
turn_off.assert_called_once_with(False)
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REQUEST_REFRESH_DEFAULT_COOLDOWN)
)
await hass.async_block_till_done()
# Switch on
mock_device.device.async_get_led_setting.return_value = True
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_set_led_setting",
new=AsyncMock(),
) as turn_on:
await hass.services.async_call(
PLATFORM, SERVICE_TURN_ON, {"entity_id": state_key}, blocking=True
)
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_ON
turn_on.assert_called_once_with(True)
async_fire_time_changed(
hass, dt.utcnow() + timedelta(seconds=REQUEST_REFRESH_DEFAULT_COOLDOWN)
)
await hass.async_block_till_done()
# Device unavailable
mock_device.device.async_get_led_setting.side_effect = DeviceUnavailable()
with patch(
"devolo_plc_api.device_api.deviceapi.DeviceApi.async_set_led_setting",
side_effect=DeviceUnavailable,
):
await hass.services.async_call(
PLATFORM, SERVICE_TURN_OFF, {"entity_id": state_key}, blocking=True
)
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
await hass.config_entries.async_unload(entry.entry_id)
@pytest.mark.parametrize(
"name, get_method, update_interval",
[
["enable_guest_wifi", "async_get_wifi_guest_access", SHORT_UPDATE_INTERVAL],
["enable_leds", "async_get_led_setting", SHORT_UPDATE_INTERVAL],
],
)
async def test_device_failure(
hass: HomeAssistant,
mock_device: MockDevice,
name: str,
get_method: str,
update_interval: timedelta,
):
"""Test device failure."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
state_key = f"{PLATFORM}.{device_name}_{name}"
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
api = getattr(mock_device.device, get_method)
api.side_effect = DeviceUnavailable
async_fire_time_changed(hass, dt.utcnow() + update_interval)
await hass.async_block_till_done()
state = hass.states.get(state_key)
assert state is not None
assert state.state == STATE_UNAVAILABLE
@pytest.mark.parametrize(
"name, set_method",
[
["enable_guest_wifi", "async_set_wifi_guest_access"],
["enable_leds", "async_set_led_setting"],
],
)
async def test_auth_failed(
hass: HomeAssistant, mock_device: MockDevice, name: str, set_method: str
):
"""Test setting unautherized triggers the reauth flow."""
entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower()
state_key = f"{PLATFORM}.{device_name}_{name}"
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
setattr(mock_device.device, set_method, AsyncMock())
api = getattr(mock_device.device, set_method)
api.side_effect = DevicePasswordProtected
await hass.services.async_call(
PLATFORM, SERVICE_TURN_ON, {"entity_id": state_key}, blocking=True
)
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.services.async_call(
PLATFORM, SERVICE_TURN_OFF, {"entity_id": state_key}, blocking=True
)
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)