Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implement stable unique id for Huawei LTE, requires credentials on setup (#49878)

pull/52899/head
Ville Skyttä 2021-07-12 07:25:00 +03:00 committed by GitHub
parent e652ef51a1
commit 91a2b96da0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 231 additions and 162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

154
homeassistant/components/huawei_lte/__init__.py
View File

@ -4,15 +4,11 @@ from __future__ import annotations
from collections import defaultdict
from contextlib import suppress
from datetime import timedelta
from functools import partial
import ipaddress
import logging
import time
from typing import Any, Callable, cast
from urllib.parse import urlparse
import attr
from getmac import get_mac_address
from huawei_lte_api.AuthorizedConnection import AuthorizedConnection
from huawei_lte_api.Client import Client
from huawei_lte_api.Connection import Connection
@ -34,6 +30,7 @@ from homeassistant.components.sensor import DOMAIN as SENSOR_DOMAIN
from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
from homeassistant.const import (
CONF_MAC,
CONF_NAME,
CONF_PASSWORD,
CONF_RECIPIENT,
@ -41,12 +38,13 @@ from homeassistant.const import (
CONF_USERNAME,
EVENT_HOMEASSISTANT_STOP,
)
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, ServiceCall
from homeassistant.core import HomeAssistant, ServiceCall
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import (
config_validation as cv,
device_registry as dr,
discovery,
entity_registry,
)
from homeassistant.helpers.dispatcher import async_dispatcher_connect, dispatcher_send
from homeassistant.helpers.entity import DeviceInfo, Entity
@ -56,6 +54,8 @@ from homeassistant.helpers.typing import ConfigType
from .const import (
ADMIN_SERVICES,
ALL_KEYS,
ATTR_UNIQUE_ID,
CONF_UNAUTHENTICATED_MODE,
CONNECTION_TIMEOUT,
DEFAULT_DEVICE_NAME,
DEFAULT_NOTIFY_SERVICE_NAME,
@ -81,6 +81,7 @@ from .const import (
SERVICE_SUSPEND_INTEGRATION,
UPDATE_SIGNAL,
)
from .utils import get_device_macs
_LOGGER = logging.getLogger(__name__)
@ -131,11 +132,10 @@ CONFIG_ENTRY_PLATFORMS = (
class Router:
"""Class for router state."""
hass: HomeAssistant = attr.ib()
config_entry: ConfigEntry = attr.ib()
connection: Connection = attr.ib()
url: str = attr.ib()
mac: str = attr.ib()
signal_update: CALLBACK_TYPE = attr.ib()
data: dict[str, Any] = attr.ib(init=False, factory=dict)
subscriptions: dict[str, set[str]] = attr.ib(
@ -165,15 +165,15 @@ class Router:
@property
def device_identifiers(self) -> set[tuple[str, str]]:
"""Get router identifiers for device registry."""
try:
return {(DOMAIN, self.data[KEY_DEVICE_INFORMATION]["SerialNumber"])}
except (KeyError, TypeError):
return set()
assert self.config_entry.unique_id is not None
return {(DOMAIN, self.config_entry.unique_id)}
@property
def device_connections(self) -> set[tuple[str, str]]:
"""Get router connections for device registry."""
return {(dr.CONNECTION_NETWORK_MAC, self.mac)} if self.mac else set()
return {
(dr.CONNECTION_NETWORK_MAC, x) for x in self.config_entry.data[CONF_MAC]
}
def _get_data(self, key: str, func: Callable[[], Any]) -> None:
if not self.subscriptions.get(key):
@ -271,7 +271,7 @@ class Router:
KEY_WLAN_WIFI_FEATURE_SWITCH, self.client.wlan.wifi_feature_switch
)
self.signal_update()
dispatcher_send(self.hass, UPDATE_SIGNAL, self.config_entry.unique_id)
def logout(self) -> None:
"""Log out router session."""
@ -304,7 +304,9 @@ class HuaweiLteData:
routers: dict[str, Router] = attr.ib(init=False, factory=dict)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry( # noqa: C901
hass: HomeAssistant, entry: ConfigEntry
) -> bool:
"""Set up Huawei LTE component from config entry."""
url = entry.data[CONF_URL]
@ -342,61 +344,92 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
options={**entry.options, **new_options},
)
# Get MAC address for use in unique ids. Being able to use something
# from the API would be nice, but all of that seems to be available only
# through authenticated calls (e.g. device_information.SerialNumber), and
# we want this available and the same when unauthenticated too.
host = urlparse(url).hostname
try:
if ipaddress.ip_address(host).version == 6:
mode = "ip6"
else:
mode = "ip"
except ValueError:
mode = "hostname"
mac = await hass.async_add_executor_job(partial(get_mac_address, **{mode: host}))
def get_connection() -> Connection:
"""
Set up a connection.
Authorized one if username/pass specified (even if empty), unauthorized one otherwise.
"""
username = entry.data.get(CONF_USERNAME)
password = entry.data.get(CONF_PASSWORD)
if username or password:
connection: Connection = AuthorizedConnection(
"""Set up a connection."""
if entry.options.get(CONF_UNAUTHENTICATED_MODE):
_LOGGER.debug("Connecting in unauthenticated mode, reduced feature set")
connection = Connection(url, timeout=CONNECTION_TIMEOUT)
else:
_LOGGER.debug("Connecting in authenticated mode, full feature set")
username = entry.data.get(CONF_USERNAME) or ""
password = entry.data.get(CONF_PASSWORD) or ""
connection = AuthorizedConnection(
url, username=username, password=password, timeout=CONNECTION_TIMEOUT
)
else:
connection = Connection(url, timeout=CONNECTION_TIMEOUT)
return connection
def signal_update() -> None:
"""Signal updates to data."""
dispatcher_send(hass, UPDATE_SIGNAL, url)
try:
connection = await hass.async_add_executor_job(get_connection)
except Timeout as ex:
raise ConfigEntryNotReady from ex
# Set up router and store reference to it
router = Router(entry, connection, url, mac, signal_update)
hass.data[DOMAIN].routers[url] = router
# Set up router
router = Router(hass, entry, connection, url)
# Do initial data update
await hass.async_add_executor_job(router.update)
# Check that we found required information
device_info = router.data.get(KEY_DEVICE_INFORMATION)
if not entry.unique_id:
# Transitional from < 2021.8: update None config entry and entity unique ids
if device_info and (serial_number := device_info.get("SerialNumber")):
hass.config_entries.async_update_entry(entry, unique_id=serial_number)
ent_reg = entity_registry.async_get(hass)
for entity_entry in entity_registry.async_entries_for_config_entry(
ent_reg, entry.entry_id
):
if not entity_entry.unique_id.startswith("None-"):
continue
new_unique_id = (
f"{serial_number}-{entity_entry.unique_id.split('-', 1)[1]}"
)
ent_reg.async_update_entity(
entity_entry.entity_id, new_unique_id=new_unique_id
)
else:
await hass.async_add_executor_job(router.cleanup)
msg = (
"Could not resolve serial number to use as unique id for router at %s"
", setup failed"
)
if not entry.data.get(CONF_PASSWORD):
msg += (
". Try setting up credentials for the router for one startup, "
"unauthenticated mode can be enabled after that in integration "
"settings"
)
_LOGGER.error(msg, url)
return False
# Store reference to router
hass.data[DOMAIN].routers[entry.unique_id] = router
# Clear all subscriptions, enabled entities will push back theirs
router.subscriptions.clear()
# Update device MAC addresses on record. These can change due to toggling between
# authenticated and unauthenticated modes, or likely also when enabling/disabling
# SSIDs in the router config.
try:
wlan_settings = await hass.async_add_executor_job(
router.client.wlan.multi_basic_settings
)
except Exception: # pylint: disable=broad-except
# Assume not supported, or authentication required but in unauthenticated mode
wlan_settings = {}
macs = get_device_macs(device_info or {}, wlan_settings)
# Be careful not to overwrite a previous, more complete set with a partial one
if macs and (not entry.data[CONF_MAC] or (device_info and wlan_settings)):
new_data = dict(entry.data)
new_data[CONF_MAC] = macs
hass.config_entries.async_update_entry(entry, data=new_data)
# Set up device registry
if router.device_identifiers or router.device_connections:
device_data = {}
sw_version = None
if router.data.get(KEY_DEVICE_INFORMATION):
device_info = router.data[KEY_DEVICE_INFORMATION]
if device_info:
sw_version = device_info.get("SoftwareVersion")
if device_info.get("DeviceName"):
device_data["model"] = device_info["DeviceName"]
@ -425,7 +458,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
NOTIFY_DOMAIN,
DOMAIN,
{
CONF_URL: url,
ATTR_UNIQUE_ID: entry.unique_id,
CONF_NAME: entry.options.get(CONF_NAME, DEFAULT_NOTIFY_SERVICE_NAME),
CONF_RECIPIENT: entry.options.get(CONF_RECIPIENT),
},
@ -462,7 +495,7 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
)
# Forget about the router and invoke its cleanup
router = hass.data[DOMAIN].routers.pop(config_entry.data[CONF_URL])
router = hass.data[DOMAIN].routers.pop(config_entry.unique_id)
await hass.async_add_executor_job(router.cleanup)
return True
@ -483,10 +516,17 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
domain_config[url_normalize(router_config.pop(CONF_URL))] = router_config
def service_handler(service: ServiceCall) -> None:
"""Apply a service."""
"""
Apply a service.
We key this using the router URL instead of its unique id / serial number,
because the latter is not available anywhere in the UI.
"""
routers = hass.data[DOMAIN].routers
if url := service.data.get(CONF_URL):
router = routers.get(url)
router = next(
(router for router in routers.values() if router.url == url), None
)
elif not routers:
_LOGGER.error("%s: no routers configured", service.service)
return
@ -496,7 +536,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
_LOGGER.error(
"%s: more than one router configured, must specify one of URLs %s",
service.service,
sorted(routers),
sorted(router.url for router in routers.values()),
)
return
if not router:
@ -560,6 +600,12 @@ async def async_migrate_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
config_entry.version = 2
hass.config_entries.async_update_entry(config_entry, options=options)
_LOGGER.info("Migrated config entry to version %d", config_entry.version)
if config_entry.version == 2:
config_entry.version = 3
data = dict(config_entry.data)
data[CONF_MAC] = []
hass.config_entries.async_update_entry(config_entry, data=data)
_LOGGER.info("Migrated config entry to version %d", config_entry.version)
return True
@ -584,7 +630,7 @@ class HuaweiLteBaseEntity(Entity):
@property
def unique_id(self) -> str:
"""Return unique ID for entity."""
return f"{self.router.mac}-{self._device_unique_id}"
return f"{self.router.config_entry.unique_id}-{self._device_unique_id}"
@property
def name(self) -> str:

3
homeassistant/components/huawei_lte/binary_sensor.py
View File

@ -12,7 +12,6 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -34,7 +33,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up from config entry."""
router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]]
router = hass.data[DOMAIN].routers[config_entry.unique_id]
entities: list[Entity] = []
if router.data.get(KEY_MONITORING_STATUS):

141
homeassistant/components/huawei_lte/config_flow.py
View File

@ -7,7 +7,7 @@ from urllib.parse import urlparse
from huawei_lte_api.AuthorizedConnection import AuthorizedConnection
from huawei_lte_api.Client import Client
from huawei_lte_api.Connection import Connection
from huawei_lte_api.Connection import GetResponseType
from huawei_lte_api.exceptions import (
LoginErrorPasswordWrongException,
LoginErrorUsernamePasswordOverrunException,
@ -22,6 +22,7 @@ import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.const import (
CONF_MAC,
CONF_NAME,
CONF_PASSWORD,
CONF_RECIPIENT,
@ -34,12 +35,15 @@ from homeassistant.helpers.typing import DiscoveryInfoType
from .const import (
CONF_TRACK_WIRED_CLIENTS,
CONF_UNAUTHENTICATED_MODE,
CONNECTION_TIMEOUT,
DEFAULT_DEVICE_NAME,
DEFAULT_NOTIFY_SERVICE_NAME,
DEFAULT_TRACK_WIRED_CLIENTS,
DEFAULT_UNAUTHENTICATED_MODE,
DOMAIN,
)
from .utils import get_device_macs
_LOGGER = logging.getLogger(__name__)
@ -47,7 +51,7 @@ _LOGGER = logging.getLogger(__name__)
class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle Huawei LTE config flow."""
VERSION = 2
VERSION = 3
@staticmethod
@callback
@ -76,10 +80,10 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
),
): str,
vol.Optional(
CONF_USERNAME, default=user_input.get(CONF_USERNAME, "")
CONF_USERNAME, default=user_input.get(CONF_USERNAME) or ""
): str,
vol.Optional(
CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "")
CONF_PASSWORD, default=user_input.get(CONF_PASSWORD) or ""
): str,
}
),
@ -92,15 +96,7 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle import initiated config flow."""
return await self.async_step_user(user_input)
def _already_configured(self, user_input: dict[str, Any]) -> bool:
"""See if we already have a router matching user input configured."""
existing_urls = {
url_normalize(entry.data[CONF_URL], default_scheme="http")
for entry in self._async_current_entries()
}
return user_input[CONF_URL] in existing_urls
async def async_step_user( # noqa: C901
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle user initiated config flow."""
@ -119,68 +115,46 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
user_input=user_input, errors=errors
)
if self._already_configured(user_input):
return self.async_abort(reason="already_configured")
conn: Connection | None = None
conn: AuthorizedConnection
def logout() -> None:
if isinstance(conn, AuthorizedConnection):
try:
conn.user.logout()
except Exception: # pylint: disable=broad-except
_LOGGER.debug("Could not logout", exc_info=True)
try:
conn.user.logout()
except Exception: # pylint: disable=broad-except
_LOGGER.debug("Could not logout", exc_info=True)
def try_connect(user_input: dict[str, Any]) -> Connection:
def try_connect(user_input: dict[str, Any]) -> AuthorizedConnection:
"""Try connecting with given credentials."""
username = user_input.get(CONF_USERNAME)
password = user_input.get(CONF_PASSWORD)
conn: Connection
if username or password:
conn = AuthorizedConnection(
user_input[CONF_URL],
username=username,
password=password,
timeout=CONNECTION_TIMEOUT,
)
else:
try:
conn = AuthorizedConnection(
user_input[CONF_URL],
username="",
password="",
timeout=CONNECTION_TIMEOUT,
)
user_input[CONF_USERNAME] = ""
user_input[CONF_PASSWORD] = ""
except ResponseErrorException:
_LOGGER.debug(
"Could not login with empty credentials, proceeding unauthenticated",
exc_info=True,
)
conn = Connection(user_input[CONF_URL], timeout=CONNECTION_TIMEOUT)
del user_input[CONF_USERNAME]
del user_input[CONF_PASSWORD]
username = user_input.get(CONF_USERNAME) or ""
password = user_input.get(CONF_PASSWORD) or ""
conn = AuthorizedConnection(
user_input[CONF_URL],
username=username,
password=password,
timeout=CONNECTION_TIMEOUT,
)
return conn
def get_router_title(conn: Connection) -> str:
"""Get title for router."""
title = None
def get_device_info() -> tuple[GetResponseType, GetResponseType]:
"""Get router info."""
client = Client(conn)
try:
info = client.device.basic_information()
device_info = client.device.information()
except Exception: # pylint: disable=broad-except
_LOGGER.debug("Could not get device.basic_information", exc_info=True)
else:
title = info.get("devicename")
if not title:
_LOGGER.debug("Could not get device.information", exc_info=True)
try:
info = client.device.information()
device_info = client.device.basic_information()
except Exception: # pylint: disable=broad-except
_LOGGER.debug("Could not get device.information", exc_info=True)
else:
title = info.get("DeviceName")
return title or DEFAULT_DEVICE_NAME
_LOGGER.debug(
"Could not get device.basic_information", exc_info=True
)
device_info = {}
try:
wlan_settings = client.wlan.multi_basic_settings()
except Exception: # pylint: disable=broad-except
_LOGGER.debug("Could not get wlan.multi_basic_settings", exc_info=True)
wlan_settings = {}
return device_info, wlan_settings
try:
conn = await self.hass.async_add_executor_job(try_connect, user_input)
@ -207,11 +181,25 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
user_input=user_input, errors=errors
)
title = self.context.get("title_placeholders", {}).get(
CONF_NAME
) or await self.hass.async_add_executor_job(get_router_title, conn)
info, wlan_settings = await self.hass.async_add_executor_job(get_device_info)
await self.hass.async_add_executor_job(logout)
if not self.unique_id:
if serial_number := info.get("SerialNumber"):
await self.async_set_unique_id(serial_number)
self._abort_if_unique_id_configured()
else:
await self._async_handle_discovery_without_unique_id()
user_input[CONF_MAC] = get_device_macs(info, wlan_settings)
title = (
self.context.get("title_placeholders", {}).get(CONF_NAME)
or info.get("DeviceName") # device.information
or info.get("devicename") # device.basic_information
or DEFAULT_DEVICE_NAME
)
return self.async_create_entry(title=title, data=user_input)
async def async_step_ssdp(self, discovery_info: DiscoveryInfoType) -> FlowResult:
@ -224,21 +212,20 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if "mobile" not in discovery_info.get(ssdp.ATTR_UPNP_FRIENDLY_NAME, "").lower():
return self.async_abort(reason="not_huawei_lte")
url = self.context[CONF_URL] = url_normalize(
url = url_normalize(
discovery_info.get(
ssdp.ATTR_UPNP_PRESENTATION_URL,
f"http://{urlparse(discovery_info[ssdp.ATTR_SSDP_LOCATION]).hostname}/",
)
)
if any(
url == flow["context"].get(CONF_URL) for flow in self._async_in_progress()
):
return self.async_abort(reason="already_in_progress")
if serial_number := discovery_info.get(ssdp.ATTR_UPNP_SERIAL):
await self.async_set_unique_id(serial_number)
self._abort_if_unique_id_configured()
else:
await self._async_handle_discovery_without_unique_id()
user_input = {CONF_URL: url}
if self._already_configured(user_input):
return self.async_abort(reason="already_configured")
self.context["title_placeholders"] = {
CONF_NAME: discovery_info.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
@ -289,6 +276,12 @@ class OptionsFlowHandler(config_entries.OptionsFlow):
CONF_TRACK_WIRED_CLIENTS, DEFAULT_TRACK_WIRED_CLIENTS
),
): bool,
vol.Optional(
CONF_UNAUTHENTICATED_MODE,
default=self.config_entry.options.get(
CONF_UNAUTHENTICATED_MODE, DEFAULT_UNAUTHENTICATED_MODE
),
): bool,
}
)
return self.async_show_form(step_id="init", data_schema=data_schema)

4
homeassistant/components/huawei_lte/const.py
View File

@ -2,11 +2,15 @@
DOMAIN = "huawei_lte"
ATTR_UNIQUE_ID = "unique_id"
CONF_TRACK_WIRED_CLIENTS = "track_wired_clients"
CONF_UNAUTHENTICATED_MODE = "unauthenticated_mode"
DEFAULT_DEVICE_NAME = "LTE"
DEFAULT_NOTIFY_SERVICE_NAME = DOMAIN
DEFAULT_TRACK_WIRED_CLIENTS = True
DEFAULT_UNAUTHENTICATED_MODE = False
UPDATE_SIGNAL = f"{DOMAIN}_update"

15
homeassistant/components/huawei_lte/device_tracker.py
View File

@ -14,7 +14,6 @@ from homeassistant.components.device_tracker.const import (
SOURCE_TYPE_ROUTER,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_URL
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import entity_registry
from homeassistant.helpers.dispatcher import async_dispatcher_connect
@ -61,7 +60,7 @@ async def async_setup_entry(
# Grab hosts list once to examine whether the initial fetch has got some data for
# us, i.e. if wlan host list is supported. Only set up a subscription and proceed
# with adding and tracking entities if it is.
router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]]
router = hass.data[DOMAIN].routers[config_entry.unique_id]
if (hosts := _get_hosts(router, True)) is None:
return
@ -94,10 +93,10 @@ async def async_setup_entry(
router.subscriptions[KEY_LAN_HOST_INFO].add(_DEVICE_SCAN)
router.subscriptions[KEY_WLAN_HOST_LIST].add(_DEVICE_SCAN)
async def _async_maybe_add_new_entities(url: str) -> None:
async def _async_maybe_add_new_entities(unique_id: str) -> None:
"""Add new entities if the update signal comes from our router."""
if url == router.url:
async_add_new_entities(hass, url, async_add_entities, tracked)
if config_entry.unique_id == unique_id:
async_add_new_entities(router, async_add_entities, tracked)
# Register to handle router data updates
disconnect_dispatcher = async_dispatcher_connect(
@ -106,7 +105,7 @@ async def async_setup_entry(
config_entry.async_on_unload(disconnect_dispatcher)
# Add new entities from initial scan
async_add_new_entities(hass, router.url, async_add_entities, tracked)
async_add_new_entities(router, async_add_entities, tracked)
def _is_wireless(host: _HostType) -> bool:
@ -129,13 +128,11 @@ def _is_us(host: _HostType) -> bool:
@callback
def async_add_new_entities(
hass: HomeAssistant,
router_url: str,
router: Router,
async_add_entities: AddEntitiesCallback,
tracked: set[str],
) -> None:
"""Add new entities that are not already being tracked."""
router = hass.data[DOMAIN].routers[router_url]
hosts = _get_hosts(router)
if not hosts:
return

1
homeassistant/components/huawei_lte/manifest.json
View File

@ -4,7 +4,6 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/huawei_lte",
"requirements": [
"getmac==0.8.2",
"huawei-lte-api==1.4.18",
"stringcase==1.2.0",
"url-normalize==1.4.1"

6
homeassistant/components/huawei_lte/notify.py
View File

@ -9,11 +9,11 @@ import attr
from huawei_lte_api.exceptions import ResponseErrorException
from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService
from homeassistant.const import CONF_RECIPIENT, CONF_URL
from homeassistant.const import CONF_RECIPIENT
from homeassistant.core import HomeAssistant
from . import Router
from .const import DOMAIN
from .const import ATTR_UNIQUE_ID, DOMAIN
_LOGGER = logging.getLogger(__name__)
@ -27,7 +27,7 @@ async def async_get_service(
if discovery_info is None:
return None
router = hass.data[DOMAIN].routers[discovery_info[CONF_URL]]
router = hass.data[DOMAIN].routers[discovery_info[ATTR_UNIQUE_ID]]
default_targets = discovery_info[CONF_RECIPIENT] or []
return HuaweiLteSmsNotificationService(router, default_targets)

3
homeassistant/components/huawei_lte/sensor.py
View File

@ -16,7 +16,6 @@ from homeassistant.components.sensor import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_URL,
DATA_BYTES,
DATA_RATE_BYTES_PER_SECOND,
PERCENTAGE,
@ -360,7 +359,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up from config entry."""
router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]]
router = hass.data[DOMAIN].routers[config_entry.unique_id]
sensors: list[Entity] = []
for key in SENSOR_KEYS:
if not (items := router.data.get(key)):

7
homeassistant/components/huawei_lte/strings.json
View File

@ -1,8 +1,6 @@
{
"config": {
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"already_in_progress": "[%key:common::config_flow::abort::already_in_progress%]",
"not_huawei_lte": "Not a Huawei LTE device"
},
"error": {
@ -23,7 +21,7 @@
"url": "[%key:common::config_flow::data::url%]",
"username": "[%key:common::config_flow::data::username%]"
},
"description": "Enter device access details. Specifying username and password is optional, but enables support for more integration features. On the other hand, use of an authorized connection may cause problems accessing the device web interface from outside Home Assistant while the integration is active, and the other way around.",
"description": "Enter device access details.",
"title": "Configure Huawei LTE"
}
}
@ -34,7 +32,8 @@
"data": {
"name": "Notification service name (change requires restart)",
"recipient": "SMS notification recipients",
"track_wired_clients": "Track wired network clients"
"track_wired_clients": "Track wired network clients",
"unauthenticated_mode": "Unauthenticated mode (change requires reload)"
}
}
}

3
homeassistant/components/huawei_lte/switch.py
View File

@ -12,7 +12,6 @@ from homeassistant.components.switch import (
SwitchEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_URL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
@ -29,7 +28,7 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up from config entry."""
router = hass.data[DOMAIN].routers[config_entry.data[CONF_URL]]
router = hass.data[DOMAIN].routers[config_entry.unique_id]
switches: list[Entity] = []
if router.data.get(KEY_DIALUP_MOBILE_DATASWITCH):

23
homeassistant/components/huawei_lte/utils.py Normal file
View File

@ -0,0 +1,23 @@
"""Utilities for the Huawei LTE integration."""
from __future__ import annotations
from huawei_lte_api.Connection import GetResponseType
from homeassistant.helpers.device_registry import format_mac
def get_device_macs(
device_info: GetResponseType, wlan_settings: GetResponseType
) -> list[str]:
"""Get list of device MAC addresses.
:param device_info: the device.information structure for the device
:param wlan_settings: the wlan.multi_basic_settings structure for the device
"""
macs = [device_info.get("MacAddress1"), device_info.get("MacAddress2")]
try:
macs.extend(x.get("WifiMac") for x in wlan_settings["Ssids"]["Ssid"])
except Exception: # pylint: disable=broad-except
# Assume not supported
pass
return sorted({format_mac(str(x)) for x in macs if x})

1
requirements_all.txt
View File

@ -670,7 +670,6 @@ georss_ign_sismologia_client==0.3
# homeassistant.components.qld_bushfire
georss_qld_bushfire_alert_client==0.5
# homeassistant.components.huawei_lte
# homeassistant.components.kef
# homeassistant.components.minecraft_server
# homeassistant.components.nmap_tracker

1
requirements_test_all.txt
View File

@ -376,7 +376,6 @@ georss_ign_sismologia_client==0.3
# homeassistant.components.qld_bushfire
georss_qld_bushfire_alert_client==0.5
# homeassistant.components.huawei_lte
# homeassistant.components.kef
# homeassistant.components.minecraft_server
# homeassistant.components.nmap_tracker

31
tests/components/huawei_lte/test_config_flow.py
View File

@ -10,7 +10,7 @@ from requests_mock import ANY
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import ssdp
from homeassistant.components.huawei_lte.const import DOMAIN
from homeassistant.components.huawei_lte.const import CONF_UNAUTHENTICATED_MODE, DOMAIN
from homeassistant.const import (
CONF_NAME,
CONF_PASSWORD,
@ -21,6 +21,8 @@ from homeassistant.const import (
from tests.common import MockConfigEntry
FIXTURE_UNIQUE_ID = "SERIALNUMBER"
FIXTURE_USER_INPUT = {
CONF_URL: "http://192.168.1.1/",
CONF_USERNAME: "admin",
@ -57,20 +59,30 @@ async def test_urlize_plain_host(hass, requests_mock):
assert user_input[CONF_URL] == f"http://{host}/"
async def test_already_configured(hass):
async def test_already_configured(hass, requests_mock, login_requests_mock):
"""Test we reject already configured devices."""
MockConfigEntry(
domain=DOMAIN, data=FIXTURE_USER_INPUT, title="Already configured"
domain=DOMAIN,
unique_id=FIXTURE_UNIQUE_ID,
data=FIXTURE_USER_INPUT,
title="Already configured",
).add_to_hass(hass)
login_requests_mock.request(
ANY,
f"{FIXTURE_USER_INPUT[CONF_URL]}api/user/login",
text="<response>OK</response>",
)
requests_mock.request(
ANY,
f"{FIXTURE_USER_INPUT[CONF_URL]}api/device/information",
text=f"<response><SerialNumber>{FIXTURE_UNIQUE_ID}</SerialNumber></response>",
)
result = await hass.config_entries.flow.async_init(
DOMAIN,
context={"source": config_entries.SOURCE_USER},
data={
**FIXTURE_USER_INPUT,
# Tweak URL a bit to check that doesn't fail duplicate detection
CONF_URL: FIXTURE_USER_INPUT[CONF_URL].replace("http", "HTTP"),
},
data=FIXTURE_USER_INPUT,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
@ -182,7 +194,7 @@ async def test_ssdp(hass):
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert context[CONF_URL] == url
assert result["data_schema"]({})[CONF_URL] == url
async def test_options(hass):
@ -203,3 +215,4 @@ async def test_options(hass):
)
assert result["data"][CONF_NAME] == DOMAIN
assert result["data"][CONF_RECIPIENT] == [recipient]
assert result["data"][CONF_UNAUTHENTICATED_MODE] is False