From d5d906e1488281dc234411a66ccceeca829d37ba Mon Sep 17 00:00:00 2001 From: Robert Hillis Date: Tue, 18 Jun 2024 03:12:02 -0400 Subject: [PATCH] Add update coordinator to Netgear LTE (#115474) --- .../components/netgear_lte/__init__.py | 142 +++++------------- .../components/netgear_lte/binary_sensor.py | 13 +- homeassistant/components/netgear_lte/const.py | 2 +- .../components/netgear_lte/coordinator.py | 43 ++++++ .../components/netgear_lte/entity.py | 50 ++---- .../components/netgear_lte/notify.py | 24 +-- .../components/netgear_lte/sensor.py | 28 ++-- .../components/netgear_lte/services.py | 23 +-- .../netgear_lte/test_config_flow.py | 9 +- tests/components/netgear_lte/test_init.py | 26 ++++ 10 files changed, 166 insertions(+), 194 deletions(-) create mode 100644 homeassistant/components/netgear_lte/coordinator.py diff --git a/homeassistant/components/netgear_lte/__init__.py b/homeassistant/components/netgear_lte/__init__.py index c47a5088887..1846d1f7992 100644 --- a/homeassistant/components/netgear_lte/__init__.py +++ b/homeassistant/components/netgear_lte/__init__.py @@ -1,25 +1,17 @@ """Support for Netgear LTE modems.""" -from datetime import timedelta +from typing import Any from aiohttp.cookiejar import CookieJar -import attr import eternalegypt +from eternalegypt.eternalegypt import SMS from homeassistant.config_entries import ConfigEntry, ConfigEntryState -from homeassistant.const import ( - CONF_HOST, - CONF_NAME, - CONF_PASSWORD, - EVENT_HOMEASSISTANT_STOP, - Platform, -) -from homeassistant.core import Event, HomeAssistant +from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, Platform +from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.helpers import config_validation as cv, discovery from homeassistant.helpers.aiohttp_client import async_create_clientsession -from homeassistant.helpers.dispatcher import async_dispatcher_send -from homeassistant.helpers.event import async_track_time_interval from homeassistant.helpers.typing import ConfigType from .const import ( @@ -28,14 +20,12 @@ from .const import ( ATTR_MESSAGE, ATTR_SMS_ID, DATA_HASS_CONFIG, - DISPATCHER_NETGEAR_LTE, + DATA_SESSION, DOMAIN, - LOGGER, ) +from .coordinator import NetgearLTEDataUpdateCoordinator from .services import async_setup_services -SCAN_INTERVAL = timedelta(seconds=10) - EVENT_SMS = "netgear_lte_sms" ALL_SENSORS = [ @@ -65,54 +55,11 @@ PLATFORMS = [ Platform.NOTIFY, Platform.SENSOR, ] +type NetgearLTEConfigEntry = ConfigEntry[NetgearLTEDataUpdateCoordinator] CONFIG_SCHEMA = cv.empty_config_schema(DOMAIN) -@attr.s -class ModemData: - """Class for modem state.""" - - hass = attr.ib() - host = attr.ib() - modem = attr.ib() - - data = attr.ib(init=False, default=None) - connected = attr.ib(init=False, default=True) - - async def async_update(self): - """Call the API to update the data.""" - - try: - self.data = await self.modem.information() - if not self.connected: - LOGGER.warning("Connected to %s", self.host) - self.connected = True - except eternalegypt.Error: - if self.connected: - LOGGER.warning("Lost connection to %s", self.host) - self.connected = False - self.data = None - - async_dispatcher_send(self.hass, DISPATCHER_NETGEAR_LTE) - - -@attr.s -class LTEData: - """Shared state.""" - - websession = attr.ib() - modem_data: dict[str, ModemData] = attr.ib(init=False, factory=dict) - - def get_modem_data(self, config): - """Get modem_data for the host in config.""" - if config[CONF_HOST] is not None: - return self.modem_data.get(config[CONF_HOST]) - if len(self.modem_data) != 1: - return None - return next(iter(self.modem_data.values())) - - async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up Netgear LTE component.""" hass.data[DATA_HASS_CONFIG] = config @@ -120,44 +67,44 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: return True -async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: +async def async_setup_entry(hass: HomeAssistant, entry: NetgearLTEConfigEntry) -> bool: """Set up Netgear LTE from a config entry.""" host = entry.data[CONF_HOST] password = entry.data[CONF_PASSWORD] - if not (data := hass.data.get(DOMAIN)) or data.websession.closed: - websession = async_create_clientsession(hass, cookie_jar=CookieJar(unsafe=True)) + data: dict[str, Any] = hass.data.setdefault(DOMAIN, {}) + if not (session := data.get(DATA_SESSION)) or session.closed: + session = async_create_clientsession(hass, cookie_jar=CookieJar(unsafe=True)) + modem = eternalegypt.Modem(hostname=host, websession=session) - hass.data[DOMAIN] = LTEData(websession) + try: + await modem.login(password=password) + except eternalegypt.Error as ex: + raise ConfigEntryNotReady("Cannot connect/authenticate") from ex - modem = eternalegypt.Modem(hostname=host, websession=hass.data[DOMAIN].websession) - modem_data = ModemData(hass, host, modem) + def fire_sms_event(sms: SMS) -> None: + """Send an SMS event.""" + data = { + ATTR_HOST: modem.hostname, + ATTR_SMS_ID: sms.id, + ATTR_FROM: sms.sender, + ATTR_MESSAGE: sms.message, + } + hass.bus.async_fire(EVENT_SMS, data) - await _login(hass, modem_data, password) + await modem.add_sms_listener(fire_sms_event) - async def _update(now): - """Periodic update.""" - await modem_data.async_update() + coordinator = NetgearLTEDataUpdateCoordinator(hass, modem) + await coordinator.async_config_entry_first_refresh() + entry.runtime_data = coordinator - update_unsub = async_track_time_interval(hass, _update, SCAN_INTERVAL) - - async def cleanup(event: Event | None = None) -> None: - """Clean up resources.""" - update_unsub() - await modem.logout() - if DOMAIN in hass.data: - del hass.data[DOMAIN].modem_data[modem_data.host] - - entry.async_on_unload(cleanup) - entry.async_on_unload(hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, cleanup)) - - await async_setup_services(hass) + await async_setup_services(hass, modem) await discovery.async_load_platform( hass, Platform.NOTIFY, DOMAIN, - {CONF_HOST: entry.data[CONF_HOST], CONF_NAME: entry.title}, + {CONF_NAME: entry.title, "modem": modem}, hass.data[DATA_HASS_CONFIG], ) @@ -168,7 +115,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return True -async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: +async def async_unload_entry(hass: HomeAssistant, entry: NetgearLTEConfigEntry) -> bool: """Unload a config entry.""" unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) loaded_entries = [ @@ -178,28 +125,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: ] if len(loaded_entries) == 1: hass.data.pop(DOMAIN, None) + for service_name in hass.services.async_services()[DOMAIN]: + hass.services.async_remove(DOMAIN, service_name) return unload_ok - - -async def _login(hass: HomeAssistant, modem_data: ModemData, password: str) -> None: - """Log in and complete setup.""" - try: - await modem_data.modem.login(password=password) - except eternalegypt.Error as ex: - raise ConfigEntryNotReady("Cannot connect/authenticate") from ex - - def fire_sms_event(sms): - """Send an SMS event.""" - data = { - ATTR_HOST: modem_data.host, - ATTR_SMS_ID: sms.id, - ATTR_FROM: sms.sender, - ATTR_MESSAGE: sms.message, - } - hass.bus.async_fire(EVENT_SMS, data) - - await modem_data.modem.add_sms_listener(fire_sms_event) - - await modem_data.async_update() - hass.data[DOMAIN].modem_data[modem_data.host] = modem_data diff --git a/homeassistant/components/netgear_lte/binary_sensor.py b/homeassistant/components/netgear_lte/binary_sensor.py index 43a9c1bd260..280d240b90f 100644 --- a/homeassistant/components/netgear_lte/binary_sensor.py +++ b/homeassistant/components/netgear_lte/binary_sensor.py @@ -7,12 +7,11 @@ from homeassistant.components.binary_sensor import ( BinarySensorEntity, BinarySensorEntityDescription, ) -from homeassistant.config_entries import ConfigEntry from homeassistant.const import EntityCategory from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback -from .const import DOMAIN +from . import NetgearLTEConfigEntry from .entity import LTEEntity BINARY_SENSORS: tuple[BinarySensorEntityDescription, ...] = ( @@ -38,13 +37,13 @@ BINARY_SENSORS: tuple[BinarySensorEntityDescription, ...] = ( async def async_setup_entry( - hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback + hass: HomeAssistant, + entry: NetgearLTEConfigEntry, + async_add_entities: AddEntitiesCallback, ) -> None: """Set up the Netgear LTE binary sensor.""" - modem_data = hass.data[DOMAIN].get_modem_data(entry.data) - async_add_entities( - NetgearLTEBinarySensor(entry, modem_data, sensor) for sensor in BINARY_SENSORS + NetgearLTEBinarySensor(entry, description) for description in BINARY_SENSORS ) @@ -54,4 +53,4 @@ class NetgearLTEBinarySensor(LTEEntity, BinarySensorEntity): @property def is_on(self): """Return true if the binary sensor is on.""" - return getattr(self.modem_data.data, self.entity_description.key) + return getattr(self.coordinator.data, self.entity_description.key) diff --git a/homeassistant/components/netgear_lte/const.py b/homeassistant/components/netgear_lte/const.py index 69a96c289e8..1b8a96319c2 100644 --- a/homeassistant/components/netgear_lte/const.py +++ b/homeassistant/components/netgear_lte/const.py @@ -16,9 +16,9 @@ CONF_NOTIFY: Final = "notify" CONF_SENSOR: Final = "sensor" DATA_HASS_CONFIG = "netgear_lte_hass_config" +DATA_SESSION = "session" # https://kb.netgear.com/31160/How-do-I-change-my-4G-LTE-Modem-s-IP-address-range DEFAULT_HOST = "192.168.5.1" -DISPATCHER_NETGEAR_LTE = "netgear_lte_update" DOMAIN: Final = "netgear_lte" FAILOVER_MODES = ["auto", "wire", "mobile"] diff --git a/homeassistant/components/netgear_lte/coordinator.py b/homeassistant/components/netgear_lte/coordinator.py new file mode 100644 index 00000000000..afd0cb743bf --- /dev/null +++ b/homeassistant/components/netgear_lte/coordinator.py @@ -0,0 +1,43 @@ +"""Data update coordinator for the Netgear LTE integration.""" + +from __future__ import annotations + +from datetime import timedelta +from typing import TYPE_CHECKING + +from eternalegypt.eternalegypt import Error, Information, Modem + +from homeassistant.core import HomeAssistant +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed + +from .const import DOMAIN, LOGGER + +if TYPE_CHECKING: + from . import NetgearLTEConfigEntry + + +class NetgearLTEDataUpdateCoordinator(DataUpdateCoordinator[Information]): + """Data update coordinator for the Netgear LTE integration.""" + + config_entry: NetgearLTEConfigEntry + + def __init__( + self, + hass: HomeAssistant, + modem: Modem, + ) -> None: + """Initialize the coordinator.""" + super().__init__( + hass=hass, + logger=LOGGER, + name=DOMAIN, + update_interval=timedelta(seconds=10), + ) + self.modem = modem + + async def _async_update_data(self) -> Information: + """Get the latest data.""" + try: + return await self.modem.information() + except Error as ex: + raise UpdateFailed(ex) from ex diff --git a/homeassistant/components/netgear_lte/entity.py b/homeassistant/components/netgear_lte/entity.py index 0ec16ceff9d..3353da6dc77 100644 --- a/homeassistant/components/netgear_lte/entity.py +++ b/homeassistant/components/netgear_lte/entity.py @@ -1,54 +1,36 @@ """Entity representing a Netgear LTE entity.""" -from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_HOST from homeassistant.helpers.device_registry import DeviceInfo -from homeassistant.helpers.dispatcher import async_dispatcher_connect -from homeassistant.helpers.entity import Entity, EntityDescription +from homeassistant.helpers.entity import EntityDescription +from homeassistant.helpers.update_coordinator import CoordinatorEntity -from . import ModemData -from .const import DISPATCHER_NETGEAR_LTE, DOMAIN, MANUFACTURER +from . import NetgearLTEConfigEntry +from .const import DOMAIN, MANUFACTURER +from .coordinator import NetgearLTEDataUpdateCoordinator -class LTEEntity(Entity): +class LTEEntity(CoordinatorEntity[NetgearLTEDataUpdateCoordinator]): """Base LTE entity.""" _attr_has_entity_name = True - _attr_should_poll = False def __init__( self, - config_entry: ConfigEntry, - modem_data: ModemData, + entry: NetgearLTEConfigEntry, description: EntityDescription, ) -> None: """Initialize a Netgear LTE entity.""" + super().__init__(entry.runtime_data) self.entity_description = description - self.modem_data = modem_data - self._attr_unique_id = f"{description.key}_{modem_data.data.serial_number}" + data = entry.runtime_data.data + self._attr_unique_id = f"{description.key}_{data.serial_number}" self._attr_device_info = DeviceInfo( - configuration_url=f"http://{config_entry.data[CONF_HOST]}", - identifiers={(DOMAIN, modem_data.data.serial_number)}, + configuration_url=f"http://{entry.data[CONF_HOST]}", + identifiers={(DOMAIN, data.serial_number)}, manufacturer=MANUFACTURER, - model=modem_data.data.items["general.model"], - serial_number=modem_data.data.serial_number, - sw_version=modem_data.data.items["general.fwversion"], - hw_version=modem_data.data.items["general.hwversion"], + model=data.items["general.model"], + serial_number=data.serial_number, + sw_version=data.items["general.fwversion"], + hw_version=data.items["general.hwversion"], ) - - async def async_added_to_hass(self) -> None: - """Register callback.""" - self.async_on_remove( - async_dispatcher_connect( - self.hass, DISPATCHER_NETGEAR_LTE, self.async_write_ha_state - ) - ) - - async def async_update(self) -> None: - """Force update of state.""" - await self.modem_data.async_update() - - @property - def available(self) -> bool: - """Return the availability of the sensor.""" - return self.modem_data.data is not None diff --git a/homeassistant/components/netgear_lte/notify.py b/homeassistant/components/netgear_lte/notify.py index 97ba402dc35..763581b9cad 100644 --- a/homeassistant/components/netgear_lte/notify.py +++ b/homeassistant/components/netgear_lte/notify.py @@ -2,15 +2,17 @@ from __future__ import annotations -import attr +from typing import Any + import eternalegypt +from eternalegypt.eternalegypt import Modem from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService from homeassistant.const import CONF_RECIPIENT from homeassistant.core import HomeAssistant from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType -from .const import CONF_NOTIFY, DOMAIN, LOGGER +from .const import CONF_NOTIFY, LOGGER async def async_get_service( @@ -22,21 +24,25 @@ async def async_get_service( if discovery_info is None: return None - return NetgearNotifyService(hass, discovery_info) + return NetgearNotifyService(config, discovery_info) -@attr.s class NetgearNotifyService(BaseNotificationService): """Implementation of a notification service.""" - hass = attr.ib() - config = attr.ib() + def __init__( + self, + config: ConfigType, + discovery_info: dict[str, Any], + ) -> None: + """Initialize the service.""" + self.config = config + self.modem: Modem = discovery_info["modem"] async def async_send_message(self, message="", **kwargs): """Send a message to a user.""" - modem_data = self.hass.data[DOMAIN].get_modem_data(self.config) - if not modem_data: + if not self.modem.token: LOGGER.error("Modem not ready") return if not (targets := kwargs.get(ATTR_TARGET)): @@ -50,6 +56,6 @@ class NetgearNotifyService(BaseNotificationService): for target in targets: try: - await modem_data.modem.sms(target, message) + await self.modem.sms(target, message) except eternalegypt.Error: LOGGER.error("Unable to send to %s", target) diff --git a/homeassistant/components/netgear_lte/sensor.py b/homeassistant/components/netgear_lte/sensor.py index 62b4796f068..73e5de7eaeb 100644 --- a/homeassistant/components/netgear_lte/sensor.py +++ b/homeassistant/components/netgear_lte/sensor.py @@ -5,12 +5,13 @@ from __future__ import annotations from collections.abc import Callable from dataclasses import dataclass +from eternalegypt.eternalegypt import Information + from homeassistant.components.sensor import ( SensorDeviceClass, SensorEntity, SensorEntityDescription, ) -from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( PERCENTAGE, SIGNAL_STRENGTH_DECIBELS_MILLIWATT, @@ -21,8 +22,7 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.typing import StateType -from . import ModemData -from .const import DOMAIN +from . import NetgearLTEConfigEntry from .entity import LTEEntity @@ -30,7 +30,7 @@ from .entity import LTEEntity class NetgearLTESensorEntityDescription(SensorEntityDescription): """Class describing Netgear LTE entities.""" - value_fn: Callable[[ModemData], StateType] | None = None + value_fn: Callable[[Information], StateType] | None = None SENSORS: tuple[NetgearLTESensorEntityDescription, ...] = ( @@ -38,13 +38,13 @@ SENSORS: tuple[NetgearLTESensorEntityDescription, ...] = ( key="sms", translation_key="sms", native_unit_of_measurement="unread", - value_fn=lambda modem_data: sum(1 for x in modem_data.data.sms if x.unread), + value_fn=lambda data: sum(1 for x in data.sms if x.unread), ), NetgearLTESensorEntityDescription( key="sms_total", translation_key="sms_total", native_unit_of_measurement="messages", - value_fn=lambda modem_data: len(modem_data.data.sms), + value_fn=lambda data: len(data.sms), ), NetgearLTESensorEntityDescription( key="usage", @@ -54,7 +54,7 @@ SENSORS: tuple[NetgearLTESensorEntityDescription, ...] = ( native_unit_of_measurement=UnitOfInformation.BYTES, suggested_unit_of_measurement=UnitOfInformation.MEBIBYTES, suggested_display_precision=1, - value_fn=lambda modem_data: modem_data.data.usage, + value_fn=lambda data: data.usage, ), NetgearLTESensorEntityDescription( key="radio_quality", @@ -125,14 +125,12 @@ SENSORS: tuple[NetgearLTESensorEntityDescription, ...] = ( async def async_setup_entry( - hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback + hass: HomeAssistant, + entry: NetgearLTEConfigEntry, + async_add_entities: AddEntitiesCallback, ) -> None: """Set up the Netgear LTE sensor.""" - modem_data = hass.data[DOMAIN].get_modem_data(entry.data) - - async_add_entities( - NetgearLTESensor(entry, modem_data, sensor) for sensor in SENSORS - ) + async_add_entities(NetgearLTESensor(entry, description) for description in SENSORS) class NetgearLTESensor(LTEEntity, SensorEntity): @@ -144,5 +142,5 @@ class NetgearLTESensor(LTEEntity, SensorEntity): def native_value(self) -> StateType: """Return the state of the sensor.""" if self.entity_description.value_fn is not None: - return self.entity_description.value_fn(self.modem_data) - return getattr(self.modem_data.data, self.entity_description.key) + return self.entity_description.value_fn(self.coordinator.data) + return getattr(self.coordinator.data, self.entity_description.key) diff --git a/homeassistant/components/netgear_lte/services.py b/homeassistant/components/netgear_lte/services.py index 02000820119..77ed1b91f31 100644 --- a/homeassistant/components/netgear_lte/services.py +++ b/homeassistant/components/netgear_lte/services.py @@ -1,10 +1,8 @@ """Services for the Netgear LTE integration.""" -from typing import TYPE_CHECKING - +from eternalegypt.eternalegypt import Modem import voluptuous as vol -from homeassistant.const import CONF_HOST from homeassistant.core import HomeAssistant, ServiceCall from homeassistant.helpers import config_validation as cv @@ -19,9 +17,6 @@ from .const import ( LOGGER, ) -if TYPE_CHECKING: - from . import LTEData, ModemData - SERVICE_DELETE_SMS = "delete_sms" SERVICE_SET_OPTION = "set_option" SERVICE_CONNECT_LTE = "connect_lte" @@ -50,31 +45,29 @@ CONNECT_LTE_SCHEMA = vol.Schema({vol.Optional(ATTR_HOST): cv.string}) DISCONNECT_LTE_SCHEMA = vol.Schema({vol.Optional(ATTR_HOST): cv.string}) -async def async_setup_services(hass: HomeAssistant) -> None: +async def async_setup_services(hass: HomeAssistant, modem: Modem) -> None: """Set up services for Netgear LTE integration.""" async def service_handler(call: ServiceCall) -> None: """Apply a service.""" host = call.data.get(ATTR_HOST) - data: LTEData = hass.data[DOMAIN] - modem_data: ModemData = data.get_modem_data({CONF_HOST: host}) - if not modem_data: + if not modem.token: LOGGER.error("%s: host %s unavailable", call.service, host) return if call.service == SERVICE_DELETE_SMS: for sms_id in call.data[ATTR_SMS_ID]: - await modem_data.modem.delete_sms(sms_id) + await modem.delete_sms(sms_id) elif call.service == SERVICE_SET_OPTION: if failover := call.data.get(ATTR_FAILOVER): - await modem_data.modem.set_failover_mode(failover) + await modem.set_failover_mode(failover) if autoconnect := call.data.get(ATTR_AUTOCONNECT): - await modem_data.modem.set_autoconnect_mode(autoconnect) + await modem.set_autoconnect_mode(autoconnect) elif call.service == SERVICE_CONNECT_LTE: - await modem_data.modem.connect_lte() + await modem.connect_lte() elif call.service == SERVICE_DISCONNECT_LTE: - await modem_data.modem.disconnect_lte() + await modem.disconnect_lte() service_schemas = { SERVICE_DELETE_SMS: DELETE_SMS_SCHEMA, diff --git a/tests/components/netgear_lte/test_config_flow.py b/tests/components/netgear_lte/test_config_flow.py index 16feb88172b..ec649f4def0 100644 --- a/tests/components/netgear_lte/test_config_flow.py +++ b/tests/components/netgear_lte/test_config_flow.py @@ -2,7 +2,6 @@ from unittest.mock import patch -from homeassistant import data_entry_flow from homeassistant.components.netgear_lte.const import DOMAIN from homeassistant.config_entries import SOURCE_USER from homeassistant.const import CONF_SOURCE @@ -25,7 +24,7 @@ async def test_flow_user_form(hass: HomeAssistant, connection: None) -> None: context={CONF_SOURCE: SOURCE_USER}, ) - assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["type"] is FlowResultType.FORM assert result["step_id"] == "user" with _patch_setup(): @@ -33,7 +32,7 @@ async def test_flow_user_form(hass: HomeAssistant, connection: None) -> None: result["flow_id"], user_input=CONF_DATA, ) - assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY + assert result["type"] is FlowResultType.CREATE_ENTRY assert result["title"] == "Netgear LM1200" assert result["data"] == CONF_DATA assert result["context"]["unique_id"] == "FFFFFFFFFFFFF" @@ -63,7 +62,7 @@ async def test_flow_user_cannot_connect( data=CONF_DATA, ) - assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["type"] is FlowResultType.FORM assert result["step_id"] == "user" assert result["errors"]["base"] == "cannot_connect" @@ -78,6 +77,6 @@ async def test_flow_user_unknown_error(hass: HomeAssistant, unknown: None) -> No result["flow_id"], user_input=CONF_DATA, ) - assert result["type"] == data_entry_flow.RESULT_TYPE_FORM + assert result["type"] is FlowResultType.FORM assert result["step_id"] == "user" assert result["errors"]["base"] == "unknown" diff --git a/tests/components/netgear_lte/test_init.py b/tests/components/netgear_lte/test_init.py index ef3109123fa..ca5a22cf259 100644 --- a/tests/components/netgear_lte/test_init.py +++ b/tests/components/netgear_lte/test_init.py @@ -1,14 +1,22 @@ """Test Netgear LTE integration.""" +from datetime import timedelta +from unittest.mock import patch + +from eternalegypt.eternalegypt import Error from syrupy.assertion import SnapshotAssertion from homeassistant.components.netgear_lte.const import DOMAIN from homeassistant.config_entries import ConfigEntryState +from homeassistant.const import STATE_UNAVAILABLE from homeassistant.core import HomeAssistant from homeassistant.helpers import device_registry as dr +import homeassistant.util.dt as dt_util from .conftest import CONF_DATA +from tests.common import async_fire_time_changed + async def test_setup_unload(hass: HomeAssistant, setup_integration: None) -> None: """Test setup and unload.""" @@ -43,3 +51,21 @@ async def test_device( await hass.async_block_till_done() device = device_registry.async_get_device(identifiers={(DOMAIN, entry.unique_id)}) assert device == snapshot + + +async def test_update_failed( + hass: HomeAssistant, + entity_registry_enabled_by_default: None, + setup_integration: None, +) -> None: + """Test coordinator throws UpdateFailed after failed update.""" + with patch( + "homeassistant.components.netgear_lte.eternalegypt.Modem.information", + side_effect=Error, + ) as updater: + next_update = dt_util.utcnow() + timedelta(seconds=10) + async_fire_time_changed(hass, next_update) + await hass.async_block_till_done() + updater.assert_called_once() + state = hass.states.get("sensor.netgear_lm1200_radio_quality") + assert state.state == STATE_UNAVAILABLE