mirror of
https://github.com/home-assistant/core.git
synced 2025-04-19 14:57:52 +00:00
Netgear config flow (#54479)
* Original work from Quentame * Small adjustments * Add properties and method_version * fix unknown name * add consider_home functionality * fix typo * fix key * swao setup order * use formatted mac * add tracked_list option * add options flow * add config flow * add config flow * clean up registries * only remove if no other integration has that device * tracked_list formatting * convert tracked list * add import * move imports * use new tracked list on update * use update_device instead of remove * add strings * initialize already known devices * Update router.py * Update router.py * Update router.py * small fixes * styling * fix typing * fix spelling * Update router.py * get model of router * add router device info * fix api * add listeners * update router device info * remove method version option * Update __init__.py * fix styling * ignore typing * remove typing * fix mypy config * Update mypy.ini * add options flow tests * Update .coveragerc * fix styling * Update homeassistant/components/netgear/__init__.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/netgear/__init__.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/netgear/__init__.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/netgear/config_flow.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/netgear/router.py Co-authored-by: J. Nick Koston <nick@koston.org> * add ConfigEntryNotReady * Update router.py * use entry.async_on_unload * Update homeassistant/components/netgear/device_tracker.py Co-authored-by: J. Nick Koston <nick@koston.org> * use cv.ensure_list_csv * add hostname property * Update device_tracker.py * fix typo * fix isort * add myself to codeowners * clean config flow * further clean config flow * deprecate old netgear discovery * split out _async_remove_untracked_registries * Update homeassistant/components/netgear/config_flow.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/netgear/config_flow.py Co-authored-by: J. Nick Koston <nick@koston.org> * cleanup * fix rename * fix typo * remove URL option * fixes * add sensor platform * fixes * fix removing multiple entities * remove extra attributes * initialize sensors correctly * extra sensors disabled by default * fix styling and unused imports * fix tests * Update .coveragerc * fix requirements * remove tracked list * remove tracked registry editing * fix styling * fix discovery test * simplify unload * Update homeassistant/components/netgear/router.py Co-authored-by: J. Nick Koston <nick@koston.org> * add typing Co-authored-by: J. Nick Koston <nick@koston.org> * add typing Co-authored-by: J. Nick Koston <nick@koston.org> * add typing Co-authored-by: J. Nick Koston <nick@koston.org> * condense NetgearSensorEntities Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/netgear/router.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/netgear/router.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/netgear/router.py Co-authored-by: J. Nick Koston <nick@koston.org> * Update homeassistant/components/netgear/router.py Co-authored-by: J. Nick Koston <nick@koston.org> * add typing * styling * add typing * use ForwardRefrence for typing * Update homeassistant/components/netgear/device_tracker.py Co-authored-by: J. Nick Koston <nick@koston.org> * add typing * Apply suggestions from code review Thanks! Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * process review comments * fix styling * fix devicename not available on all models * ensure DeviceName is not needed * Update homeassistant/components/netgear/config_flow.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Update homeassistant/components/netgear/config_flow.py Co-authored-by: Martin Hjelmare <marhje52@gmail.com> * Update __init__.py * fix styling Co-authored-by: J. Nick Koston <nick@koston.org> Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
starkillerOG
GitHub
parent
37f263e2ac
commit
5f86388f1c
@ -699,7 +699,10 @@ omit =
|
||||
homeassistant/components/nello/lock.py
|
||||
homeassistant/components/nest/legacy/*
|
||||
homeassistant/components/netdata/sensor.py
|
||||
homeassistant/components/netgear/__init__.py
|
||||
homeassistant/components/netgear/device_tracker.py
|
||||
homeassistant/components/netgear/router.py
|
||||
homeassistant/components/netgear/sensor.py
|
||||
homeassistant/components/netgear_lte/*
|
||||
homeassistant/components/netio/switch.py
|
||||
homeassistant/components/neurio_energy/sensor.py
|
||||
|
||||
@ -335,6 +335,7 @@ homeassistant/components/ness_alarm/* @nickw444
|
||||
homeassistant/components/nest/* @allenporter
|
||||
homeassistant/components/netatmo/* @cgtobi
|
||||
homeassistant/components/netdata/* @fabaff
|
||||
homeassistant/components/netgear/* @hacf-fr @Quentame @starkillerOG
|
||||
homeassistant/components/nexia/* @bdraco
|
||||
homeassistant/components/nextbus/* @vividboarder
|
||||
homeassistant/components/nextcloud/* @meichthys
|
||||
|
||||
@ -46,7 +46,6 @@ CONFIG_ENTRY_HANDLERS = {
|
||||
|
||||
# These have no config flows
|
||||
SERVICE_HANDLERS = {
|
||||
SERVICE_NETGEAR: ("device_tracker", None),
|
||||
SERVICE_ENIGMA2: ("media_player", "enigma2"),
|
||||
SERVICE_SABNZBD: ("sabnzbd", None),
|
||||
"yamaha": ("media_player", "yamaha"),
|
||||
@ -76,6 +75,7 @@ MIGRATED_SERVICE_HANDLERS = [
|
||||
"kodi",
|
||||
SERVICE_KONNECTED,
|
||||
SERVICE_MOBILE_APP,
|
||||
SERVICE_NETGEAR,
|
||||
SERVICE_OCTOPRINT,
|
||||
"philips_hue",
|
||||
SERVICE_SAMSUNG_PRINTER,
|
||||
|
||||
@ -1 +1,59 @@
|
||||
"""The netgear component."""
|
||||
"""Support for Netgear routers."""
|
||||
import logging
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.exceptions import ConfigEntryNotReady
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from .const import DOMAIN, PLATFORMS
|
||||
from .errors import CannotLoginException
|
||||
from .router import NetgearRouter
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
|
||||
"""Set up Netgear component."""
|
||||
router = NetgearRouter(hass, entry)
|
||||
try:
|
||||
await router.async_setup()
|
||||
except CannotLoginException as ex:
|
||||
raise ConfigEntryNotReady from ex
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
hass.data[DOMAIN][entry.unique_id] = router
|
||||
|
||||
entry.async_on_unload(entry.add_update_listener(update_listener))
|
||||
|
||||
device_registry = await dr.async_get_registry(hass)
|
||||
device_registry.async_get_or_create(
|
||||
config_entry_id=entry.entry_id,
|
||||
identifiers={(DOMAIN, entry.unique_id)},
|
||||
manufacturer="Netgear",
|
||||
name=router.device_name,
|
||||
model=router.model,
|
||||
sw_version=router.firmware_version,
|
||||
)
|
||||
|
||||
hass.config_entries.async_setup_platforms(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistantType, entry: ConfigEntry) -> bool:
|
||||
"""Unload a config entry."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
|
||||
if unload_ok:
|
||||
hass.data[DOMAIN].pop(entry.unique_id)
|
||||
if not hass.data[DOMAIN]:
|
||||
hass.data.pop(DOMAIN)
|
||||
|
||||
return unload_ok
|
||||
|
||||
|
||||
async def update_listener(hass: HomeAssistant, config_entry: ConfigEntry) -> None:
|
||||
"""Handle options update."""
|
||||
await hass.config_entries.async_reload(config_entry.entry_id)
|
||||
|
||||
184
homeassistant/components/netgear/config_flow.py
Normal file
184
homeassistant/components/netgear/config_flow.py
Normal file
@ -0,0 +1,184 @@
|
||||
"""Config flow to configure the Netgear integration."""
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from pynetgear import DEFAULT_HOST, DEFAULT_PORT, DEFAULT_USER
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components import ssdp
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_SSL,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from .const import CONF_CONSIDER_HOME, DEFAULT_CONSIDER_HOME, DEFAULT_NAME, DOMAIN
|
||||
from .errors import CannotLoginException
|
||||
from .router import get_api
|
||||
|
||||
|
||||
def _discovery_schema_with_defaults(discovery_info):
|
||||
return vol.Schema(_ordered_shared_schema(discovery_info))
|
||||
|
||||
|
||||
def _user_schema_with_defaults(user_input):
|
||||
user_schema = {
|
||||
vol.Optional(CONF_HOST, default=user_input.get(CONF_HOST, "")): str,
|
||||
vol.Optional(CONF_PORT, default=user_input.get(CONF_PORT, DEFAULT_PORT)): int,
|
||||
vol.Optional(CONF_SSL, default=user_input.get(CONF_SSL, False)): bool,
|
||||
}
|
||||
user_schema.update(_ordered_shared_schema(user_input))
|
||||
|
||||
return vol.Schema(user_schema)
|
||||
|
||||
|
||||
def _ordered_shared_schema(schema_input):
|
||||
return {
|
||||
vol.Optional(CONF_USERNAME, default=schema_input.get(CONF_USERNAME, "")): str,
|
||||
vol.Required(CONF_PASSWORD, default=schema_input.get(CONF_PASSWORD, "")): str,
|
||||
}
|
||||
|
||||
|
||||
class OptionsFlowHandler(config_entries.OptionsFlow):
|
||||
"""Options for the component."""
|
||||
|
||||
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
|
||||
"""Init object."""
|
||||
self.config_entry = config_entry
|
||||
|
||||
async def async_step_init(self, user_input=None):
|
||||
"""Manage the options."""
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(title="", data=user_input)
|
||||
|
||||
settings_schema = vol.Schema(
|
||||
{
|
||||
vol.Optional(
|
||||
CONF_CONSIDER_HOME,
|
||||
default=self.config_entry.options.get(
|
||||
CONF_CONSIDER_HOME, DEFAULT_CONSIDER_HOME.total_seconds()
|
||||
),
|
||||
): int,
|
||||
}
|
||||
)
|
||||
|
||||
return self.async_show_form(step_id="init", data_schema=settings_schema)
|
||||
|
||||
|
||||
class NetgearFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow."""
|
||||
|
||||
VERSION = 1
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the netgear config flow."""
|
||||
self.placeholders = {
|
||||
CONF_HOST: DEFAULT_HOST,
|
||||
CONF_PORT: DEFAULT_PORT,
|
||||
CONF_USERNAME: DEFAULT_USER,
|
||||
CONF_SSL: False,
|
||||
}
|
||||
self.discovered = False
|
||||
|
||||
@staticmethod
|
||||
@callback
|
||||
def async_get_options_flow(
|
||||
config_entry: config_entries.ConfigEntry,
|
||||
) -> OptionsFlowHandler:
|
||||
"""Get the options flow."""
|
||||
return OptionsFlowHandler(config_entry)
|
||||
|
||||
async def _show_setup_form(self, user_input=None, errors=None):
|
||||
"""Show the setup form to the user."""
|
||||
if not user_input:
|
||||
user_input = {}
|
||||
|
||||
if self.discovered:
|
||||
data_schema = _discovery_schema_with_defaults(user_input)
|
||||
else:
|
||||
data_schema = _user_schema_with_defaults(user_input)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=data_schema,
|
||||
errors=errors or {},
|
||||
description_placeholders=self.placeholders,
|
||||
)
|
||||
|
||||
async def async_step_import(self, user_input=None):
|
||||
"""Import a config entry."""
|
||||
return await self.async_step_user(user_input)
|
||||
|
||||
async def async_step_ssdp(self, discovery_info: dict) -> FlowResult:
|
||||
"""Initialize flow from ssdp."""
|
||||
updated_data = {}
|
||||
|
||||
device_url = urlparse(discovery_info[ssdp.ATTR_SSDP_LOCATION])
|
||||
if device_url.hostname:
|
||||
updated_data[CONF_HOST] = device_url.hostname
|
||||
if device_url.port:
|
||||
updated_data[CONF_PORT] = device_url.port
|
||||
if device_url.scheme == "https":
|
||||
updated_data[CONF_SSL] = True
|
||||
else:
|
||||
updated_data[CONF_SSL] = False
|
||||
|
||||
await self.async_set_unique_id(discovery_info[ssdp.ATTR_UPNP_SERIAL])
|
||||
self._abort_if_unique_id_configured(updates=updated_data)
|
||||
self.placeholders.update(updated_data)
|
||||
self.discovered = True
|
||||
|
||||
return await self.async_step_user()
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
"""Handle a flow initiated by the user."""
|
||||
errors = {}
|
||||
|
||||
if user_input is None:
|
||||
return await self._show_setup_form()
|
||||
|
||||
host = user_input.get(CONF_HOST, self.placeholders[CONF_HOST])
|
||||
port = user_input.get(CONF_PORT, self.placeholders[CONF_PORT])
|
||||
ssl = user_input.get(CONF_SSL, self.placeholders[CONF_SSL])
|
||||
username = user_input.get(CONF_USERNAME, self.placeholders[CONF_USERNAME])
|
||||
password = user_input[CONF_PASSWORD]
|
||||
if not username:
|
||||
username = self.placeholders[CONF_USERNAME]
|
||||
|
||||
# Open connection and check authentication
|
||||
try:
|
||||
api = await self.hass.async_add_executor_job(
|
||||
get_api, password, host, username, port, ssl
|
||||
)
|
||||
except CannotLoginException:
|
||||
errors["base"] = "config"
|
||||
|
||||
if errors:
|
||||
return await self._show_setup_form(user_input, errors)
|
||||
|
||||
# Check if already configured
|
||||
info = await self.hass.async_add_executor_job(api.get_info)
|
||||
await self.async_set_unique_id(info["SerialNumber"], raise_on_progress=False)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
config_data = {
|
||||
CONF_USERNAME: username,
|
||||
CONF_PASSWORD: password,
|
||||
CONF_HOST: host,
|
||||
CONF_PORT: port,
|
||||
CONF_SSL: ssl,
|
||||
}
|
||||
|
||||
if info.get("ModelName") is not None and info.get("DeviceName") is not None:
|
||||
name = f"{info['ModelName']} - {info['DeviceName']}"
|
||||
else:
|
||||
name = info.get("ModelName", DEFAULT_NAME)
|
||||
|
||||
return self.async_create_entry(
|
||||
title=name,
|
||||
data=config_data,
|
||||
)
|
||||
60
homeassistant/components/netgear/const.py
Normal file
60
homeassistant/components/netgear/const.py
Normal file
@ -0,0 +1,60 @@
|
||||
"""Netgear component constants."""
|
||||
from datetime import timedelta
|
||||
|
||||
DOMAIN = "netgear"
|
||||
|
||||
PLATFORMS = ["device_tracker", "sensor"]
|
||||
|
||||
CONF_CONSIDER_HOME = "consider_home"
|
||||
|
||||
DEFAULT_CONSIDER_HOME = timedelta(seconds=180)
|
||||
DEFAULT_NAME = "Netgear router"
|
||||
|
||||
# update method V2 models
|
||||
MODELS_V2 = ["Orbi"]
|
||||
|
||||
# Icons
|
||||
DEVICE_ICONS = {
|
||||
0: "mdi:access-point-network", # Router (Orbi ...)
|
||||
1: "mdi:book-open-variant", # Amazon Kindle
|
||||
2: "mdi:android", # Android Device
|
||||
3: "mdi:cellphone-android", # Android Phone
|
||||
4: "mdi:tablet-android", # Android Tablet
|
||||
5: "mdi:router-wireless", # Apple Airport Express
|
||||
6: "mdi:disc-player", # Blu-ray Player
|
||||
7: "mdi:router-network", # Bridge
|
||||
8: "mdi:play-network", # Cable STB
|
||||
9: "mdi:camera", # Camera
|
||||
10: "mdi:router-network", # Router
|
||||
11: "mdi:play-network", # DVR
|
||||
12: "mdi:gamepad-variant", # Gaming Console
|
||||
13: "mdi:desktop-mac", # iMac
|
||||
14: "mdi:tablet-ipad", # iPad
|
||||
15: "mdi:tablet-ipad", # iPad Mini
|
||||
16: "mdi:cellphone-iphone", # iPhone 5/5S/5C
|
||||
17: "mdi:cellphone-iphone", # iPhone
|
||||
18: "mdi:ipod", # iPod Touch
|
||||
19: "mdi:linux", # Linux PC
|
||||
20: "mdi:apple-finder", # Mac Mini
|
||||
21: "mdi:desktop-tower", # Mac Pro
|
||||
22: "mdi:laptop-mac", # MacBook
|
||||
23: "mdi:play-network", # Media Device
|
||||
24: "mdi:network", # Network Device
|
||||
25: "mdi:play-network", # Other STB
|
||||
26: "mdi:power-plug", # Powerline
|
||||
27: "mdi:printer", # Printer
|
||||
28: "mdi:access-point", # Repeater
|
||||
29: "mdi:play-network", # Satellite STB
|
||||
30: "mdi:scanner", # Scanner
|
||||
31: "mdi:play-network", # SlingBox
|
||||
32: "mdi:cellphone", # Smart Phone
|
||||
33: "mdi:nas", # Storage (NAS)
|
||||
34: "mdi:switch", # Switch
|
||||
35: "mdi:television", # TV
|
||||
36: "mdi:tablet", # Tablet
|
||||
37: "mdi:desktop-classic", # UNIX PC
|
||||
38: "mdi:desktop-tower-monitor", # Windows PC
|
||||
39: "mdi:laptop-windows", # Surface
|
||||
40: "mdi:access-point-network", # Wifi Extender
|
||||
41: "mdi:apple-airplay", # Apple TV
|
||||
}
|
||||
@ -1,15 +1,14 @@
|
||||
"""Support for Netgear routers."""
|
||||
import logging
|
||||
from pprint import pformat
|
||||
|
||||
from pynetgear import Netgear
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.device_tracker import (
|
||||
DOMAIN,
|
||||
PLATFORM_SCHEMA as PARENT_PLATFORM_SCHEMA,
|
||||
DeviceScanner,
|
||||
SOURCE_TYPE_ROUTER,
|
||||
)
|
||||
from homeassistant.components.device_tracker.config_entry import ScannerEntity
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_DEVICES,
|
||||
CONF_EXCLUDE,
|
||||
@ -19,7 +18,13 @@ from homeassistant.const import (
|
||||
CONF_SSL,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import callback
|
||||
import homeassistant.helpers.config_validation as cv
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from .const import DEVICE_ICONS, DOMAIN
|
||||
from .router import NetgearDeviceEntity, NetgearRouter, async_setup_netgear_entry
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
@ -27,9 +32,9 @@ CONF_APS = "accesspoints"
|
||||
|
||||
PLATFORM_SCHEMA = PARENT_PLATFORM_SCHEMA.extend(
|
||||
{
|
||||
vol.Optional(CONF_HOST, default=""): cv.string,
|
||||
vol.Optional(CONF_SSL, default=False): cv.boolean,
|
||||
vol.Optional(CONF_USERNAME, default=""): cv.string,
|
||||
vol.Optional(CONF_HOST): cv.string,
|
||||
vol.Optional(CONF_SSL): cv.boolean,
|
||||
vol.Optional(CONF_USERNAME): cv.string,
|
||||
vol.Required(CONF_PASSWORD): cv.string,
|
||||
vol.Optional(CONF_PORT): cv.port,
|
||||
vol.Optional(CONF_DEVICES, default=[]): vol.All(cv.ensure_list, [cv.string]),
|
||||
@ -39,132 +44,88 @@ PLATFORM_SCHEMA = PARENT_PLATFORM_SCHEMA.extend(
|
||||
)
|
||||
|
||||
|
||||
def get_scanner(hass, config):
|
||||
"""Validate the configuration and returns a Netgear scanner."""
|
||||
info = config[DOMAIN]
|
||||
host = info[CONF_HOST]
|
||||
ssl = info[CONF_SSL]
|
||||
username = info[CONF_USERNAME]
|
||||
password = info[CONF_PASSWORD]
|
||||
port = info.get(CONF_PORT)
|
||||
devices = info[CONF_DEVICES]
|
||||
excluded_devices = info[CONF_EXCLUDE]
|
||||
accesspoints = info[CONF_APS]
|
||||
async def async_get_scanner(hass, config):
|
||||
"""Import Netgear configuration from YAML."""
|
||||
hass.async_create_task(
|
||||
hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data=config,
|
||||
)
|
||||
)
|
||||
|
||||
api = Netgear(password, host, username, port, ssl)
|
||||
scanner = NetgearDeviceScanner(api, devices, excluded_devices, accesspoints)
|
||||
_LOGGER.warning(
|
||||
"Your Netgear configuration has been imported into the UI, "
|
||||
"please remove it from configuration.yaml. "
|
||||
"Loading Netgear via platform setup is now deprecated"
|
||||
)
|
||||
|
||||
_LOGGER.debug("Logging in")
|
||||
|
||||
results = scanner.get_attached_devices()
|
||||
|
||||
if results is not None:
|
||||
scanner.last_results = results
|
||||
else:
|
||||
_LOGGER.error("Failed to Login")
|
||||
return None
|
||||
|
||||
return scanner
|
||||
return None
|
||||
|
||||
|
||||
class NetgearDeviceScanner(DeviceScanner):
|
||||
"""Queries a Netgear wireless router using the SOAP-API."""
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up device tracker for Netgear component."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
api,
|
||||
devices,
|
||||
excluded_devices,
|
||||
accesspoints,
|
||||
):
|
||||
"""Initialize the scanner."""
|
||||
self.tracked_devices = devices
|
||||
self.excluded_devices = excluded_devices
|
||||
self.tracked_accesspoints = accesspoints
|
||||
self.last_results = []
|
||||
self._api = api
|
||||
def generate_classes(router: NetgearRouter, device: dict):
|
||||
return [NetgearScannerEntity(router, device)]
|
||||
|
||||
def scan_devices(self):
|
||||
"""Scan for new devices and return a list with found device IDs."""
|
||||
self._update_info()
|
||||
async_setup_netgear_entry(hass, entry, async_add_entities, generate_classes)
|
||||
|
||||
devices = []
|
||||
|
||||
for dev in self.last_results:
|
||||
tracked = (
|
||||
not self.tracked_devices
|
||||
or dev.mac in self.tracked_devices
|
||||
or dev.name in self.tracked_devices
|
||||
)
|
||||
tracked = tracked and (
|
||||
not self.excluded_devices
|
||||
or not (
|
||||
dev.mac in self.excluded_devices
|
||||
or dev.name in self.excluded_devices
|
||||
)
|
||||
)
|
||||
if tracked:
|
||||
devices.append(dev.mac)
|
||||
if (
|
||||
self.tracked_accesspoints
|
||||
and dev.conn_ap_mac in self.tracked_accesspoints
|
||||
):
|
||||
devices.append(f"{dev.mac}_{dev.conn_ap_mac}")
|
||||
class NetgearScannerEntity(NetgearDeviceEntity, ScannerEntity):
|
||||
"""Representation of a device connected to a Netgear router."""
|
||||
|
||||
return devices
|
||||
def __init__(self, router: NetgearRouter, device: dict) -> None:
|
||||
"""Initialize a Netgear device."""
|
||||
super().__init__(router, device)
|
||||
self._hostname = self.get_hostname()
|
||||
self._icon = DEVICE_ICONS.get(device["device_type"], "mdi:help-network")
|
||||
|
||||
def get_device_name(self, device):
|
||||
"""Return the name of the given device or the MAC if we don't know."""
|
||||
parts = device.split("_")
|
||||
mac = parts[0]
|
||||
ap_mac = None
|
||||
if len(parts) > 1:
|
||||
ap_mac = parts[1]
|
||||
def get_hostname(self):
|
||||
"""Return the hostname of the given device or None if we don't know."""
|
||||
hostname = self._device["name"]
|
||||
if hostname == "--":
|
||||
return None
|
||||
|
||||
name = None
|
||||
for dev in self.last_results:
|
||||
if dev.mac == mac:
|
||||
name = dev.name
|
||||
break
|
||||
return hostname
|
||||
|
||||
if not name or name == "--":
|
||||
name = mac
|
||||
@callback
|
||||
def async_update_device(self) -> None:
|
||||
"""Update the Netgear device."""
|
||||
self._device = self._router.devices[self._mac]
|
||||
self._active = self._device["active"]
|
||||
self._icon = DEVICE_ICONS.get(self._device["device_type"], "mdi:help-network")
|
||||
|
||||
if ap_mac:
|
||||
ap_name = "Router"
|
||||
for dev in self.last_results:
|
||||
if dev.mac == ap_mac:
|
||||
ap_name = dev.name
|
||||
break
|
||||
self.async_write_ha_state()
|
||||
|
||||
return f"{name} on {ap_name}"
|
||||
@property
|
||||
def is_connected(self):
|
||||
"""Return true if the device is connected to the router."""
|
||||
return self._active
|
||||
|
||||
return name
|
||||
@property
|
||||
def source_type(self) -> str:
|
||||
"""Return the source type."""
|
||||
return SOURCE_TYPE_ROUTER
|
||||
|
||||
def _update_info(self):
|
||||
"""Retrieve latest information from the Netgear router.
|
||||
@property
|
||||
def ip_address(self) -> str:
|
||||
"""Return the IP address."""
|
||||
return self._device["ip"]
|
||||
|
||||
Returns boolean if scanning successful.
|
||||
"""
|
||||
_LOGGER.debug("Scanning")
|
||||
@property
|
||||
def mac_address(self) -> str:
|
||||
"""Return the mac address."""
|
||||
return self._mac
|
||||
|
||||
results = self.get_attached_devices()
|
||||
@property
|
||||
def hostname(self) -> str:
|
||||
"""Return the hostname."""
|
||||
return self._hostname
|
||||
|
||||
if _LOGGER.isEnabledFor(logging.DEBUG):
|
||||
_LOGGER.debug("Scan result: \n%s", pformat(results))
|
||||
|
||||
if results is None:
|
||||
_LOGGER.warning("Error scanning devices")
|
||||
|
||||
self.last_results = results or []
|
||||
|
||||
def get_attached_devices(self):
|
||||
"""List attached devices with pynetgear.
|
||||
|
||||
The v2 method takes more time and is more heavy on the router
|
||||
so we only use it if we need connected AP info.
|
||||
"""
|
||||
if self.tracked_accesspoints:
|
||||
return self._api.get_attached_devices_2()
|
||||
|
||||
return self._api.get_attached_devices()
|
||||
@property
|
||||
def icon(self) -> str:
|
||||
"""Return the icon."""
|
||||
return self._icon
|
||||
|
||||
10
homeassistant/components/netgear/errors.py
Normal file
10
homeassistant/components/netgear/errors.py
Normal file
@ -0,0 +1,10 @@
|
||||
"""Errors for the Netgear component."""
|
||||
from homeassistant.exceptions import HomeAssistantError
|
||||
|
||||
|
||||
class NetgearException(HomeAssistantError):
|
||||
"""Base class for Netgear exceptions."""
|
||||
|
||||
|
||||
class CannotLoginException(NetgearException):
|
||||
"""Unable to login to the router."""
|
||||
@ -2,7 +2,14 @@
|
||||
"domain": "netgear",
|
||||
"name": "NETGEAR",
|
||||
"documentation": "https://www.home-assistant.io/integrations/netgear",
|
||||
"requirements": ["pynetgear==0.6.1"],
|
||||
"codeowners": [],
|
||||
"iot_class": "local_polling"
|
||||
"requirements": ["pynetgear==0.7.0"],
|
||||
"codeowners": ["@hacf-fr", "@Quentame", "@starkillerOG"],
|
||||
"iot_class": "local_polling",
|
||||
"config_flow": true,
|
||||
"ssdp": [
|
||||
{
|
||||
"manufacturer": "NETGEAR, Inc.",
|
||||
"deviceType": "urn:schemas-upnp-org:device:InternetGatewayDevice:1"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
292
homeassistant/components/netgear/router.py
Normal file
292
homeassistant/components/netgear/router.py
Normal file
@ -0,0 +1,292 @@
|
||||
"""Represent the Netgear router and its devices."""
|
||||
from abc import abstractmethod
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Callable
|
||||
|
||||
from pynetgear import Netgear
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_SSL,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers import device_registry as dr
|
||||
from homeassistant.helpers.device_registry import CONNECTION_NETWORK_MAC, format_mac
|
||||
from homeassistant.helpers.dispatcher import (
|
||||
async_dispatcher_connect,
|
||||
async_dispatcher_send,
|
||||
)
|
||||
from homeassistant.helpers.entity import Entity
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.event import async_track_time_interval
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
from homeassistant.util import dt as dt_util
|
||||
|
||||
from .const import (
|
||||
CONF_CONSIDER_HOME,
|
||||
DEFAULT_CONSIDER_HOME,
|
||||
DEFAULT_NAME,
|
||||
DOMAIN,
|
||||
MODELS_V2,
|
||||
)
|
||||
from .errors import CannotLoginException
|
||||
|
||||
SCAN_INTERVAL = timedelta(seconds=30)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def get_api(
|
||||
password: str,
|
||||
host: str = None,
|
||||
username: str = None,
|
||||
port: int = None,
|
||||
ssl: bool = False,
|
||||
) -> Netgear:
|
||||
"""Get the Netgear API and login to it."""
|
||||
api: Netgear = Netgear(password, host, username, port, ssl)
|
||||
|
||||
if not api.login():
|
||||
raise CannotLoginException
|
||||
|
||||
return api
|
||||
|
||||
|
||||
@callback
|
||||
def async_setup_netgear_entry(
|
||||
hass: HomeAssistantType,
|
||||
entry: ConfigEntry,
|
||||
async_add_entities: AddEntitiesCallback,
|
||||
entity_class_generator: Callable[["NetgearRouter", dict], list],
|
||||
) -> None:
|
||||
"""Set up device tracker for Netgear component."""
|
||||
router = hass.data[DOMAIN][entry.unique_id]
|
||||
tracked = set()
|
||||
|
||||
@callback
|
||||
def _async_router_updated():
|
||||
"""Update the values of the router."""
|
||||
async_add_new_entities(
|
||||
router, async_add_entities, tracked, entity_class_generator
|
||||
)
|
||||
|
||||
entry.async_on_unload(
|
||||
async_dispatcher_connect(hass, router.signal_device_new, _async_router_updated)
|
||||
)
|
||||
|
||||
_async_router_updated()
|
||||
|
||||
|
||||
@callback
|
||||
def async_add_new_entities(router, async_add_entities, tracked, entity_class_generator):
|
||||
"""Add new tracker entities from the router."""
|
||||
new_tracked = []
|
||||
|
||||
for mac, device in router.devices.items():
|
||||
if mac in tracked:
|
||||
continue
|
||||
|
||||
new_tracked.extend(entity_class_generator(router, device))
|
||||
tracked.add(mac)
|
||||
|
||||
if new_tracked:
|
||||
async_add_entities(new_tracked, True)
|
||||
|
||||
|
||||
class NetgearRouter:
|
||||
"""Representation of a Netgear router."""
|
||||
|
||||
def __init__(self, hass: HomeAssistantType, entry: ConfigEntry) -> None:
|
||||
"""Initialize a Netgear router."""
|
||||
self.hass = hass
|
||||
self.entry = entry
|
||||
self.entry_id = entry.entry_id
|
||||
self.unique_id = entry.unique_id
|
||||
self._host = entry.data.get(CONF_HOST)
|
||||
self._port = entry.data.get(CONF_PORT)
|
||||
self._ssl = entry.data.get(CONF_SSL)
|
||||
self._username = entry.data.get(CONF_USERNAME)
|
||||
self._password = entry.data[CONF_PASSWORD]
|
||||
|
||||
self._info = None
|
||||
self.model = None
|
||||
self.device_name = None
|
||||
self.firmware_version = None
|
||||
|
||||
self._method_version = 1
|
||||
consider_home_int = entry.options.get(
|
||||
CONF_CONSIDER_HOME, DEFAULT_CONSIDER_HOME.total_seconds()
|
||||
)
|
||||
self._consider_home = timedelta(seconds=consider_home_int)
|
||||
|
||||
self._api: Netgear = None
|
||||
self._attrs = {}
|
||||
|
||||
self.devices = {}
|
||||
|
||||
def _setup(self) -> None:
|
||||
"""Set up a Netgear router sync portion."""
|
||||
self._api = get_api(
|
||||
self._password,
|
||||
self._host,
|
||||
self._username,
|
||||
self._port,
|
||||
self._ssl,
|
||||
)
|
||||
|
||||
self._info = self._api.get_info()
|
||||
self.device_name = self._info.get("DeviceName", DEFAULT_NAME)
|
||||
self.model = self._info.get("ModelName")
|
||||
self.firmware_version = self._info.get("Firmwareversion")
|
||||
|
||||
if self.model in MODELS_V2:
|
||||
self._method_version = 2
|
||||
|
||||
async def async_setup(self) -> None:
|
||||
"""Set up a Netgear router."""
|
||||
await self.hass.async_add_executor_job(self._setup)
|
||||
|
||||
# set already known devices to away instead of unavailable
|
||||
device_registry = dr.async_get(self.hass)
|
||||
devices = dr.async_entries_for_config_entry(device_registry, self.entry_id)
|
||||
for device_entry in devices:
|
||||
if device_entry.via_device_id is None:
|
||||
continue # do not add the router itself
|
||||
|
||||
device_mac = dict(device_entry.connections).get(dr.CONNECTION_NETWORK_MAC)
|
||||
self.devices[device_mac] = {
|
||||
"mac": device_mac,
|
||||
"name": device_entry.name,
|
||||
"active": False,
|
||||
"last_seen": dt_util.utcnow() - timedelta(days=365),
|
||||
"device_model": None,
|
||||
"device_type": None,
|
||||
"type": None,
|
||||
"link_rate": None,
|
||||
"signal": None,
|
||||
"ip": None,
|
||||
}
|
||||
|
||||
await self.async_update_device_trackers()
|
||||
self.entry.async_on_unload(
|
||||
async_track_time_interval(
|
||||
self.hass, self.async_update_device_trackers, SCAN_INTERVAL
|
||||
)
|
||||
)
|
||||
|
||||
async_dispatcher_send(self.hass, self.signal_device_new)
|
||||
|
||||
async def async_get_attached_devices(self) -> list:
|
||||
"""Get the devices connected to the router."""
|
||||
if self._method_version == 1:
|
||||
return await self.hass.async_add_executor_job(
|
||||
self._api.get_attached_devices
|
||||
)
|
||||
|
||||
return await self.hass.async_add_executor_job(self._api.get_attached_devices_2)
|
||||
|
||||
async def async_update_device_trackers(self, now=None) -> None:
|
||||
"""Update Netgear devices."""
|
||||
new_device = False
|
||||
ntg_devices = await self.async_get_attached_devices()
|
||||
now = dt_util.utcnow()
|
||||
|
||||
for ntg_device in ntg_devices:
|
||||
device_mac = format_mac(ntg_device.mac)
|
||||
|
||||
if self._method_version == 2 and not ntg_device.link_rate:
|
||||
continue
|
||||
|
||||
if not self.devices.get(device_mac):
|
||||
new_device = True
|
||||
|
||||
# ntg_device is a namedtuple from the collections module that needs conversion to a dict through ._asdict method
|
||||
self.devices[device_mac] = ntg_device._asdict()
|
||||
self.devices[device_mac]["mac"] = device_mac
|
||||
self.devices[device_mac]["last_seen"] = now
|
||||
|
||||
for device in self.devices.values():
|
||||
device["active"] = now - device["last_seen"] <= self._consider_home
|
||||
|
||||
async_dispatcher_send(self.hass, self.signal_device_update)
|
||||
|
||||
if new_device:
|
||||
_LOGGER.debug("Netgear tracker: new device found")
|
||||
async_dispatcher_send(self.hass, self.signal_device_new)
|
||||
|
||||
@property
|
||||
def signal_device_new(self) -> str:
|
||||
"""Event specific per Netgear entry to signal new device."""
|
||||
return f"{DOMAIN}-{self._host}-device-new"
|
||||
|
||||
@property
|
||||
def signal_device_update(self) -> str:
|
||||
"""Event specific per Netgear entry to signal updates in devices."""
|
||||
return f"{DOMAIN}-{self._host}-device-update"
|
||||
|
||||
|
||||
class NetgearDeviceEntity(Entity):
|
||||
"""Base class for a device connected to a Netgear router."""
|
||||
|
||||
def __init__(self, router: NetgearRouter, device: dict) -> None:
|
||||
"""Initialize a Netgear device."""
|
||||
self._router = router
|
||||
self._device = device
|
||||
self._mac = device["mac"]
|
||||
self._name = self.get_device_name()
|
||||
self._device_name = self._name
|
||||
self._unique_id = self._mac
|
||||
self._active = device["active"]
|
||||
|
||||
def get_device_name(self):
|
||||
"""Return the name of the given device or the MAC if we don't know."""
|
||||
name = self._device["name"]
|
||||
if not name or name == "--":
|
||||
name = self._mac
|
||||
|
||||
return name
|
||||
|
||||
@abstractmethod
|
||||
@callback
|
||||
def async_update_device(self) -> None:
|
||||
"""Update the Netgear device."""
|
||||
|
||||
@property
|
||||
def unique_id(self) -> str:
|
||||
"""Return a unique ID."""
|
||||
return self._unique_id
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Return the name."""
|
||||
return self._name
|
||||
|
||||
@property
|
||||
def device_info(self):
|
||||
"""Return the device information."""
|
||||
return {
|
||||
"connections": {(CONNECTION_NETWORK_MAC, self._mac)},
|
||||
"name": self._device_name,
|
||||
"model": self._device["device_model"],
|
||||
"via_device": (DOMAIN, self._router.unique_id),
|
||||
}
|
||||
|
||||
@property
|
||||
def should_poll(self) -> bool:
|
||||
"""No polling needed."""
|
||||
return False
|
||||
|
||||
async def async_added_to_hass(self):
|
||||
"""Register state update callback."""
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass,
|
||||
self._router.signal_device_update,
|
||||
self.async_update_device,
|
||||
)
|
||||
)
|
||||
83
homeassistant/components/netgear/sensor.py
Normal file
83
homeassistant/components/netgear/sensor.py
Normal file
@ -0,0 +1,83 @@
|
||||
"""Support for Netgear routers."""
|
||||
import logging
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import PERCENTAGE
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
from homeassistant.helpers.typing import HomeAssistantType
|
||||
|
||||
from .router import NetgearDeviceEntity, NetgearRouter, async_setup_netgear_entry
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
SENSOR_TYPES = {
|
||||
"type": SensorEntityDescription(
|
||||
key="type",
|
||||
name="link type",
|
||||
native_unit_of_measurement=None,
|
||||
device_class=None,
|
||||
),
|
||||
"link_rate": SensorEntityDescription(
|
||||
key="link_rate",
|
||||
name="link rate",
|
||||
native_unit_of_measurement="Mbps",
|
||||
device_class=None,
|
||||
),
|
||||
"signal": SensorEntityDescription(
|
||||
key="signal",
|
||||
name="signal strength",
|
||||
native_unit_of_measurement=PERCENTAGE,
|
||||
device_class=DEVICE_CLASS_SIGNAL_STRENGTH,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistantType, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
|
||||
) -> None:
|
||||
"""Set up device tracker for Netgear component."""
|
||||
|
||||
def generate_sensor_classes(router: NetgearRouter, device: dict):
|
||||
return [
|
||||
NetgearSensorEntity(router, device, attribute)
|
||||
for attribute in ("type", "link_rate", "signal")
|
||||
]
|
||||
|
||||
async_setup_netgear_entry(hass, entry, async_add_entities, generate_sensor_classes)
|
||||
|
||||
|
||||
class NetgearSensorEntity(NetgearDeviceEntity, SensorEntity):
|
||||
"""Representation of a device connected to a Netgear router."""
|
||||
|
||||
_attr_entity_registry_enabled_default = False
|
||||
|
||||
def __init__(self, router: NetgearRouter, device: dict, attribute: str) -> None:
|
||||
"""Initialize a Netgear device."""
|
||||
super().__init__(router, device)
|
||||
self._attribute = attribute
|
||||
self.entity_description = SENSOR_TYPES[self._attribute]
|
||||
self._name = f"{self.get_device_name()} {self.entity_description.name}"
|
||||
self._unique_id = f"{self._mac}-{self._attribute}"
|
||||
self._state = self._device[self._attribute]
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
"""Return the state of the sensor."""
|
||||
return self._state
|
||||
|
||||
@callback
|
||||
def async_update_device(self) -> None:
|
||||
"""Update the Netgear device."""
|
||||
self._device = self._router.devices[self._mac]
|
||||
self._active = self._device["active"]
|
||||
if self._device[self._attribute] is not None:
|
||||
self._state = self._device[self._attribute]
|
||||
|
||||
self.async_write_ha_state()
|
||||
34
homeassistant/components/netgear/strings.json
Normal file
34
homeassistant/components/netgear/strings.json
Normal file
@ -0,0 +1,34 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Netgear",
|
||||
"description": "Default host: {host}\n Default port: {port}\n Default username: {username}",
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%] (Optional)",
|
||||
"port": "[%key:common::config_flow::data::port%] (Optional)",
|
||||
"ssl": "[%key:common::config_flow::data::ssl%]",
|
||||
"username": "[%key:common::config_flow::data::username%] (Optional)",
|
||||
"password": "[%key:common::config_flow::data::password%]"
|
||||
}
|
||||
}
|
||||
},
|
||||
"error": {
|
||||
"config": "Connection or login error: please check your configuration"
|
||||
},
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"title": "Netgear",
|
||||
"description": "Specify optional settings",
|
||||
"data": {
|
||||
"consider_home": "Consider home time (seconds)"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
34
homeassistant/components/netgear/translations/en.json
Normal file
34
homeassistant/components/netgear/translations/en.json
Normal file
@ -0,0 +1,34 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "Host already configured"
|
||||
},
|
||||
"error": {
|
||||
"config": "Connection or login error: please check your configuration"
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "Host (Optional)",
|
||||
"password": "Password",
|
||||
"port": "Port (Optional)",
|
||||
"ssl": "Use SSL (Optional)",
|
||||
"username": "Username (Optional)"
|
||||
},
|
||||
"description": "Default host: {host}\n Default port: {port}\n Default username: {username}",
|
||||
"title": "Netgear"
|
||||
}
|
||||
}
|
||||
},
|
||||
"options": {
|
||||
"step": {
|
||||
"init": {
|
||||
"title": "Netgear",
|
||||
"description": "Specify optional settings",
|
||||
"data": {
|
||||
"consider_home": "Consider home time (seconds)"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -181,6 +181,7 @@ FLOWS = [
|
||||
"neato",
|
||||
"nest",
|
||||
"netatmo",
|
||||
"netgear",
|
||||
"nexia",
|
||||
"nfandroidtv",
|
||||
"nightscout",
|
||||
|
||||
@ -151,6 +151,12 @@ SSDP = {
|
||||
"manufacturer": "konnected.io"
|
||||
}
|
||||
],
|
||||
"netgear": [
|
||||
{
|
||||
"deviceType": "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
|
||||
"manufacturer": "NETGEAR, Inc."
|
||||
}
|
||||
],
|
||||
"roku": [
|
||||
{
|
||||
"deviceType": "urn:roku-com:device:player:1-0",
|
||||
|
||||
3
mypy.ini
3
mypy.ini
@ -1500,6 +1500,9 @@ ignore_errors = true
|
||||
[mypy-homeassistant.components.nest.legacy.*]
|
||||
ignore_errors = true
|
||||
|
||||
[mypy-homeassistant.components.netgear.*]
|
||||
ignore_errors = true
|
||||
|
||||
[mypy-homeassistant.components.nightscout.*]
|
||||
ignore_errors = true
|
||||
|
||||
|
||||
@ -1641,7 +1641,7 @@ pynanoleaf==0.1.0
|
||||
pynello==2.0.3
|
||||
|
||||
# homeassistant.components.netgear
|
||||
pynetgear==0.6.1
|
||||
pynetgear==0.7.0
|
||||
|
||||
# homeassistant.components.netio
|
||||
pynetio==0.1.9.1
|
||||
|
||||
@ -950,6 +950,9 @@ pymysensors==0.21.0
|
||||
# homeassistant.components.nanoleaf
|
||||
pynanoleaf==0.1.0
|
||||
|
||||
# homeassistant.components.netgear
|
||||
pynetgear==0.7.0
|
||||
|
||||
# homeassistant.components.nuki
|
||||
pynuki==1.4.1
|
||||
|
||||
|
||||
@ -85,6 +85,7 @@ IGNORED_MODULES: Final[list[str]] = [
|
||||
"homeassistant.components.mullvad.*",
|
||||
"homeassistant.components.ness_alarm.*",
|
||||
"homeassistant.components.nest.legacy.*",
|
||||
"homeassistant.components.netgear.*",
|
||||
"homeassistant.components.nightscout.*",
|
||||
"homeassistant.components.nilu.*",
|
||||
"homeassistant.components.nsw_fuel_station.*",
|
||||
|
||||
@ -16,8 +16,10 @@ from tests.common import async_fire_time_changed, mock_coro
|
||||
SERVICE = "yamaha"
|
||||
SERVICE_COMPONENT = "media_player"
|
||||
|
||||
SERVICE_NO_PLATFORM = "netgear_router"
|
||||
SERVICE_NO_PLATFORM_COMPONENT = "device_tracker"
|
||||
# sabnzbd is the last no platform integration to be migrated
|
||||
# drop these tests once it is migrated
|
||||
SERVICE_NO_PLATFORM = "sabnzbd"
|
||||
SERVICE_NO_PLATFORM_COMPONENT = "sabnzbd"
|
||||
SERVICE_INFO = {"key": "value"} # Can be anything
|
||||
|
||||
UNKNOWN_SERVICE = "this_service_will_never_be_supported"
|
||||
|
||||
1
tests/components/netgear/__init__.py
Normal file
1
tests/components/netgear/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
"""Tests for the Netgear component."""
|
||||
14
tests/components/netgear/conftest.py
Normal file
14
tests/components/netgear/conftest.py
Normal file
@ -0,0 +1,14 @@
|
||||
"""Configure Netgear tests."""
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(name="bypass_setup", autouse=True)
|
||||
def bypass_setup_fixture():
|
||||
"""Mock component setup."""
|
||||
with patch(
|
||||
"homeassistant.components.netgear.device_tracker.async_get_scanner",
|
||||
return_value=None,
|
||||
):
|
||||
yield
|
||||
284
tests/components/netgear/test_config_flow.py
Normal file
284
tests/components/netgear/test_config_flow.py
Normal file
@ -0,0 +1,284 @@
|
||||
"""Tests for the Netgear config flow."""
|
||||
import logging
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from pynetgear import DEFAULT_HOST, DEFAULT_PORT, DEFAULT_USER
|
||||
import pytest
|
||||
|
||||
from homeassistant import data_entry_flow
|
||||
from homeassistant.components import ssdp
|
||||
from homeassistant.components.netgear.const import CONF_CONSIDER_HOME, DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_IMPORT, SOURCE_SSDP, SOURCE_USER
|
||||
from homeassistant.const import (
|
||||
CONF_HOST,
|
||||
CONF_PASSWORD,
|
||||
CONF_PORT,
|
||||
CONF_SSL,
|
||||
CONF_USERNAME,
|
||||
)
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
URL = "http://routerlogin.net"
|
||||
SERIAL = "5ER1AL0000001"
|
||||
|
||||
ROUTER_INFOS = {
|
||||
"Description": "Netgear Smart Wizard 3.0, specification 1.6 version",
|
||||
"SignalStrength": "-4",
|
||||
"SmartAgentversion": "3.0",
|
||||
"FirewallVersion": "net-wall 2.0",
|
||||
"VPNVersion": None,
|
||||
"OthersoftwareVersion": "N/A",
|
||||
"Hardwareversion": "N/A",
|
||||
"Otherhardwareversion": "N/A",
|
||||
"FirstUseDate": "Sunday, 30 Sep 2007 01:10:03",
|
||||
"DeviceMode": "0",
|
||||
"ModelName": "RBR20",
|
||||
"SerialNumber": SERIAL,
|
||||
"Firmwareversion": "V2.3.5.26",
|
||||
"DeviceName": "Desk",
|
||||
"DeviceNameUserSet": "true",
|
||||
"FirmwareDLmethod": "HTTPS",
|
||||
"FirmwareLastUpdate": "2019_10.5_18:42:58",
|
||||
"FirmwareLastChecked": "2020_5.3_1:33:0",
|
||||
"DeviceModeCapability": "0;1",
|
||||
}
|
||||
TITLE = f"{ROUTER_INFOS['ModelName']} - {ROUTER_INFOS['DeviceName']}"
|
||||
|
||||
HOST = "10.0.0.1"
|
||||
SERIAL_2 = "5ER1AL0000002"
|
||||
PORT = 80
|
||||
SSL = False
|
||||
USERNAME = "Home_Assistant"
|
||||
PASSWORD = "password"
|
||||
SSDP_URL = f"http://{HOST}:{PORT}/rootDesc.xml"
|
||||
SSDP_URL_SLL = f"https://{HOST}:{PORT}/rootDesc.xml"
|
||||
|
||||
|
||||
@pytest.fixture(name="service")
|
||||
def mock_controller_service():
|
||||
"""Mock a successful service."""
|
||||
with patch(
|
||||
"homeassistant.components.netgear.async_setup_entry", return_value=True
|
||||
), patch("homeassistant.components.netgear.router.Netgear") as service_mock:
|
||||
service_mock.return_value.get_info = Mock(return_value=ROUTER_INFOS)
|
||||
yield service_mock
|
||||
|
||||
|
||||
@pytest.fixture(name="service_failed")
|
||||
def mock_controller_service_failed():
|
||||
"""Mock a failed service."""
|
||||
with patch("homeassistant.components.netgear.router.Netgear") as service_mock:
|
||||
service_mock.return_value.login = Mock(return_value=None)
|
||||
service_mock.return_value.get_info = Mock(return_value=None)
|
||||
yield service_mock
|
||||
|
||||
|
||||
async def test_user(hass, service):
|
||||
"""Test user step."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
# Have to provide all config
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{
|
||||
CONF_HOST: HOST,
|
||||
CONF_PORT: PORT,
|
||||
CONF_SSL: SSL,
|
||||
CONF_USERNAME: USERNAME,
|
||||
CONF_PASSWORD: PASSWORD,
|
||||
},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["result"].unique_id == SERIAL
|
||||
assert result["title"] == TITLE
|
||||
assert result["data"].get(CONF_HOST) == HOST
|
||||
assert result["data"].get(CONF_PORT) == PORT
|
||||
assert result["data"].get(CONF_SSL) == SSL
|
||||
assert result["data"].get(CONF_USERNAME) == USERNAME
|
||||
assert result["data"][CONF_PASSWORD] == PASSWORD
|
||||
|
||||
|
||||
async def test_import_required(hass, service):
|
||||
"""Test import step, with required config only."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data={CONF_PASSWORD: PASSWORD}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["result"].unique_id == SERIAL
|
||||
assert result["title"] == TITLE
|
||||
assert result["data"].get(CONF_HOST) == DEFAULT_HOST
|
||||
assert result["data"].get(CONF_PORT) == DEFAULT_PORT
|
||||
assert result["data"].get(CONF_SSL) is False
|
||||
assert result["data"].get(CONF_USERNAME) == DEFAULT_USER
|
||||
assert result["data"][CONF_PASSWORD] == PASSWORD
|
||||
|
||||
|
||||
async def test_import_required_login_failed(hass, service_failed):
|
||||
"""Test import step, with required config only, while wrong password or connection issue."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_IMPORT}, data={CONF_PASSWORD: PASSWORD}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "config"}
|
||||
|
||||
|
||||
async def test_import_all(hass, service):
|
||||
"""Test import step, with all config provided."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={
|
||||
CONF_HOST: HOST,
|
||||
CONF_PORT: PORT,
|
||||
CONF_SSL: SSL,
|
||||
CONF_USERNAME: USERNAME,
|
||||
CONF_PASSWORD: PASSWORD,
|
||||
},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["result"].unique_id == SERIAL
|
||||
assert result["title"] == TITLE
|
||||
assert result["data"].get(CONF_HOST) == HOST
|
||||
assert result["data"].get(CONF_PORT) == PORT
|
||||
assert result["data"].get(CONF_SSL) == SSL
|
||||
assert result["data"].get(CONF_USERNAME) == USERNAME
|
||||
assert result["data"][CONF_PASSWORD] == PASSWORD
|
||||
|
||||
|
||||
async def test_import_all_connection_failed(hass, service_failed):
|
||||
"""Test import step, with all config provided, while wrong host."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={
|
||||
CONF_HOST: HOST,
|
||||
CONF_PORT: PORT,
|
||||
CONF_SSL: SSL,
|
||||
CONF_USERNAME: USERNAME,
|
||||
CONF_PASSWORD: PASSWORD,
|
||||
},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert result["errors"] == {"base": "config"}
|
||||
|
||||
|
||||
async def test_abort_if_already_setup(hass, service):
|
||||
"""Test we abort if the router is already setup."""
|
||||
MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={CONF_PASSWORD: PASSWORD},
|
||||
unique_id=SERIAL,
|
||||
).add_to_hass(hass)
|
||||
|
||||
# Should fail, same SERIAL (import)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_IMPORT},
|
||||
data={CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
# Should fail, same SERIAL (flow)
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"],
|
||||
{CONF_PASSWORD: PASSWORD},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_ssdp_already_configured(hass):
|
||||
"""Test ssdp abort when the router is already configured."""
|
||||
MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={CONF_PASSWORD: PASSWORD},
|
||||
unique_id=SERIAL,
|
||||
).add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_SSDP},
|
||||
data={
|
||||
ssdp.ATTR_SSDP_LOCATION: SSDP_URL_SLL,
|
||||
ssdp.ATTR_UPNP_MODEL_NUMBER: "RBR20",
|
||||
ssdp.ATTR_UPNP_PRESENTATION_URL: URL,
|
||||
ssdp.ATTR_UPNP_SERIAL: SERIAL,
|
||||
},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
|
||||
|
||||
async def test_ssdp(hass, service):
|
||||
"""Test ssdp step."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN,
|
||||
context={"source": SOURCE_SSDP},
|
||||
data={
|
||||
ssdp.ATTR_SSDP_LOCATION: SSDP_URL,
|
||||
ssdp.ATTR_UPNP_MODEL_NUMBER: "RBR20",
|
||||
ssdp.ATTR_UPNP_PRESENTATION_URL: URL,
|
||||
ssdp.ATTR_UPNP_SERIAL: SERIAL,
|
||||
},
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "user"
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], {CONF_PASSWORD: PASSWORD}
|
||||
)
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert result["result"].unique_id == SERIAL
|
||||
assert result["title"] == TITLE
|
||||
assert result["data"].get(CONF_HOST) == HOST
|
||||
assert result["data"].get(CONF_PORT) == PORT
|
||||
assert result["data"].get(CONF_SSL) == SSL
|
||||
assert result["data"].get(CONF_USERNAME) == DEFAULT_USER
|
||||
assert result["data"][CONF_PASSWORD] == PASSWORD
|
||||
|
||||
|
||||
async def test_options_flow(hass, service):
|
||||
"""Test specifying non default settings using options flow."""
|
||||
config_entry = MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
data={CONF_PASSWORD: PASSWORD},
|
||||
unique_id=SERIAL,
|
||||
title=TITLE,
|
||||
)
|
||||
config_entry.add_to_hass(hass)
|
||||
|
||||
assert await hass.config_entries.async_setup(config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
result = await hass.config_entries.options.async_init(config_entry.entry_id)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
|
||||
assert result["step_id"] == "init"
|
||||
|
||||
result = await hass.config_entries.options.async_configure(
|
||||
result["flow_id"],
|
||||
user_input={
|
||||
CONF_CONSIDER_HOME: 1800,
|
||||
},
|
||||
)
|
||||
|
||||
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
|
||||
assert config_entry.options == {
|
||||
CONF_CONSIDER_HOME: 1800,
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user