Add config flow to Netgear LTE (#93002)

* Add config flow to Netgear LTE

* uno mas

* uno mas

* forgot one

* uno mas

* uno mas

* apply suggestions

* tweak user step

* fix load/unload/dep

* clean up

* fix tests

* test yaml before importing

* uno mas

* uno mas

* uno mas

* uno mas

* uno mas

* fix startup hanging

* break out yaml import

* fix doc string

---------

Co-authored-by: Robert Resch <robert@resch.dev>
This commit is contained in:
Robert Hillis 2023-12-25 23:19:28 -05:00 committed by GitHub
parent f0e080f958
commit 6f9bff7602
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 1169 additions and 161 deletions

View File

@ -804,7 +804,8 @@ omit =
homeassistant/components/netgear/sensor.py
homeassistant/components/netgear/switch.py
homeassistant/components/netgear/update.py
homeassistant/components/netgear_lte/*
homeassistant/components/netgear_lte/__init__.py
homeassistant/components/netgear_lte/notify.py
homeassistant/components/netio/switch.py
homeassistant/components/neurio_energy/sensor.py
homeassistant/components/nexia/climate.py

View File

@ -847,6 +847,7 @@ build.json @home-assistant/supervisor
/homeassistant/components/netgear/ @hacf-fr @Quentame @starkillerOG
/tests/components/netgear/ @hacf-fr @Quentame @starkillerOG
/homeassistant/components/netgear_lte/ @tkdrob
/tests/components/netgear_lte/ @tkdrob
/homeassistant/components/network/ @home-assistant/core
/tests/components/network/ @home-assistant/core
/homeassistant/components/nexia/ @bdraco

View File

@ -1,12 +1,12 @@
"""Support for Netgear LTE modems."""
import asyncio
from datetime import timedelta
import aiohttp
from aiohttp.cookiejar import CookieJar
import attr
import eternalegypt
import voluptuous as vol
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry, ConfigEntryState
from homeassistant.const import (
CONF_HOST,
CONF_MONITORED_CONDITIONS,
@ -16,11 +16,13 @@ from homeassistant.const import (
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.core import DOMAIN as HOMEASSISTANT_DOMAIN, Event, 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.issue_registry import IssueSeverity, async_create_issue
from homeassistant.helpers.typing import ConfigType
from . import sensor_types
@ -32,6 +34,7 @@ from .const import (
CONF_BINARY_SENSOR,
CONF_NOTIFY,
CONF_SENSOR,
DATA_HASS_CONFIG,
DISPATCHER_NETGEAR_LTE,
DOMAIN,
LOGGER,
@ -90,6 +93,12 @@ CONFIG_SCHEMA = vol.Schema(
extra=vol.ALLOW_EXTRA,
)
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.NOTIFY,
Platform.SENSOR,
]
@attr.s
class ModemData:
@ -137,90 +146,108 @@ class LTEData:
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up Netgear LTE component."""
if DOMAIN not in hass.data:
websession = async_create_clientsession(
hass, cookie_jar=aiohttp.CookieJar(unsafe=True)
)
hass.data[DOMAIN] = LTEData(websession)
hass.data[DATA_HASS_CONFIG] = config
await async_setup_services(hass)
netgear_lte_config = config[DOMAIN]
# Set up each modem
tasks = [
hass.async_create_task(_setup_lte(hass, lte_conf))
for lte_conf in netgear_lte_config
]
await asyncio.wait(tasks)
# Load platforms for each modem
for lte_conf in netgear_lte_config:
# Notify
for notify_conf in lte_conf[CONF_NOTIFY]:
discovery_info = {
CONF_HOST: lte_conf[CONF_HOST],
CONF_NAME: notify_conf.get(CONF_NAME),
CONF_NOTIFY: notify_conf,
}
hass.async_create_task(
discovery.async_load_platform(
hass, Platform.NOTIFY, DOMAIN, discovery_info, config
)
)
# Sensor
sensor_conf = lte_conf[CONF_SENSOR]
discovery_info = {CONF_HOST: lte_conf[CONF_HOST], CONF_SENSOR: sensor_conf}
hass.async_create_task(
discovery.async_load_platform(
hass, Platform.SENSOR, DOMAIN, discovery_info, config
)
)
# Binary Sensor
binary_sensor_conf = lte_conf[CONF_BINARY_SENSOR]
discovery_info = {
CONF_HOST: lte_conf[CONF_HOST],
CONF_BINARY_SENSOR: binary_sensor_conf,
}
hass.async_create_task(
discovery.async_load_platform(
hass, Platform.BINARY_SENSOR, DOMAIN, discovery_info, config
)
)
if lte_config := config.get(DOMAIN):
await hass.async_create_task(import_yaml(hass, lte_config))
return True
async def _setup_lte(hass, lte_config):
"""Set up a Netgear LTE modem."""
async def import_yaml(hass: HomeAssistant, lte_config: ConfigType) -> None:
"""Import yaml if we can connect. Create appropriate issue registry entries."""
for entry in lte_config:
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": SOURCE_IMPORT}, data=entry
)
if result.get("reason") == "cannot_connect":
async_create_issue(
hass,
DOMAIN,
"import_failure",
is_fixable=False,
severity=IssueSeverity.ERROR,
translation_key="import_failure",
)
else:
async_create_issue(
hass,
HOMEASSISTANT_DOMAIN,
f"deprecated_yaml_{DOMAIN}",
breaks_in_ha_version="2024.7.0",
is_fixable=False,
issue_domain=DOMAIN,
severity=IssueSeverity.WARNING,
translation_key="deprecated_yaml",
translation_placeholders={
"domain": DOMAIN,
"integration_title": "Netgear LTE",
},
)
host = lte_config[CONF_HOST]
password = lte_config[CONF_PASSWORD]
websession = hass.data[DOMAIN].websession
modem = eternalegypt.Modem(hostname=host, websession=websession)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Netgear LTE from a config entry."""
host = entry.data[CONF_HOST]
password = entry.data[CONF_PASSWORD]
if DOMAIN not in hass.data:
websession = async_create_clientsession(hass, cookie_jar=CookieJar(unsafe=True))
hass.data[DOMAIN] = LTEData(websession)
modem = eternalegypt.Modem(hostname=host, websession=hass.data[DOMAIN].websession)
modem_data = ModemData(hass, host, modem)
try:
await _login(hass, modem_data, password)
except eternalegypt.Error:
retry_task = hass.loop.create_task(_retry_login(hass, modem_data, password))
await _login(hass, modem_data, password)
@callback
def cleanup_retry(event):
"""Clean up retry task resources."""
if not retry_task.done():
retry_task.cancel()
async def _update(now):
"""Periodic update."""
await modem_data.async_update()
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, cleanup_retry)
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)
_legacy_task(hass, entry)
await hass.config_entries.async_forward_entry_setups(
entry, [platform for platform in PLATFORMS if platform != Platform.NOTIFY]
)
return True
async def _login(hass, modem_data, password):
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)
loaded_entries = [
entry
for entry in hass.config_entries.async_entries(DOMAIN)
if entry.state == ConfigEntryState.LOADED
]
if len(loaded_entries) == 1:
hass.data.pop(DOMAIN)
return unload_ok
async def _login(hass: HomeAssistant, modem_data: ModemData, password: str) -> None:
"""Log in and complete setup."""
await modem_data.modem.login(password=password)
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."""
@ -237,33 +264,63 @@ async def _login(hass, modem_data, password):
await modem_data.async_update()
hass.data[DOMAIN].modem_data[modem_data.host] = modem_data
async def _update(now):
"""Periodic update."""
await modem_data.async_update()
update_unsub = async_track_time_interval(hass, _update, SCAN_INTERVAL)
def _legacy_task(hass: HomeAssistant, entry: ConfigEntry) -> None:
"""Create notify service and add a repair issue when appropriate."""
# Discovery can happen up to 2 times for notify depending on existing yaml config
# One for the name of the config entry, allows the user to customize the name
# One for each notify described in the yaml config which goes away with config flow
# One for the default if the user never specified one
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.NOTIFY,
DOMAIN,
{CONF_HOST: entry.data[CONF_HOST], CONF_NAME: entry.title},
hass.data[DATA_HASS_CONFIG],
)
)
if not (lte_configs := hass.data[DATA_HASS_CONFIG].get(DOMAIN, [])):
return
async_create_issue(
hass,
DOMAIN,
"deprecated_notify",
breaks_in_ha_version="2024.7.0",
is_fixable=False,
severity=IssueSeverity.WARNING,
translation_key="deprecated_notify",
translation_placeholders={
"name": f"{Platform.NOTIFY}.{entry.title.lower().replace(' ', '_')}"
},
)
async def cleanup(event):
"""Clean up resources."""
update_unsub()
await modem_data.modem.logout()
del hass.data[DOMAIN].modem_data[modem_data.host]
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, cleanup)
async def _retry_login(hass, modem_data, password):
"""Sleep and retry setup."""
LOGGER.warning("Could not connect to %s. Will keep trying", modem_data.host)
modem_data.connected = False
delay = 15
while not modem_data.connected:
await asyncio.sleep(delay)
try:
await _login(hass, modem_data, password)
except eternalegypt.Error:
delay = min(2 * delay, 300)
for lte_config in lte_configs:
if lte_config[CONF_HOST] == entry.data[CONF_HOST]:
if not lte_config[CONF_NOTIFY]:
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.NOTIFY,
DOMAIN,
{CONF_HOST: entry.data[CONF_HOST], CONF_NAME: DOMAIN},
hass.data[DATA_HASS_CONFIG],
)
)
break
for notify_conf in lte_config[CONF_NOTIFY]:
discovery_info = {
CONF_HOST: lte_config[CONF_HOST],
CONF_NAME: notify_conf.get(CONF_NAME),
CONF_NOTIFY: notify_conf,
}
hass.async_create_task(
discovery.async_load_platform(
hass,
Platform.NOTIFY,
DOMAIN,
discovery_info,
hass.data[DATA_HASS_CONFIG],
)
)
break

View File

@ -2,40 +2,24 @@
from __future__ import annotations
from homeassistant.components.binary_sensor import BinarySensorEntity
from homeassistant.const import CONF_MONITORED_CONDITIONS
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import CONF_BINARY_SENSOR, DOMAIN
from .const import DOMAIN
from .entity import LTEEntity
from .sensor_types import BINARY_SENSOR_CLASSES
from .sensor_types import ALL_BINARY_SENSORS, BINARY_SENSOR_CLASSES
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Netgear LTE binary sensor devices."""
if discovery_info is None:
return
"""Set up the Netgear LTE binary sensor."""
modem_data = hass.data[DOMAIN].get_modem_data(entry.data)
modem_data = hass.data[DOMAIN].get_modem_data(discovery_info)
if not modem_data or not modem_data.data:
raise PlatformNotReady
binary_sensor_conf = discovery_info[CONF_BINARY_SENSOR]
monitored_conditions = binary_sensor_conf[CONF_MONITORED_CONDITIONS]
binary_sensors = []
for sensor_type in monitored_conditions:
binary_sensors.append(LTEBinarySensor(modem_data, sensor_type))
async_add_entities(binary_sensors)
async_add_entities(
LTEBinarySensor(modem_data, sensor) for sensor in ALL_BINARY_SENSORS
)
class LTEBinarySensor(LTEEntity, BinarySensorEntity):

View File

@ -0,0 +1,103 @@
"""Config flow for Netgear LTE integration."""
from __future__ import annotations
from typing import Any
from aiohttp.cookiejar import CookieJar
from eternalegypt import Error, Modem
from eternalegypt.eternalegypt import Information
import voluptuous as vol
from homeassistant import config_entries, exceptions
from homeassistant.const import CONF_HOST, CONF_PASSWORD
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .const import DEFAULT_HOST, DOMAIN, LOGGER, MANUFACTURER
class NetgearLTEFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Netgear LTE."""
async def async_step_import(self, config: dict[str, Any]) -> FlowResult:
"""Import a configuration from config.yaml."""
host = config[CONF_HOST]
password = config[CONF_PASSWORD]
self._async_abort_entries_match({CONF_HOST: host})
try:
info = await self._async_validate_input(host, password)
except InputValidationError:
return self.async_abort(reason="cannot_connect")
await self.async_set_unique_id(info.serial_number)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"{MANUFACTURER} {info.items['general.devicename']}",
data={CONF_HOST: host, CONF_PASSWORD: password},
)
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
errors = {}
if user_input:
host = user_input[CONF_HOST]
password = user_input[CONF_PASSWORD]
try:
info = await self._async_validate_input(host, password)
except InputValidationError as ex:
errors["base"] = ex.base
else:
await self.async_set_unique_id(info.serial_number)
self._abort_if_unique_id_configured()
return self.async_create_entry(
title=f"{MANUFACTURER} {info.items['general.devicename']}",
data={CONF_HOST: host, CONF_PASSWORD: password},
)
return self.async_show_form(
step_id="user",
data_schema=self.add_suggested_values_to_schema(
vol.Schema(
{
vol.Required(CONF_HOST): str,
vol.Required(CONF_PASSWORD): str,
}
),
user_input or {CONF_HOST: DEFAULT_HOST},
),
errors=errors,
)
async def _async_validate_input(self, host: str, password: str) -> Information:
"""Validate login credentials."""
websession = async_create_clientsession(
self.hass, cookie_jar=CookieJar(unsafe=True)
)
modem = Modem(
hostname=host,
password=password,
websession=websession,
)
try:
await modem.login()
info = await modem.information()
except Error as ex:
raise InputValidationError("cannot_connect") from ex
except Exception as ex:
LOGGER.exception("Unexpected exception")
raise InputValidationError("unknown") from ex
await modem.logout()
return info
class InputValidationError(exceptions.HomeAssistantError):
"""Error to indicate we cannot proceed due to invalid input."""
def __init__(self, base: str) -> None:
"""Initialize with error base."""
super().__init__()
self.base = base

View File

@ -14,9 +14,14 @@ CONF_BINARY_SENSOR: Final = "binary_sensor"
CONF_NOTIFY: Final = "notify"
CONF_SENSOR: Final = "sensor"
DATA_HASS_CONFIG = "netgear_lte_hass_config"
# 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"]
LOGGER = logging.getLogger(__package__)
MANUFACTURER: Final = "Netgear"

View File

@ -2,6 +2,7 @@
"domain": "netgear_lte",
"name": "NETGEAR LTE",
"codeowners": ["@tkdrob"],
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/netgear_lte",
"iot_class": "local_polling",
"loggers": ["eternalegypt"],

View File

@ -38,8 +38,8 @@ class NetgearNotifyService(BaseNotificationService):
if not modem_data:
LOGGER.error("Modem not ready")
return
targets = kwargs.get(ATTR_TARGET, self.config[CONF_NOTIFY][CONF_RECIPIENT])
if not (targets := kwargs.get(ATTR_TARGET)):
targets = self.config[CONF_NOTIFY][CONF_RECIPIENT]
if not targets:
LOGGER.warning("No recipients")
return

View File

@ -2,45 +2,37 @@
from __future__ import annotations
from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
from homeassistant.const import CONF_MONITORED_CONDITIONS
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import PlatformNotReady
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import CONF_SENSOR, DOMAIN
from .const import DOMAIN
from .entity import LTEEntity
from .sensor_types import SENSOR_SMS, SENSOR_SMS_TOTAL, SENSOR_UNITS, SENSOR_USAGE
from .sensor_types import (
ALL_SENSORS,
SENSOR_SMS,
SENSOR_SMS_TOTAL,
SENSOR_UNITS,
SENSOR_USAGE,
)
async def async_setup_platform(
hass: HomeAssistant,
config: ConfigType,
async_add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
"""Set up Netgear LTE sensor devices."""
if discovery_info is None:
return
modem_data = hass.data[DOMAIN].get_modem_data(discovery_info)
if not modem_data or not modem_data.data:
raise PlatformNotReady
sensor_conf = discovery_info[CONF_SENSOR]
monitored_conditions = sensor_conf[CONF_MONITORED_CONDITIONS]
"""Set up the Netgear LTE sensor."""
modem_data = hass.data[DOMAIN].get_modem_data(entry.data)
sensors: list[SensorEntity] = []
for sensor_type in monitored_conditions:
if sensor_type == SENSOR_SMS:
sensors.append(SMSUnreadSensor(modem_data, sensor_type))
elif sensor_type == SENSOR_SMS_TOTAL:
sensors.append(SMSTotalSensor(modem_data, sensor_type))
elif sensor_type == SENSOR_USAGE:
sensors.append(UsageSensor(modem_data, sensor_type))
for sensor in ALL_SENSORS:
if sensor == SENSOR_SMS:
sensors.append(SMSUnreadSensor(modem_data, sensor))
elif sensor == SENSOR_SMS_TOTAL:
sensors.append(SMSTotalSensor(modem_data, sensor))
elif sensor == SENSOR_USAGE:
sensors.append(UsageSensor(modem_data, sensor))
else:
sensors.append(GenericSensor(modem_data, sensor_type))
sensors.append(GenericSensor(modem_data, sensor))
async_add_entities(sensors)

View File

@ -1,4 +1,32 @@
{
"config": {
"step": {
"user": {
"data": {
"host": "[%key:common::config_flow::data::host%]",
"password": "[%key:common::config_flow::data::password%]"
}
}
},
"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%]",
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
}
},
"issues": {
"deprecated_notify": {
"title": "The Netgear LTE notify service is changing",
"description": "The Netgear LTE notify service was previously set up via YAML configuration.\n\nThis created a service for a specified recipient without having to include the phone number.\n\nPlease adjust any automations or scripts you may have to use the `{name}` service and include target for specifying a recipient."
},
"import_failure": {
"title": "The Netgear LTE integration failed to import",
"description": "The Netgear LTE notify service was previously set up via YAML configuration.\n\nAn error occurred when trying to communicate with the device while attempting to import the configuration to the UI.\n\nPlease remove the Netgear LTE notify section from your YAML configuration and set it up in the UI instead."
}
},
"services": {
"delete_sms": {
"name": "Delete SMS",
@ -52,6 +80,5 @@
}
}
}
},
"selector": {}
}
}

View File

@ -319,6 +319,7 @@ FLOWS = {
"nest",
"netatmo",
"netgear",
"netgear_lte",
"nexia",
"nextbus",
"nextcloud",

View File

@ -3791,7 +3791,7 @@
},
"netgear_lte": {
"integration_type": "hub",
"config_flow": false,
"config_flow": true,
"iot_class": "local_polling",
"name": "NETGEAR LTE"
}

View File

@ -641,6 +641,9 @@ epson-projector==0.5.1
# homeassistant.components.esphome
esphome-dashboard-api==1.2.3
# homeassistant.components.netgear_lte
eternalegypt==0.0.16
# homeassistant.components.eufylife_ble
eufylife-ble-client==0.1.8

View File

@ -0,0 +1 @@
"""Tests for the Netgear LTE component."""

View File

@ -0,0 +1,85 @@
"""Configure pytest for Netgear LTE tests."""
from __future__ import annotations
from aiohttp.client_exceptions import ClientError
import pytest
from homeassistant.components.netgear_lte.const import DOMAIN
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONTENT_TYPE_JSON
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry, load_fixture
from tests.test_util.aiohttp import AiohttpClientMocker
HOST = "192.168.5.1"
PASSWORD = "password"
CONF_DATA = {CONF_HOST: HOST, CONF_PASSWORD: PASSWORD}
@pytest.fixture
def cannot_connect(aioclient_mock: AiohttpClientMocker) -> None:
"""Mock cannot connect error."""
aioclient_mock.get(f"http://{HOST}/model.json", exc=ClientError)
aioclient_mock.post(f"http://{HOST}/Forms/config", exc=ClientError)
@pytest.fixture
def unknown(aioclient_mock: AiohttpClientMocker) -> None:
"""Mock Netgear LTE unknown error."""
aioclient_mock.get(
f"http://{HOST}/model.json",
text="something went wrong",
headers={"Content-Type": "application/javascript"},
)
@pytest.fixture(name="connection")
def mock_connection(aioclient_mock: AiohttpClientMocker) -> None:
"""Mock Netgear LTE connection."""
aioclient_mock.get(
f"http://{HOST}/model.json",
text=load_fixture("netgear_lte/model.json"),
headers={"Content-Type": CONTENT_TYPE_JSON},
)
aioclient_mock.post(
f"http://{HOST}/Forms/config",
headers={"Content-Type": CONTENT_TYPE_JSON},
)
aioclient_mock.post(
f"http://{HOST}/Forms/smsSendMsg",
headers={"Content-Type": CONTENT_TYPE_JSON},
)
@pytest.fixture(name="config_entry")
def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Create Netgear LTE entry in Home Assistant."""
return MockConfigEntry(
domain=DOMAIN, data=CONF_DATA, unique_id="FFFFFFFFFFFFF", title="Netgear LM1200"
)
@pytest.fixture(name="setup_integration")
async def mock_setup_integration(
hass: HomeAssistant,
config_entry: MockConfigEntry,
connection: None,
) -> None:
"""Set up the Netgear LTE integration in Home Assistant."""
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()
@pytest.fixture(name="setup_cannot_connect")
async def setup_cannot_connect(
hass: HomeAssistant,
config_entry: MockConfigEntry,
cannot_connect: None,
) -> None:
"""Set up the Netgear LTE integration in Home Assistant."""
config_entry.add_to_hass(hass)
assert await async_setup_component(hass, DOMAIN, {})
await hass.async_block_till_done()

View File

@ -0,0 +1,450 @@
{
"custom": { "AtTcpEnable": false, "end": 0 },
"webd": {
"adminPassword": "****************",
"ownerModeEnabled": false,
"hideAdminPassword": true,
"end": ""
},
"lcd": { "end": "" },
"sim": {
"pin": { "mode": "Disabled", "retry": 3, "end": "" },
"puk": { "retry": 10 },
"mep": {},
"phoneNumber": "(555) 555-5555",
"iccid": "1234567890123456789",
"imsi": "123456789012345",
"SPN": "",
"status": "Ready",
"end": ""
},
"sms": {
"ready": true,
"sendEnabled": true,
"sendSupported": true,
"alertSupported": true,
"alertEnabled": false,
"alertNumList": "",
"alertCfgList": [
{ "category": "FWUpdate", "enabled": false },
{ "category": "DataUsageWarning", "enabled": false },
{ "category": "DataUsageExceeded", "enabled": false },
{ "category": "LTEFailoverLTE", "enabled": false },
{ "category": "LTEFailoverETH", "enabled": false },
{}
],
"unreadMsgs": 1,
"msgCount": 1,
"msgs": [
{
"id": "1",
"rxTime": "20/01/23 03:39:35 PM",
"text": "text",
"sender": "889",
"read": false
},
{}
],
"trans": [{}],
"sendMsg": [
{
"clientId": "eternalegypt.eternalegypt",
"enc": "Gsm7Bit",
"errorCode": 0,
"msgId": 1,
"receiver": "+15555555555",
"status": "Succeeded",
"text": "test SMS from Home Assistant",
"txTime": "1367252824"
},
{}
],
"end": ""
},
"session": {
"userRole": "Admin",
"lang": "en",
"secToken": "secret"
},
"general": {
"defaultLanguage": "en",
"PRIid": "12345678",
"genericResetStatus": "NotStarted",
"manufacturer": "Netgear",
"model": "LM1200",
"HWversion": "1.0",
"FWversion": "EC25AFFDR07A09M4G",
"appVersion": "NTG9X07C_20.06.09.00",
"buildDate": "Unknown",
"BLversion": "",
"PRIversion": "04.19",
"IMEI": "123456789012345",
"SVN": "9",
"MEID": "",
"ESN": "0",
"FSN": "FFFFFFFFFFFFF",
"activated": true,
"webAppVersion": "LM1200-HDATA_03.03.103.201",
"HIDenabled": false,
"TCAaccepted": true,
"LEDenabled": true,
"showAdvHelp": true,
"keyLockState": "Unlocked",
"devTemperature": 30,
"verMajor": 1000,
"verMinor": 0,
"environment": "Application",
"currTime": 1367257216,
"timeZoneOffset": -14400,
"deviceName": "LM1200",
"useMetricSystem": true,
"factoryResetStatus": "NotStarted",
"setupCompleted": true,
"languageSelected": false,
"systemAlertList": { "list": [{}], "count": 0 },
"apiVersion": "2.0",
"companyName": "NETGEAR",
"configURL": "/Forms/config",
"profileURL": "/Forms/profile",
"pinChangeURL": "/Forms/pinChange",
"portCfgURL": "/Forms/portCfg",
"portFilterURL": "/Forms/portFilter",
"wifiACLURL": "/Forms/wifiACL",
"supportedLangList": [
{
"id": "en",
"isCurrent": "true",
"isDefault": "true",
"label": "English",
"token1": "/romfs/lcd/en_us.tr",
"token2": ""
},
{
"id": "de_DE",
"isCurrent": "false",
"isDefault": "false",
"label": "Deutsch (German)",
"token1": "/romfs/lcd/de_de.tr",
"token2": ""
},
{
"id": "ar_AR",
"isCurrent": "false",
"isDefault": "false",
"label": "العربية (Arabic)",
"token1": "/romfs/lcd/ar_AR.tr",
"token2": ""
},
{
"id": "es_ES",
"isCurrent": "false",
"isDefault": "false",
"label": "Español (Spanish)",
"token1": "/romfs/lcd/es_es.tr",
"token2": ""
},
{
"id": "fr_FR",
"isCurrent": "false",
"isDefault": "false",
"label": "Français (French)",
"token1": "/romfs/lcd/fr_fr.tr",
"token2": ""
},
{
"id": "it_IT",
"isCurrent": "false",
"isDefault": "false",
"label": "Italiano (Italian)",
"token1": "/romfs/lcd/it_it.tr",
"token2": ""
},
{
"id": "pl_PL",
"isCurrent": "false",
"isDefault": "false",
"label": "Polski (Polish)",
"token1": "/romfs/lcd/pl_pl.tr",
"token2": ""
},
{
"id": "fi_FI",
"isCurrent": "false",
"isDefault": "false",
"label": "Suomi (Finnish)",
"token1": "/romfs/lcd/fi_fi.tr",
"token2": ""
},
{
"id": "sv_SE",
"isCurrent": "false",
"isDefault": "false",
"label": "Svenska (Swedish)",
"token1": "/romfs/lcd/sv_se.tr",
"token2": ""
},
{
"id": "tu_TU",
"isCurrent": "false",
"isDefault": "false",
"label": "Türkçe (Turkish)",
"token1": "/romfs/lcd/tu_tu.tr",
"token2": ""
},
{}
]
},
"power": {
"PMState": "Init",
"SmState": "Online",
"autoOff": {
"onUSBdisconnect": { "enable": false, "countdownTimer": 0, "end": "" },
"onIdle": { "timer": { "onAC": 0, "onBattery": 0, "end": "" } }
},
"standby": {
"onIdle": {
"timer": { "onAC": 0, "onBattery": 600, "onUSB": 0, "end": "" }
}
},
"autoOn": { "enable": true, "end": "" },
"buttonHoldTime": 3,
"deviceTempCritical": false,
"resetreason": 16,
"resetRequired": "NoResetRequired",
"lpm": false,
"end": ""
},
"wwan": {
"netScanStatus": "NotStarted",
"inactivityCause": 307,
"currentNWserviceType": "LteService",
"registerRejectCode": 0,
"netSelEnabled": "Enabled",
"netRegMode": "Auto",
"IPv6": "1234:abcd::1234:abcd",
"roaming": false,
"IP": "10.0.0.5",
"registerNetworkDisplay": "T-Mobile",
"RAT": "Only4G",
"bandRegion": [
{ "index": 0, "name": "Auto", "current": false },
{ "index": 1, "name": "LTE Only", "current": true },
{ "index": 2, "name": "WCDMA Only", "current": false },
{}
],
"autoconnect": "HomeNetwork",
"profileList": [
{
"index": 1,
"id": "T-Mobile 9",
"name": "T Mobile",
"apn": "fast.t-mobile.com",
"username": "",
"password": "",
"authtype": "None",
"ipaddr": "0.0.0.0",
"type": "IPV4V6",
"pdproamingtype": "IPV4"
},
{
"index": 2,
"id": "Mint",
"name": "Mint",
"apn": "wholesale",
"username": "",
"password": "",
"authtype": "None",
"ipaddr": "0.0.0.0",
"type": "IPV4V6",
"pdproamingtype": "IPV4"
},
{}
],
"profile": {
"default": "T-Mobile 9",
"defaultLTE": "T-Mobile 9",
"full": false,
"promptForApnSelection": false,
"end": ""
},
"dataUsage": {
"total": {
"lteBillingTx": 0,
"lteBillingRx": 0,
"cdmaBillingTx": 0,
"cdmaBillingRx": 0,
"gwBillingTx": 0,
"gwBillingRx": 0,
"lteLifeTx": 0,
"lteLifeRx": 0,
"cdmaLifeTx": 0,
"cdmaLifeRx": 0,
"gwLifeTx": 0,
"gwLifeRx": 0,
"end": ""
},
"server": { "accountType": "", "subAccountType": "", "end": "" },
"serverDataRemaining": 0,
"serverDataTransferred": 0,
"serverDataTransferredIntl": 0,
"serverDataValidState": "Invalid",
"serverDaysLeft": 0,
"serverErrorCode": "",
"serverLowBalance": false,
"serverMsisdn": "",
"serverRechargeUrl": "",
"dataWarnEnable": true,
"prepaidAccountState": "Hot",
"accountType": "Unknown",
"share": {
"enabled": false,
"dataTransferredOthers": 0,
"lastSync": "0",
"end": ""
},
"generic": {
"dataLimitValid": false,
"usageHighWarning": 80,
"lastSucceeded": "0",
"billingDay": 1,
"nextBillingDate": "1369627200",
"lastSync": "0",
"billingCycleRemainder": 27,
"billingCycleLimit": 0,
"dataTransferred": 42484315,
"dataTransferredRoaming": 0,
"lastReset": "1366948800",
"end": ""
}
},
"netManualNoCvg": false,
"connection": "Connected",
"connectionType": "IPv4AndIPv6",
"currentPSserviceType": "LTE",
"ca": { "end": "" },
"connectionText": "4G",
"sessDuration": 4282,
"sessStartTime": 1367252934,
"dataTransferred": { "totalb": "345036", "rxb": "184700", "txb": "160336" },
"signalStrength": {
"rssi": 0,
"rscp": 0,
"ecio": 0,
"rsrp": -113,
"rsrq": -20,
"bars": 2,
"sinr": 0,
"end": ""
}
},
"wwanadv": {
"curBand": "LTE B4",
"radioQuality": 52,
"country": "USA",
"RAC": 0,
"LAC": 12345,
"MCC": "123",
"MNC": "456",
"MNCFmt": 3,
"cellId": 12345678,
"chanId": 2300,
"primScode": -1,
"plmnSrvErrBitMask": 0,
"chanIdUl": 20300,
"txLevel": 4,
"rxLevel": -113,
"end": ""
},
"ethernet": {
"offload": { "ipv4Addr": "0.0.0.0", "ipv6Addr": "", "end": "" }
},
"wifi": {
"enabled": true,
"maxClientSupported": 0,
"maxClientLimit": 0,
"maxClientCnt": 0,
"channel": 0,
"hiddenSSID": true,
"passPhrase": "",
"RTSthreshold": 0,
"fragThreshold": 0,
"SSID": "",
"clientCount": 0,
"country": "",
"wps": { "supported": "Disabled", "end": "" },
"guest": {
"maxClientCnt": 0,
"enabled": false,
"SSID": "",
"passPhrase": "",
"generatePassphrase": false,
"hiddenSSID": true,
"chan": 0,
"DHCP": { "range": { "end": "" } }
},
"offload": { "end": "" },
"end": ""
},
"router": {
"gatewayIP": "192.168.5.1",
"DMZaddress": "192.168.5.4",
"DMZenabled": false,
"forceSetup": false,
"DHCP": {
"serverEnabled": true,
"DNS1": "1.1.1.1",
"DNS2": "1.1.2.2",
"DNSmode": "Auto",
"USBpcIP": "0.0.0.0",
"leaseTime": 43200,
"range": { "high": "192.168.5.99", "low": "192.168.5.20", "end": "" }
},
"usbMode": "None",
"usbNetworkTethering": true,
"portFwdEnabled": false,
"portFwdList": [{}],
"portFilteringEnabled": false,
"portFilteringMode": "None",
"portFilterWhiteList": [{}],
"portFilterBlackList": [{}],
"hostName": "routerlogin",
"domainName": "net",
"ipPassThroughEnabled": false,
"ipPassThroughSupported": true,
"Ipv6Supported": true,
"UPNPsupported": false,
"UPNPenabled": false,
"clientList": { "list": [{}], "count": 0 },
"end": ""
},
"fota": {
"fwupdater": {
"available": false,
"chkallow": true,
"chkstatus": "Initial",
"dloadProg": 0,
"error": false,
"lastChkDate": 1367200419,
"state": "NoNewFw",
"isPostponable": false,
"statusCode": 200,
"chkTimeLeft": 0,
"dloadSize": 0,
"end": ""
}
},
"failover": {
"mode": "Auto",
"backhaul": "LTE",
"supported": true,
"monitorPeriod": 10,
"wanConnected": false,
"keepaliveEnable": false,
"keepaliveSleep": 15,
"ipv4Targets": [{ "id": "0", "string": "8.8.8.8" }, {}],
"ipv6Targets": [{}],
"end": ""
},
"eventlog": { "level": 0, "end": 0 },
"ui": { "serverDaysLeftHide": false, "promptActivation": true, "end": 0 }
}

View File

@ -0,0 +1,19 @@
"""The tests for Netgear LTE binary sensor platform."""
import pytest
from homeassistant.components.binary_sensor import BinarySensorDeviceClass
from homeassistant.const import ATTR_DEVICE_CLASS, STATE_OFF, STATE_ON
from homeassistant.core import HomeAssistant
@pytest.mark.usefixtures("setup_integration", "entity_registry_enabled_by_default")
async def test_binary_sensors(hass: HomeAssistant) -> None:
"""Test for successfully setting up the Netgear LTE binary sensor platform."""
state = hass.states.get("binary_sensor.netgear_lte_mobile_connected")
assert state.state == STATE_ON
assert state.attributes[ATTR_DEVICE_CLASS] == BinarySensorDeviceClass.CONNECTIVITY
state = hass.states.get("binary_sensor.netgear_lte_wire_connected")
assert state.state == STATE_OFF
assert state.attributes[ATTR_DEVICE_CLASS] == BinarySensorDeviceClass.CONNECTIVITY
state = hass.states.get("binary_sensor.netgear_lte_roaming")
assert state.state == STATE_OFF

View File

@ -0,0 +1,110 @@
"""Test Netgear LTE config flow."""
from unittest.mock import patch
import pytest
from homeassistant import data_entry_flow
from homeassistant.components.netgear_lte.const import DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_USER
from homeassistant.const import CONF_SOURCE
from homeassistant.core import HomeAssistant
from .conftest import CONF_DATA
def _patch_setup():
return patch(
"homeassistant.components.netgear_lte.async_setup_entry", return_value=True
)
async def test_flow_user_form(hass: HomeAssistant, connection: None) -> None:
"""Test that the user set up form is served."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
with _patch_setup():
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Netgear LM1200"
assert result["data"] == CONF_DATA
assert result["context"]["unique_id"] == "FFFFFFFFFFFFF"
@pytest.mark.parametrize("source", (SOURCE_USER, SOURCE_IMPORT))
async def test_flow_already_configured(
hass: HomeAssistant, setup_integration: None, source: str
) -> None:
"""Test config flow aborts when already configured."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: source},
data=CONF_DATA,
)
assert result["type"] == data_entry_flow.FlowResultType.ABORT
assert result["reason"] == "already_configured"
async def test_flow_user_cannot_connect(
hass: HomeAssistant, cannot_connect: None
) -> None:
"""Test connection error."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
data=CONF_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "cannot_connect"
async def test_flow_user_unknown_error(hass: HomeAssistant, unknown: None) -> None:
"""Test unknown error."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_USER},
)
result = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=CONF_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["errors"]["base"] == "unknown"
async def test_flow_import(hass: HomeAssistant, connection: None) -> None:
"""Test import step."""
with _patch_setup():
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_IMPORT},
data=CONF_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["title"] == "Netgear LM1200"
assert result["data"] == CONF_DATA
async def test_flow_import_failure(hass: HomeAssistant, cannot_connect: None) -> None:
"""Test import step failure."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={CONF_SOURCE: SOURCE_IMPORT},
data=CONF_DATA,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "cannot_connect"

View File

@ -0,0 +1,28 @@
"""Test Netgear LTE integration."""
from homeassistant.components.netgear_lte.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from .conftest import CONF_DATA
async def test_setup_unload(hass: HomeAssistant, setup_integration: None) -> None:
"""Test setup and unload."""
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert entry.state == ConfigEntryState.LOADED
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert entry.data == CONF_DATA
await hass.config_entries.async_unload(entry.entry_id)
await hass.async_block_till_done()
assert not hass.data.get(DOMAIN)
async def test_async_setup_entry_not_ready(
hass: HomeAssistant, setup_cannot_connect: None
) -> None:
"""Test that it throws ConfigEntryNotReady when exception occurs during setup."""
entry = hass.config_entries.async_entries(DOMAIN)[0]
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
assert entry.state == ConfigEntryState.SETUP_RETRY

View File

@ -0,0 +1,29 @@
"""The tests for the Netgear LTE notify platform."""
from unittest.mock import patch
from homeassistant.components.notify import (
ATTR_MESSAGE,
ATTR_TARGET,
DOMAIN as NOTIFY_DOMAIN,
)
from homeassistant.core import HomeAssistant
ICON_PATH = "/some/path"
MESSAGE = "one, two, testing, testing"
async def test_notify(hass: HomeAssistant, setup_integration: None) -> None:
"""Test sending a message."""
assert hass.services.has_service(NOTIFY_DOMAIN, "netgear_lm1200")
with patch("homeassistant.components.netgear_lte.eternalegypt.Modem.sms") as mock:
await hass.services.async_call(
NOTIFY_DOMAIN,
"netgear_lm1200",
{
ATTR_MESSAGE: MESSAGE,
ATTR_TARGET: "5555555556",
},
blocking=True,
)
assert len(mock.mock_calls) == 1

View File

@ -0,0 +1,56 @@
"""The tests for Netgear LTE sensor platform."""
import pytest
from homeassistant.components.sensor import SensorDeviceClass
from homeassistant.const import (
ATTR_DEVICE_CLASS,
ATTR_UNIT_OF_MEASUREMENT,
PERCENTAGE,
SIGNAL_STRENGTH_DECIBELS_MILLIWATT,
UnitOfInformation,
)
from homeassistant.core import HomeAssistant
@pytest.mark.usefixtures("setup_integration", "entity_registry_enabled_by_default")
async def test_sensors(hass: HomeAssistant) -> None:
"""Test for successfully setting up the Netgear LTE sensor platform."""
state = hass.states.get("sensor.netgear_lte_cell_id")
assert state.state == "12345678"
state = hass.states.get("sensor.netgear_lte_connection_text")
assert state.state == "4G"
state = hass.states.get("sensor.netgear_lte_connection_type")
assert state.state == "IPv4AndIPv6"
state = hass.states.get("sensor.netgear_lte_current_band")
assert state.state == "LTE B4"
state = hass.states.get("sensor.netgear_lte_current_ps_service_type")
assert state.state == "LTE"
state = hass.states.get("sensor.netgear_lte_radio_quality")
assert state.state == "52"
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == PERCENTAGE
state = hass.states.get("sensor.netgear_lte_register_network_display")
assert state.state == "T-Mobile"
state = hass.states.get("sensor.netgear_lte_rx_level")
assert state.state == "-113"
assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== SIGNAL_STRENGTH_DECIBELS_MILLIWATT
)
state = hass.states.get("sensor.netgear_lte_sms")
assert state.state == "1"
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == "unread"
state = hass.states.get("sensor.netgear_lte_sms_total")
assert state.state == "1"
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == "messages"
state = hass.states.get("sensor.netgear_lte_tx_level")
assert state.state == "4"
assert (
state.attributes.get(ATTR_UNIT_OF_MEASUREMENT)
== SIGNAL_STRENGTH_DECIBELS_MILLIWATT
)
state = hass.states.get("sensor.netgear_lte_upstream")
assert state.state == "LTE"
state = hass.states.get("sensor.netgear_lte_usage")
assert state.state == "40.5"
assert state.attributes.get(ATTR_UNIT_OF_MEASUREMENT) == UnitOfInformation.MEBIBYTES
assert state.attributes[ATTR_DEVICE_CLASS] == SensorDeviceClass.DATA_SIZE

View File

@ -0,0 +1,55 @@
"""Services tests for the Netgear LTE integration."""
from unittest.mock import patch
from homeassistant.components.netgear_lte.const import DOMAIN
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from .conftest import HOST
async def test_set_option(hass: HomeAssistant, setup_integration: None) -> None:
"""Test service call set option."""
with patch(
"homeassistant.components.netgear_lte.eternalegypt.Modem.set_failover_mode"
) as mock_client:
await hass.services.async_call(
DOMAIN,
"set_option",
{CONF_HOST: HOST, "failover": "auto", "autoconnect": "home"},
blocking=True,
)
assert len(mock_client.mock_calls) == 1
with patch(
"homeassistant.components.netgear_lte.eternalegypt.Modem.connect_lte"
) as mock_client:
await hass.services.async_call(
DOMAIN,
"connect_lte",
{CONF_HOST: HOST},
blocking=True,
)
assert len(mock_client.mock_calls) == 1
with patch(
"homeassistant.components.netgear_lte.eternalegypt.Modem.disconnect_lte"
) as mock_client:
await hass.services.async_call(
DOMAIN,
"disconnect_lte",
{CONF_HOST: HOST},
blocking=True,
)
assert len(mock_client.mock_calls) == 1
with patch(
"homeassistant.components.netgear_lte.eternalegypt.Modem.delete_sms"
) as mock_client:
await hass.services.async_call(
DOMAIN,
"delete_sms",
{CONF_HOST: HOST, "sms_id": 1},
blocking=True,
)
assert len(mock_client.mock_calls) == 1