github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

SSDP Discovery for NDMS2 routers (#47312) 

Co-authored-by: Franck Nijhof <frenck@frenck.nl>
This commit is contained in:
Andrey Kupreychik 2021-05-25 20:36:03 +07:00 committed by GitHub
parent 7c9d8cfdec
commit cbe4df1893
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 185 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
homeassistant/components/keenetic_ndms2/config_flow.py
View File

@ -1,10 +1,13 @@
"""Config flow for Keenetic NDMS2."""
from __future__ import annotations
from urllib.parse import urlparse
from ndms2_client import Client, ConnectionException, InterfaceInfo, TelnetConnection
import voluptuous as vol
from homeassistant import config_entries
from homeassistant.components import ssdp
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_HOST,
@ -14,7 +17,9 @@ from homeassistant.const import (
CONF_USERNAME,
)
from homeassistant.core import callback
import homeassistant.helpers.config_validation as cv
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from .const import (
CONF_CONSIDER_HOME,
@ -39,19 +44,22 @@ class KeeneticFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(
config_entry: ConfigEntry,
) -> KeeneticOptionsFlowHandler:
"""Get the options flow for this handler."""
return KeeneticOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input=None):
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
"""Handle a flow initialized by the user."""
errors = {}
if user_input is not None:
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
host = self.context.get(CONF_HOST) or user_input[CONF_HOST]
self._async_abort_entries_match({CONF_HOST: host})
_client = Client(
TelnetConnection(
user_input[CONF_HOST],
host,
user_input[CONF_PORT],
user_input[CONF_USERNAME],
user_input[CONF_PASSWORD],
@ -66,13 +74,19 @@ class KeeneticFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
except ConnectionException:
errors["base"] = "cannot_connect"
else:
return self.async_create_entry(title=router_info.name, data=user_input)
return self.async_create_entry(
title=router_info.name, data={CONF_HOST: host, **user_input}
)
host_schema = (
{vol.Required(CONF_HOST): str} if CONF_HOST not in self.context else {}
)
return self.async_show_form(
step_id="user",
data_schema=vol.Schema(
{
vol.Required(CONF_HOST): str,
**host_schema,
vol.Required(CONF_USERNAME): str,
vol.Required(CONF_PASSWORD): str,
vol.Optional(CONF_PORT, default=DEFAULT_TELNET_PORT): int,
@ -81,10 +95,37 @@ class KeeneticFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors,
)
async def async_step_import(self, user_input=None):
async def async_step_import(
self, user_input: ConfigType | None = None
) -> FlowResult:
"""Import a config entry."""
return await self.async_step_user(user_input)
async def async_step_ssdp(self, discovery_info: DiscoveryInfoType) -> FlowResult:
"""Handle a discovered device."""
friendly_name = discovery_info.get(ssdp.ATTR_UPNP_FRIENDLY_NAME, "")
# Filter out items not having "keenetic" in their name
if "keenetic" not in friendly_name.lower():
return self.async_abort(reason="not_keenetic_ndms2")
# Filters out items having no/empty UDN
if not discovery_info.get(ssdp.ATTR_UPNP_UDN):
return self.async_abort(reason="no_udn")
host = urlparse(discovery_info[ssdp.ATTR_SSDP_LOCATION]).hostname
await self.async_set_unique_id(discovery_info[ssdp.ATTR_UPNP_UDN])
self._async_abort_entries_match({CONF_HOST: host})
self.context[CONF_HOST] = host
self.context["title_placeholders"] = {
"name": friendly_name,
"host": host,
}
return await self.async_step_user()
class KeeneticOptionsFlowHandler(config_entries.OptionsFlow):
"""Handle options."""
@ -94,7 +135,7 @@ class KeeneticOptionsFlowHandler(config_entries.OptionsFlow):
self.config_entry = config_entry
self._interface_options = {}
async def async_step_init(self, _user_input=None):
async def async_step_init(self, user_input: ConfigType | None = None) -> FlowResult:
"""Manage the options."""
router: KeeneticRouter = self.hass.data[DOMAIN][self.config_entry.entry_id][
ROUTER
@ -111,7 +152,7 @@ class KeeneticOptionsFlowHandler(config_entries.OptionsFlow):
}
return await self.async_step_user()
async def async_step_user(self, user_input=None):
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
"""Manage the device tracker options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

10
homeassistant/components/keenetic_ndms2/manifest.json
View File

@ -4,6 +4,16 @@
"config_flow": true,
"documentation": "https://www.home-assistant.io/integrations/keenetic_ndms2",
"requirements": ["ndms2_client==0.1.1"],
"ssdp": [
{
"deviceType": "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
"manufacturer": "Keenetic Ltd."
},
{
"deviceType": "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
"manufacturer": "ZyXEL Communications Corp."
}
],
"codeowners": ["@foxel"],
"iot_class": "local_polling"
}

5
homeassistant/components/keenetic_ndms2/strings.json
View File

@ -1,5 +1,6 @@
{
"config": {
"flow_title": "{name} ({host})",
"step": {
"user": {
"title": "Set up Keenetic NDMS2 Router",
@ -15,7 +16,9 @@
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]"
},
"abort": {
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
"already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
"no_udn": "SSDP discovery info has no UDN",
"not_keenetic_ndms2": "Discovered item is not a Keenetic router"
}
},
"options": {

7
homeassistant/components/keenetic_ndms2/translations/en.json
View File

@ -1,11 +1,14 @@
{
"config": {
"abort": {
"already_configured": "Account is already configured"
"already_configured": "Account is already configured",
"no_udn": "SSDP discovery info has no UDN",
"not_keenetic_ndms2": "Discovered item is no a Keenetic router"
},
"error": {
"cannot_connect": "Failed to connect"
},
"flow_title": "{name} ({host})",
"step": {
"user": {
"data": {
@ -32,4 +35,4 @@
}
}
}
}
}

3
homeassistant/components/keenetic_ndms2/translations/ru.json
View File

@ -6,6 +6,7 @@
"error": {
"cannot_connect": "\u041d\u0435 \u0443\u0434\u0430\u043b\u043e\u0441\u044c \u043f\u043e\u0434\u043a\u043b\u044e\u0447\u0438\u0442\u044c\u0441\u044f."
},
"flow_title": "{name} ({host})",
"step": {
"user": {
"data": {
@ -32,4 +33,4 @@
}
}
}
}
}

10
homeassistant/generated/ssdp.py
View File

@ -136,6 +136,16 @@ SSDP = {
"manufacturer": "Universal Devices Inc."
}
],
"keenetic_ndms2": [
{
"deviceType": "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
"manufacturer": "Keenetic Ltd."
},
{
"deviceType": "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
"manufacturer": "ZyXEL Communications Corp."
}
],
"konnected": [
{
"manufacturer": "konnected.io"

11
tests/components/keenetic_ndms2/__init__.py
View File

@ -1,4 +1,5 @@
"""Tests for the Keenetic NDMS2 component."""
from homeassistant.components import ssdp
from homeassistant.components.keenetic_ndms2 import const
from homeassistant.const import (
CONF_HOST,
@ -9,9 +10,11 @@ from homeassistant.const import (
)
MOCK_NAME = "Keenetic Ultra 2030"
MOCK_IP = "0.0.0.0"
SSDP_LOCATION = f"http://{MOCK_IP}/"
MOCK_DATA = {
CONF_HOST: "0.0.0.0",
CONF_HOST: MOCK_IP,
CONF_USERNAME: "user",
CONF_PASSWORD: "pass",
CONF_PORT: 23,
@ -25,3 +28,9 @@ MOCK_OPTIONS = {
const.CONF_INCLUDE_ASSOCIATED: True,
const.CONF_INTERFACES: ["Home", "VPS0"],
}
MOCK_SSDP_DISCOVERY_INFO = {
ssdp.ATTR_SSDP_LOCATION: SSDP_LOCATION,
ssdp.ATTR_UPNP_UDN: "uuid:xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
ssdp.ATTR_UPNP_FRIENDLY_NAME: MOCK_NAME,
}

100
tests/components/keenetic_ndms2/test_config_flow.py
View File

@ -7,11 +7,12 @@ from ndms2_client.client import InterfaceInfo, RouterInfo
import pytest
from homeassistant import config_entries, data_entry_flow
from homeassistant.components import keenetic_ndms2 as keenetic
from homeassistant.components import keenetic_ndms2 as keenetic, ssdp
from homeassistant.components.keenetic_ndms2 import const
from homeassistant.const import CONF_HOST, CONF_SOURCE
from homeassistant.core import HomeAssistant
from . import MOCK_DATA, MOCK_NAME, MOCK_OPTIONS
from . import MOCK_DATA, MOCK_NAME, MOCK_OPTIONS, MOCK_SSDP_DISCOVERY_INFO
from tests.common import MockConfigEntry
@ -43,7 +44,7 @@ def mock_keenetic_connect_failed():
yield
async def test_flow_works(hass: HomeAssistant, connect):
async def test_flow_works(hass: HomeAssistant, connect) -> None:
"""Test config flow."""
result = await hass.config_entries.flow.async_init(
@ -67,7 +68,7 @@ async def test_flow_works(hass: HomeAssistant, connect):
assert len(mock_setup_entry.mock_calls) == 1
async def test_import_works(hass: HomeAssistant, connect):
async def test_import_works(hass: HomeAssistant, connect) -> None:
"""Test config flow."""
with patch(
@ -86,7 +87,7 @@ async def test_import_works(hass: HomeAssistant, connect):
assert len(mock_setup_entry.mock_calls) == 1
async def test_options(hass):
async def test_options(hass: HomeAssistant) -> None:
"""Test updating options."""
entry = MockConfigEntry(domain=keenetic.DOMAIN, data=MOCK_DATA)
entry.add_to_hass(hass)
@ -127,7 +128,7 @@ async def test_options(hass):
assert result2["data"] == MOCK_OPTIONS
async def test_host_already_configured(hass, connect):
async def test_host_already_configured(hass: HomeAssistant, connect) -> None:
"""Test host already configured."""
entry = MockConfigEntry(
@ -147,7 +148,7 @@ async def test_host_already_configured(hass, connect):
assert result2["reason"] == "already_configured"
async def test_connection_error(hass, connect_error):
async def test_connection_error(hass: HomeAssistant, connect_error) -> None:
"""Test error when connection is unsuccessful."""
result = await hass.config_entries.flow.async_init(
@ -158,3 +159,88 @@ async def test_connection_error(hass, connect_error):
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["errors"] == {"base": "cannot_connect"}
async def test_ssdp_works(hass: HomeAssistant, connect) -> None:
"""Test host already configured and discovered."""
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
result = await hass.config_entries.flow.async_init(
keenetic.DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_SSDP},
data=discovery_info,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
with patch(
"homeassistant.components.keenetic_ndms2.async_setup_entry", return_value=True
) as mock_setup_entry:
user_input = MOCK_DATA.copy()
user_input.pop(CONF_HOST)
result2 = await hass.config_entries.flow.async_configure(
result["flow_id"],
user_input=user_input,
)
await hass.async_block_till_done()
assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result2["title"] == MOCK_NAME
assert result2["data"] == MOCK_DATA
assert len(mock_setup_entry.mock_calls) == 1
async def test_ssdp_already_configured(hass: HomeAssistant) -> None:
"""Test host already configured and discovered."""
entry = MockConfigEntry(
domain=keenetic.DOMAIN, data=MOCK_DATA, options=MOCK_OPTIONS
)
entry.add_to_hass(hass)
discovery_info = MOCK_SSDP_DISCOVERY_INFO.copy()
result = await hass.config_entries.flow.async_init(
keenetic.DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_SSDP},
data=discovery_info,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "already_configured"
async def test_ssdp_reject_no_udn(hass: HomeAssistant) -> None:
"""Discovered device has no UDN."""
discovery_info = {
**MOCK_SSDP_DISCOVERY_INFO,
}
discovery_info.pop(ssdp.ATTR_UPNP_UDN)
result = await hass.config_entries.flow.async_init(
keenetic.DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_SSDP},
data=discovery_info,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "no_udn"
async def test_ssdp_reject_non_keenetic(hass: HomeAssistant) -> None:
"""Discovered device does not look like a keenetic router."""
discovery_info = {
**MOCK_SSDP_DISCOVERY_INFO,
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Suspicious device",
}
result = await hass.config_entries.flow.async_init(
keenetic.DOMAIN,
context={CONF_SOURCE: config_entries.SOURCE_SSDP},
data=discovery_info,
)
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "not_keenetic_ndms2"