Add last restart sensor to devolo_home_network (#122190)

* Add last restart sensor to devolo_home_network

* Add missing test

* Rename fetch function

* Fix mypy
This commit is contained in:
Guido Schmitz 2024-09-08 13:50:36 +02:00 committed by GitHub
parent 5b434aae6e
commit d4f0aaa089
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 207 additions and 39 deletions

View File

@ -39,6 +39,7 @@ from .const import (
CONNECTED_WIFI_CLIENTS, CONNECTED_WIFI_CLIENTS,
DOMAIN, DOMAIN,
FIRMWARE_UPDATE_INTERVAL, FIRMWARE_UPDATE_INTERVAL,
LAST_RESTART,
LONG_UPDATE_INTERVAL, LONG_UPDATE_INTERVAL,
NEIGHBORING_WIFI_NETWORKS, NEIGHBORING_WIFI_NETWORKS,
REGULAR_FIRMWARE, REGULAR_FIRMWARE,
@ -127,6 +128,19 @@ async def async_setup_entry(
except DeviceUnavailable as err: except DeviceUnavailable as err:
raise UpdateFailed(err) from err raise UpdateFailed(err) from err
async def async_update_last_restart() -> int:
"""Fetch data from API endpoint."""
assert device.device
update_sw_version(device_registry, device)
try:
return await device.device.async_uptime()
except DeviceUnavailable as err:
raise UpdateFailed(err) from err
except DevicePasswordProtected as err:
raise ConfigEntryAuthFailed(
err, translation_domain=DOMAIN, translation_key="password_wrong"
) from err
async def async_update_wifi_connected_station() -> list[ConnectedStationInfo]: async def async_update_wifi_connected_station() -> list[ConnectedStationInfo]:
"""Fetch data from API endpoint.""" """Fetch data from API endpoint."""
assert device.device assert device.device
@ -166,6 +180,14 @@ async def async_setup_entry(
update_method=async_update_led_status, update_method=async_update_led_status,
update_interval=SHORT_UPDATE_INTERVAL, update_interval=SHORT_UPDATE_INTERVAL,
) )
if device.device and "restart" in device.device.features:
coordinators[LAST_RESTART] = DataUpdateCoordinator(
hass,
_LOGGER,
name=LAST_RESTART,
update_method=async_update_last_restart,
update_interval=SHORT_UPDATE_INTERVAL,
)
if device.device and "update" in device.device.features: if device.device and "update" in device.device.features:
coordinators[REGULAR_FIRMWARE] = DataUpdateCoordinator( coordinators[REGULAR_FIRMWARE] = DataUpdateCoordinator(
hass, hass,

View File

@ -23,6 +23,7 @@ CONNECTED_TO_ROUTER = "connected_to_router"
CONNECTED_WIFI_CLIENTS = "connected_wifi_clients" CONNECTED_WIFI_CLIENTS = "connected_wifi_clients"
IDENTIFY = "identify" IDENTIFY = "identify"
IMAGE_GUEST_WIFI = "image_guest_wifi" IMAGE_GUEST_WIFI = "image_guest_wifi"
LAST_RESTART = "last_restart"
NEIGHBORING_WIFI_NETWORKS = "neighboring_wifi_networks" NEIGHBORING_WIFI_NETWORKS = "neighboring_wifi_networks"
PAIRING = "pairing" PAIRING = "pairing"
PLC_RX_RATE = "plc_rx_rate" PLC_RX_RATE = "plc_rx_rate"

View File

@ -26,6 +26,7 @@ type _DataType = (
| list[NeighborAPInfo] | list[NeighborAPInfo]
| WifiGuestAccessGet | WifiGuestAccessGet
| bool | bool
| int
) )

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import StrEnum from enum import StrEnum
from typing import Any, Generic, TypeVar from typing import Any, Generic, TypeVar
@ -20,11 +21,13 @@ from homeassistant.const import EntityCategory, UnitOfDataRate
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from homeassistant.util.dt import utcnow
from . import DevoloHomeNetworkConfigEntry from . import DevoloHomeNetworkConfigEntry
from .const import ( from .const import (
CONNECTED_PLC_DEVICES, CONNECTED_PLC_DEVICES,
CONNECTED_WIFI_CLIENTS, CONNECTED_WIFI_CLIENTS,
LAST_RESTART,
NEIGHBORING_WIFI_NETWORKS, NEIGHBORING_WIFI_NETWORKS,
PLC_RX_RATE, PLC_RX_RATE,
PLC_TX_RATE, PLC_TX_RATE,
@ -33,13 +36,36 @@ from .entity import DevoloCoordinatorEntity
PARALLEL_UPDATES = 1 PARALLEL_UPDATES = 1
def _last_restart(runtime: int) -> datetime:
"""Calculate uptime. As fetching the data might also take some time, let's floor to the nearest 5 seconds."""
now = utcnow()
return (
now
- timedelta(seconds=runtime)
- timedelta(seconds=(now.timestamp() - runtime) % 5)
)
_CoordinatorDataT = TypeVar( _CoordinatorDataT = TypeVar(
"_CoordinatorDataT", "_CoordinatorDataT",
bound=LogicalNetwork | DataRate | list[ConnectedStationInfo] | list[NeighborAPInfo], bound=LogicalNetwork
| DataRate
| list[ConnectedStationInfo]
| list[NeighborAPInfo]
| int,
) )
_ValueDataT = TypeVar( _ValueDataT = TypeVar(
"_ValueDataT", "_ValueDataT",
bound=LogicalNetwork | DataRate | list[ConnectedStationInfo] | list[NeighborAPInfo], bound=LogicalNetwork
| DataRate
| list[ConnectedStationInfo]
| list[NeighborAPInfo]
| int,
)
_SensorDataT = TypeVar(
"_SensorDataT",
bound=int | float | datetime,
) )
@ -52,15 +78,15 @@ class DataRateDirection(StrEnum):
@dataclass(frozen=True, kw_only=True) @dataclass(frozen=True, kw_only=True)
class DevoloSensorEntityDescription( class DevoloSensorEntityDescription(
SensorEntityDescription, Generic[_CoordinatorDataT] SensorEntityDescription, Generic[_CoordinatorDataT, _SensorDataT]
): ):
"""Describes devolo sensor entity.""" """Describes devolo sensor entity."""
value_func: Callable[[_CoordinatorDataT], float] value_func: Callable[[_CoordinatorDataT], _SensorDataT]
SENSOR_TYPES: dict[str, DevoloSensorEntityDescription[Any]] = { SENSOR_TYPES: dict[str, DevoloSensorEntityDescription[Any, Any]] = {
CONNECTED_PLC_DEVICES: DevoloSensorEntityDescription[LogicalNetwork]( CONNECTED_PLC_DEVICES: DevoloSensorEntityDescription[LogicalNetwork, int](
key=CONNECTED_PLC_DEVICES, key=CONNECTED_PLC_DEVICES,
entity_category=EntityCategory.DIAGNOSTIC, entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False, entity_registry_enabled_default=False,
@ -68,18 +94,20 @@ SENSOR_TYPES: dict[str, DevoloSensorEntityDescription[Any]] = {
{device.mac_address_from for device in data.data_rates} {device.mac_address_from for device in data.data_rates}
), ),
), ),
CONNECTED_WIFI_CLIENTS: DevoloSensorEntityDescription[list[ConnectedStationInfo]]( CONNECTED_WIFI_CLIENTS: DevoloSensorEntityDescription[
list[ConnectedStationInfo], int
](
key=CONNECTED_WIFI_CLIENTS, key=CONNECTED_WIFI_CLIENTS,
state_class=SensorStateClass.MEASUREMENT, state_class=SensorStateClass.MEASUREMENT,
value_func=len, value_func=len,
), ),
NEIGHBORING_WIFI_NETWORKS: DevoloSensorEntityDescription[list[NeighborAPInfo]]( NEIGHBORING_WIFI_NETWORKS: DevoloSensorEntityDescription[list[NeighborAPInfo], int](
key=NEIGHBORING_WIFI_NETWORKS, key=NEIGHBORING_WIFI_NETWORKS,
entity_category=EntityCategory.DIAGNOSTIC, entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False, entity_registry_enabled_default=False,
value_func=len, value_func=len,
), ),
PLC_RX_RATE: DevoloSensorEntityDescription[DataRate]( PLC_RX_RATE: DevoloSensorEntityDescription[DataRate, float](
key=PLC_RX_RATE, key=PLC_RX_RATE,
entity_category=EntityCategory.DIAGNOSTIC, entity_category=EntityCategory.DIAGNOSTIC,
name="PLC downlink PHY rate", name="PLC downlink PHY rate",
@ -88,7 +116,7 @@ SENSOR_TYPES: dict[str, DevoloSensorEntityDescription[Any]] = {
value_func=lambda data: getattr(data, DataRateDirection.RX, 0), value_func=lambda data: getattr(data, DataRateDirection.RX, 0),
suggested_display_precision=0, suggested_display_precision=0,
), ),
PLC_TX_RATE: DevoloSensorEntityDescription[DataRate]( PLC_TX_RATE: DevoloSensorEntityDescription[DataRate, float](
key=PLC_TX_RATE, key=PLC_TX_RATE,
entity_category=EntityCategory.DIAGNOSTIC, entity_category=EntityCategory.DIAGNOSTIC,
name="PLC uplink PHY rate", name="PLC uplink PHY rate",
@ -97,6 +125,13 @@ SENSOR_TYPES: dict[str, DevoloSensorEntityDescription[Any]] = {
value_func=lambda data: getattr(data, DataRateDirection.TX, 0), value_func=lambda data: getattr(data, DataRateDirection.TX, 0),
suggested_display_precision=0, suggested_display_precision=0,
), ),
LAST_RESTART: DevoloSensorEntityDescription[int, datetime](
key=LAST_RESTART,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
device_class=SensorDeviceClass.TIMESTAMP,
value_func=_last_restart,
),
} }
@ -109,7 +144,7 @@ async def async_setup_entry(
device = entry.runtime_data.device device = entry.runtime_data.device
coordinators = entry.runtime_data.coordinators coordinators = entry.runtime_data.coordinators
entities: list[BaseDevoloSensorEntity[Any, Any]] = [] entities: list[BaseDevoloSensorEntity[Any, Any, Any]] = []
if device.plcnet: if device.plcnet:
entities.append( entities.append(
DevoloSensorEntity( DevoloSensorEntity(
@ -139,6 +174,14 @@ async def async_setup_entry(
peer, peer,
) )
) )
if device.device and "restart" in device.device.features:
entities.append(
DevoloSensorEntity(
entry,
coordinators[LAST_RESTART],
SENSOR_TYPES[LAST_RESTART],
)
)
if device.device and "wifi1" in device.device.features: if device.device and "wifi1" in device.device.features:
entities.append( entities.append(
DevoloSensorEntity( DevoloSensorEntity(
@ -158,7 +201,7 @@ async def async_setup_entry(
class BaseDevoloSensorEntity( class BaseDevoloSensorEntity(
Generic[_CoordinatorDataT, _ValueDataT], Generic[_CoordinatorDataT, _ValueDataT, _SensorDataT],
DevoloCoordinatorEntity[_CoordinatorDataT], DevoloCoordinatorEntity[_CoordinatorDataT],
SensorEntity, SensorEntity,
): ):
@ -168,34 +211,38 @@ class BaseDevoloSensorEntity(
self, self,
entry: DevoloHomeNetworkConfigEntry, entry: DevoloHomeNetworkConfigEntry,
coordinator: DataUpdateCoordinator[_CoordinatorDataT], coordinator: DataUpdateCoordinator[_CoordinatorDataT],
description: DevoloSensorEntityDescription[_ValueDataT], description: DevoloSensorEntityDescription[_ValueDataT, _SensorDataT],
) -> None: ) -> None:
"""Initialize entity.""" """Initialize entity."""
self.entity_description = description self.entity_description = description
super().__init__(entry, coordinator) super().__init__(entry, coordinator)
class DevoloSensorEntity(BaseDevoloSensorEntity[_CoordinatorDataT, _CoordinatorDataT]): class DevoloSensorEntity(
BaseDevoloSensorEntity[_CoordinatorDataT, _CoordinatorDataT, _SensorDataT]
):
"""Representation of a generic devolo sensor.""" """Representation of a generic devolo sensor."""
entity_description: DevoloSensorEntityDescription[_CoordinatorDataT] entity_description: DevoloSensorEntityDescription[_CoordinatorDataT, _SensorDataT]
@property @property
def native_value(self) -> float: def native_value(self) -> int | float | datetime:
"""State of the sensor.""" """State of the sensor."""
return self.entity_description.value_func(self.coordinator.data) return self.entity_description.value_func(self.coordinator.data)
class DevoloPlcDataRateSensorEntity(BaseDevoloSensorEntity[LogicalNetwork, DataRate]): class DevoloPlcDataRateSensorEntity(
BaseDevoloSensorEntity[LogicalNetwork, DataRate, float]
):
"""Representation of a devolo PLC data rate sensor.""" """Representation of a devolo PLC data rate sensor."""
entity_description: DevoloSensorEntityDescription[DataRate] entity_description: DevoloSensorEntityDescription[DataRate, float]
def __init__( def __init__(
self, self,
entry: DevoloHomeNetworkConfigEntry, entry: DevoloHomeNetworkConfigEntry,
coordinator: DataUpdateCoordinator[LogicalNetwork], coordinator: DataUpdateCoordinator[LogicalNetwork],
description: DevoloSensorEntityDescription[DataRate], description: DevoloSensorEntityDescription[DataRate, float],
peer: str, peer: str,
) -> None: ) -> None:
"""Initialize entity.""" """Initialize entity."""

View File

@ -60,6 +60,9 @@
"connected_wifi_clients": { "connected_wifi_clients": {
"name": "Connected Wi-Fi clients" "name": "Connected Wi-Fi clients"
}, },
"last_restart": {
"name": "Last restart of the device"
},
"neighboring_wifi_networks": { "neighboring_wifi_networks": {
"name": "Neighboring Wi-Fi networks" "name": "Neighboring Wi-Fi networks"
}, },

View File

@ -171,3 +171,5 @@ PLCNET_ATTACHED = LogicalNetwork(
}, },
], ],
) )
UPTIME = 100

View File

@ -19,6 +19,7 @@ from .const import (
IP, IP,
NEIGHBOR_ACCESS_POINTS, NEIGHBOR_ACCESS_POINTS,
PLCNET, PLCNET,
UPTIME,
) )
@ -64,6 +65,7 @@ class MockDevice(Device):
) )
self.device.async_get_led_setting = AsyncMock(return_value=False) self.device.async_get_led_setting = AsyncMock(return_value=False)
self.device.async_restart = AsyncMock(return_value=True) self.device.async_restart = AsyncMock(return_value=True)
self.device.async_uptime = AsyncMock(return_value=UPTIME)
self.device.async_start_wps = AsyncMock(return_value=True) self.device.async_start_wps = AsyncMock(return_value=True)
self.device.async_get_wifi_connected_station = AsyncMock( self.device.async_get_wifi_connected_station = AsyncMock(
return_value=CONNECTED_STATIONS return_value=CONNECTED_STATIONS

View File

@ -1,5 +1,5 @@
# serializer version: 1 # serializer version: 1
# name: test_sensor[connected_plc_devices-async_get_network_overview-interval2] # name: test_sensor[connected_plc_devices-async_get_network_overview-interval2-1]
StateSnapshot({ StateSnapshot({
'attributes': ReadOnlyDict({ 'attributes': ReadOnlyDict({
'friendly_name': 'Mock Title Connected PLC devices', 'friendly_name': 'Mock Title Connected PLC devices',
@ -12,7 +12,7 @@
'state': '1', 'state': '1',
}) })
# --- # ---
# name: test_sensor[connected_plc_devices-async_get_network_overview-interval2].1 # name: test_sensor[connected_plc_devices-async_get_network_overview-interval2-1].1
EntityRegistryEntrySnapshot({ EntityRegistryEntrySnapshot({
'aliases': set({ 'aliases': set({
}), }),
@ -45,7 +45,7 @@
'unit_of_measurement': None, 'unit_of_measurement': None,
}) })
# --- # ---
# name: test_sensor[connected_wi_fi_clients-async_get_wifi_connected_station-interval0] # name: test_sensor[connected_wi_fi_clients-async_get_wifi_connected_station-interval0-1]
StateSnapshot({ StateSnapshot({
'attributes': ReadOnlyDict({ 'attributes': ReadOnlyDict({
'friendly_name': 'Mock Title Connected Wi-Fi clients', 'friendly_name': 'Mock Title Connected Wi-Fi clients',
@ -59,7 +59,7 @@
'state': '1', 'state': '1',
}) })
# --- # ---
# name: test_sensor[connected_wi_fi_clients-async_get_wifi_connected_station-interval0].1 # name: test_sensor[connected_wi_fi_clients-async_get_wifi_connected_station-interval0-1].1
EntityRegistryEntrySnapshot({ EntityRegistryEntrySnapshot({
'aliases': set({ 'aliases': set({
}), }),
@ -94,7 +94,54 @@
'unit_of_measurement': None, 'unit_of_measurement': None,
}) })
# --- # ---
# name: test_sensor[neighboring_wi_fi_networks-async_get_wifi_neighbor_access_points-interval1] # name: test_sensor[last_restart_of_the_device-async_uptime-interval3-2023-01-13T11:58:50+00:00]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'timestamp',
'friendly_name': 'Mock Title Last restart of the device',
}),
'context': <ANY>,
'entity_id': 'sensor.mock_title_last_restart_of_the_device',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '2023-01-13T11:58:20+00:00',
})
# ---
# name: test_sensor[last_restart_of_the_device-async_uptime-interval3-2023-01-13T11:58:50+00:00].1
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'sensor',
'entity_category': <EntityCategory.DIAGNOSTIC: 'diagnostic'>,
'entity_id': 'sensor.mock_title_last_restart_of_the_device',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <SensorDeviceClass.TIMESTAMP: 'timestamp'>,
'original_icon': None,
'original_name': 'Last restart of the device',
'platform': 'devolo_home_network',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'last_restart',
'unique_id': '1234567890_last_restart',
'unit_of_measurement': None,
})
# ---
# name: test_sensor[neighboring_wi_fi_networks-async_get_wifi_neighbor_access_points-interval1-1]
StateSnapshot({ StateSnapshot({
'attributes': ReadOnlyDict({ 'attributes': ReadOnlyDict({
'friendly_name': 'Mock Title Neighboring Wi-Fi networks', 'friendly_name': 'Mock Title Neighboring Wi-Fi networks',
@ -107,7 +154,7 @@
'state': '1', 'state': '1',
}) })
# --- # ---
# name: test_sensor[neighboring_wi_fi_networks-async_get_wifi_neighbor_access_points-interval1].1 # name: test_sensor[neighboring_wi_fi_networks-async_get_wifi_neighbor_access_points-interval1-1].1
EntityRegistryEntrySnapshot({ EntityRegistryEntrySnapshot({
'aliases': set({ 'aliases': set({
}), }),

View File

@ -3,16 +3,18 @@
from datetime import timedelta from datetime import timedelta
from unittest.mock import AsyncMock from unittest.mock import AsyncMock
from devolo_plc_api.exceptions.device import DeviceUnavailable from devolo_plc_api.exceptions.device import DevicePasswordProtected, DeviceUnavailable
from freezegun.api import FrozenDateTimeFactory from freezegun.api import FrozenDateTimeFactory
import pytest import pytest
from syrupy.assertion import SnapshotAssertion from syrupy.assertion import SnapshotAssertion
from homeassistant.components.devolo_home_network.const import ( from homeassistant.components.devolo_home_network.const import (
DOMAIN,
LONG_UPDATE_INTERVAL, LONG_UPDATE_INTERVAL,
SHORT_UPDATE_INTERVAL, SHORT_UPDATE_INTERVAL,
) )
from homeassistant.components.sensor import DOMAIN from homeassistant.components.sensor import DOMAIN as PLATFORM
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import STATE_UNAVAILABLE from homeassistant.const import STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er from homeassistant.helpers import entity_registry as er
@ -33,59 +35,74 @@ async def test_sensor_setup(hass: HomeAssistant) -> None:
await hass.async_block_till_done() await hass.async_block_till_done()
assert ( assert (
hass.states.get(f"{DOMAIN}.{device_name}_connected_wi_fi_clients") is not None hass.states.get(f"{PLATFORM}.{device_name}_connected_wi_fi_clients") is not None
)
assert hass.states.get(f"{PLATFORM}.{device_name}_connected_plc_devices") is None
assert (
hass.states.get(f"{PLATFORM}.{device_name}_neighboring_wi_fi_networks") is None
) )
assert hass.states.get(f"{DOMAIN}.{device_name}_connected_plc_devices") is None
assert hass.states.get(f"{DOMAIN}.{device_name}_neighboring_wi_fi_networks") is None
assert ( assert (
hass.states.get( hass.states.get(
f"{DOMAIN}.{device_name}_plc_downlink_phy_rate_{PLCNET.devices[1].user_device_name}" f"{PLATFORM}.{device_name}_plc_downlink_phy_rate_{PLCNET.devices[1].user_device_name}"
) )
is not None is not None
) )
assert ( assert (
hass.states.get( hass.states.get(
f"{DOMAIN}.{device_name}_plc_uplink_phy_rate_{PLCNET.devices[1].user_device_name}" f"{PLATFORM}.{device_name}_plc_uplink_phy_rate_{PLCNET.devices[1].user_device_name}"
) )
is not None is not None
) )
assert ( assert (
hass.states.get( hass.states.get(
f"{DOMAIN}.{device_name}_plc_downlink_phyrate_{PLCNET.devices[2].user_device_name}" f"{PLATFORM}.{device_name}_plc_downlink_phyrate_{PLCNET.devices[2].user_device_name}"
) )
is None is None
) )
assert ( assert (
hass.states.get( hass.states.get(
f"{DOMAIN}.{device_name}_plc_uplink_phyrate_{PLCNET.devices[2].user_device_name}" f"{PLATFORM}.{device_name}_plc_uplink_phyrate_{PLCNET.devices[2].user_device_name}"
) )
is None is None
) )
assert (
hass.states.get(f"{PLATFORM}.{device_name}_last_restart_of_the_device") is None
)
await hass.config_entries.async_unload(entry.entry_id) await hass.config_entries.async_unload(entry.entry_id)
@pytest.mark.parametrize( @pytest.mark.parametrize(
("name", "get_method", "interval"), ("name", "get_method", "interval", "expected_state"),
[ [
( (
"connected_wi_fi_clients", "connected_wi_fi_clients",
"async_get_wifi_connected_station", "async_get_wifi_connected_station",
SHORT_UPDATE_INTERVAL, SHORT_UPDATE_INTERVAL,
"1",
), ),
( (
"neighboring_wi_fi_networks", "neighboring_wi_fi_networks",
"async_get_wifi_neighbor_access_points", "async_get_wifi_neighbor_access_points",
LONG_UPDATE_INTERVAL, LONG_UPDATE_INTERVAL,
"1",
), ),
( (
"connected_plc_devices", "connected_plc_devices",
"async_get_network_overview", "async_get_network_overview",
LONG_UPDATE_INTERVAL, LONG_UPDATE_INTERVAL,
"1",
),
(
"last_restart_of_the_device",
"async_uptime",
SHORT_UPDATE_INTERVAL,
"2023-01-13T11:58:50+00:00",
), ),
], ],
) )
@pytest.mark.usefixtures("entity_registry_enabled_by_default") @pytest.mark.usefixtures("entity_registry_enabled_by_default")
@pytest.mark.freeze_time("2023-01-13 12:00:00+00:00")
async def test_sensor( async def test_sensor(
hass: HomeAssistant, hass: HomeAssistant,
mock_device: MockDevice, mock_device: MockDevice,
@ -95,11 +112,12 @@ async def test_sensor(
name: str, name: str,
get_method: str, get_method: str,
interval: timedelta, interval: timedelta,
expected_state: str,
) -> None: ) -> None:
"""Test state change of a sensor device.""" """Test state change of a sensor device."""
entry = configure_integration(hass) entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower() device_name = entry.title.replace(" ", "_").lower()
state_key = f"{DOMAIN}.{device_name}_{name}" state_key = f"{PLATFORM}.{device_name}_{name}"
await hass.config_entries.async_setup(entry.entry_id) await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -125,7 +143,7 @@ async def test_sensor(
state = hass.states.get(state_key) state = hass.states.get(state_key)
assert state is not None assert state is not None
assert state.state == "1" assert state.state == expected_state
await hass.config_entries.async_unload(entry.entry_id) await hass.config_entries.async_unload(entry.entry_id)
@ -140,8 +158,8 @@ async def test_update_plc_phyrates(
"""Test state change of plc_downlink_phyrate and plc_uplink_phyrate sensor devices.""" """Test state change of plc_downlink_phyrate and plc_uplink_phyrate sensor devices."""
entry = configure_integration(hass) entry = configure_integration(hass)
device_name = entry.title.replace(" ", "_").lower() device_name = entry.title.replace(" ", "_").lower()
state_key_downlink = f"{DOMAIN}.{device_name}_plc_downlink_phy_rate_{PLCNET.devices[1].user_device_name}" state_key_downlink = f"{PLATFORM}.{device_name}_plc_downlink_phy_rate_{PLCNET.devices[1].user_device_name}"
state_key_uplink = f"{DOMAIN}.{device_name}_plc_uplink_phy_rate_{PLCNET.devices[1].user_device_name}" state_key_uplink = f"{PLATFORM}.{device_name}_plc_uplink_phy_rate_{PLCNET.devices[1].user_device_name}"
await hass.config_entries.async_setup(entry.entry_id) await hass.config_entries.async_setup(entry.entry_id)
await hass.async_block_till_done() await hass.async_block_till_done()
@ -181,3 +199,28 @@ async def test_update_plc_phyrates(
assert state.state == str(PLCNET.data_rates[0].tx_rate) assert state.state == str(PLCNET.data_rates[0].tx_rate)
await hass.config_entries.async_unload(entry.entry_id) await hass.config_entries.async_unload(entry.entry_id)
async def test_update_last_update_auth_failed(
hass: HomeAssistant, mock_device: MockDevice
) -> None:
"""Test getting the last update state with wrong password triggers the reauth flow."""
entry = configure_integration(hass)
mock_device.device.async_uptime.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)