From 6a7b5657acc7a201fa4817c9e49f015a327e2e6d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Ville=20Skytt=C3=A4?= <ville.skytta@iki.fi>
Date: Mon, 4 Nov 2019 19:56:49 +0200
Subject: [PATCH] Support Huawei LTE SSDP discovery (#28214)
* Support Huawei LTE SSDP discovery
* Avoid KeyError on simultaneous user initiated flow
Co-Authored-By: Paulus Schoutsen <paulus@home-assistant.io>
* Format code
* Add already configured check
* Initialize context in test flows
* Move deviceType match to manifest
* Update generated.ssdp
* Add SSDP config flow test case
* Remove stale debug print from tests
---
.../huawei_lte/.translations/en.json | 4 +-
.../components/huawei_lte/config_flow.py | 50 +++++++++++--
.../components/huawei_lte/manifest.json | 6 ++
.../components/huawei_lte/strings.json | 4 +-
homeassistant/generated/ssdp.py | 6 ++
.../components/huawei_lte/test_config_flow.py | 73 ++++++++++++++-----
6 files changed, 116 insertions(+), 27 deletions(-)
diff --git a/homeassistant/components/huawei_lte/.translations/en.json b/homeassistant/components/huawei_lte/.translations/en.json
index 8681e3355a4..0952b05a5cf 100644
--- a/homeassistant/components/huawei_lte/.translations/en.json
+++ b/homeassistant/components/huawei_lte/.translations/en.json
@@ -1,7 +1,9 @@
{
"config": {
"abort": {
- "already_configured": "This device is already configured"
+ "already_configured": "This device has already been configured",
+ "already_in_progress": "This device is already being configured",
+ "not_huawei_lte": "Not a Huawei LTE device"
},
"error": {
"connection_failed": "Connection failed",
diff --git a/homeassistant/components/huawei_lte/config_flow.py b/homeassistant/components/huawei_lte/config_flow.py
index 992dc33a697..1bc3753bdd7 100644
--- a/homeassistant/components/huawei_lte/config_flow.py
+++ b/homeassistant/components/huawei_lte/config_flow.py
@@ -19,6 +19,7 @@ from url_normalize import url_normalize
import voluptuous as vol
from homeassistant import config_entries
+from homeassistant.components.ssdp import ATTR_HOST, ATTR_NAME, ATTR_PRESENTATIONURL
from homeassistant.const import CONF_PASSWORD, CONF_RECIPIENT, CONF_URL, CONF_USERNAME
from homeassistant.core import callback
from .const import CONNECTION_TIMEOUT, DEFAULT_DEVICE_NAME
@@ -52,7 +53,14 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
(
(
vol.Required(
- CONF_URL, default=user_input.get(CONF_URL, "")
+ CONF_URL,
+ default=user_input.get(
+ CONF_URL,
+ # https://github.com/PyCQA/pylint/issues/3167
+ self.context.get( # pylint: disable=no-member
+ CONF_URL, ""
+ ),
+ ),
),
str,
),
@@ -78,6 +86,14 @@ 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):
+ """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(self, user_input=None):
"""Handle user initiated config flow."""
if user_input is None:
@@ -95,12 +111,7 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
user_input=user_input, errors=errors
)
- # See if we already have a router configured with this URL
- existing_urls = { # existing entries
- url_normalize(entry.data[CONF_URL], default_scheme="http")
- for entry in self._async_current_entries()
- }
- if user_input[CONF_URL] in existing_urls:
+ if self._already_configured(user_input):
return self.async_abort(reason="already_configured")
conn = None
@@ -194,6 +205,31 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=title, data=user_input)
+ async def async_step_ssdp(self, discovery_info):
+ """Handle SSDP initiated config flow."""
+ # Attempt to distinguish from other non-LTE Huawei router devices, at least
+ # some ones we are interested in have "Mobile Wi-Fi" friendlyName.
+ if "mobile" not in discovery_info.get(ATTR_NAME, "").lower():
+ return self.async_abort(reason="not_huawei_lte")
+
+ # https://github.com/PyCQA/pylint/issues/3167
+ url = self.context[CONF_URL] = url_normalize( # pylint: disable=no-member
+ discovery_info.get(
+ ATTR_PRESENTATIONURL, f"http://{discovery_info[ATTR_HOST]}/"
+ )
+ )
+
+ if any(
+ url == flow["context"].get(CONF_URL) for flow in self._async_in_progress()
+ ):
+ return self.async_abort(reason="already_in_progress")
+
+ user_input = {CONF_URL: url}
+ if self._already_configured(user_input):
+ return self.async_abort(reason="already_configured")
+
+ return await self._async_show_user_form(user_input)
+
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Huawei LTE options flow."""
diff --git a/homeassistant/components/huawei_lte/manifest.json b/homeassistant/components/huawei_lte/manifest.json
index b3c4442caa9..4ea54188688 100644
--- a/homeassistant/components/huawei_lte/manifest.json
+++ b/homeassistant/components/huawei_lte/manifest.json
@@ -9,6 +9,12 @@
"stringcase==1.2.0",
"url-normalize==1.4.1"
],
+ "ssdp": [
+ {
+ "deviceType": "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
+ "manufacturer": "Huawei"
+ }
+ ],
"dependencies": [],
"codeowners": [
"@scop"
diff --git a/homeassistant/components/huawei_lte/strings.json b/homeassistant/components/huawei_lte/strings.json
index 2e76cf1b343..17684253671 100644
--- a/homeassistant/components/huawei_lte/strings.json
+++ b/homeassistant/components/huawei_lte/strings.json
@@ -1,7 +1,9 @@
{
"config": {
"abort": {
- "already_configured": "This device is already configured"
+ "already_configured": "This device has already been configured",
+ "already_in_progress": "This device is already being configured",
+ "not_huawei_lte": "Not a Huawei LTE device"
},
"error": {
"connection_failed": "Connection failed",
diff --git a/homeassistant/generated/ssdp.py b/homeassistant/generated/ssdp.py
index 472ad6683ed..adf3a345bbe 100644
--- a/homeassistant/generated/ssdp.py
+++ b/homeassistant/generated/ssdp.py
@@ -16,6 +16,12 @@ SSDP = {
"st": "urn:schemas-denon-com:device:ACT-Denon:1"
}
],
+ "huawei_lte": [
+ {
+ "deviceType": "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
+ "manufacturer": "Huawei"
+ }
+ ],
"hue": [
{
"manufacturer": "Royal Philips Electronics"
diff --git a/tests/components/huawei_lte/test_config_flow.py b/tests/components/huawei_lte/test_config_flow.py
index aafa6abd57f..a9f5034fcfe 100644
--- a/tests/components/huawei_lte/test_config_flow.py
+++ b/tests/components/huawei_lte/test_config_flow.py
@@ -10,6 +10,21 @@ from homeassistant import data_entry_flow
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD, CONF_URL
from homeassistant.components.huawei_lte.const import DOMAIN
from homeassistant.components.huawei_lte.config_flow import ConfigFlowHandler
+from homeassistant.components.ssdp import (
+ ATTR_HOST,
+ ATTR_MANUFACTURER,
+ ATTR_MANUFACTURERURL,
+ ATTR_MODEL_NAME,
+ ATTR_MODEL_NUMBER,
+ ATTR_NAME,
+ ATTR_PORT,
+ ATTR_PRESENTATIONURL,
+ ATTR_SERIAL,
+ ATTR_ST,
+ ATTR_UDN,
+ ATTR_UPNP_DEVICE_TYPE,
+)
+
from tests.common import MockConfigEntry
@@ -20,21 +35,26 @@ FIXTURE_USER_INPUT = {
}
-async def test_show_set_form(hass):
- """Test that the setup form is served."""
+@pytest.fixture
+def flow(hass):
+ """Get flow to test."""
flow = ConfigFlowHandler()
flow.hass = hass
+ flow.context = {}
+ return flow
+
+
+async def test_show_set_form(flow):
+ """Test that the setup form is served."""
result = await flow.async_step_user(user_input=None)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
-async def test_urlize_plain_host(hass, requests_mock):
+async def test_urlize_plain_host(flow, requests_mock):
"""Test that plain host or IP gets converted to a URL."""
requests_mock.request(ANY, ANY, exc=ConnectionError())
- flow = ConfigFlowHandler()
- flow.hass = hass
host = "192.168.100.1"
user_input = {**FIXTURE_USER_INPUT, CONF_URL: host}
result = await flow.async_step_user(user_input=user_input)
@@ -44,14 +64,12 @@ 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(flow):
"""Test we reject already configured devices."""
MockConfigEntry(
domain=DOMAIN, data=FIXTURE_USER_INPUT, title="Already configured"
- ).add_to_hass(hass)
+ ).add_to_hass(flow.hass)
- flow = ConfigFlowHandler()
- flow.hass = hass
# Tweak URL a bit to check that doesn't fail duplicate detection
result = await flow.async_step_user(
user_input={
@@ -64,12 +82,10 @@ async def test_already_configured(hass):
assert result["reason"] == "already_configured"
-async def test_connection_error(hass, requests_mock):
+async def test_connection_error(flow, requests_mock):
"""Test we show user form on connection error."""
requests_mock.request(ANY, ANY, exc=ConnectionError())
- flow = ConfigFlowHandler()
- flow.hass = hass
result = await flow.async_step_user(user_input=FIXTURE_USER_INPUT)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
@@ -107,15 +123,13 @@ def login_requests_mock(requests_mock):
(ResponseCodeEnum.ERROR_SYSTEM_UNKNOWN, {"base": "response_error"}),
),
)
-async def test_login_error(hass, login_requests_mock, code, errors):
+async def test_login_error(flow, login_requests_mock, code, errors):
"""Test we show user form with appropriate error on response failure."""
login_requests_mock.request(
ANY,
f"{FIXTURE_USER_INPUT[CONF_URL]}api/user/login",
text=f"<error><code>{code}</code><message/></error>",
)
- flow = ConfigFlowHandler()
- flow.hass = hass
result = await flow.async_step_user(user_input=FIXTURE_USER_INPUT)
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
@@ -123,18 +137,41 @@ async def test_login_error(hass, login_requests_mock, code, errors):
assert result["errors"] == errors
-async def test_success(hass, login_requests_mock):
+async def test_success(flow, login_requests_mock):
"""Test successful flow provides entry creation data."""
login_requests_mock.request(
ANY,
f"{FIXTURE_USER_INPUT[CONF_URL]}api/user/login",
text=f"<response>OK</response>",
)
- flow = ConfigFlowHandler()
- flow.hass = hass
result = await flow.async_step_user(user_input=FIXTURE_USER_INPUT)
assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
assert result["data"][CONF_URL] == FIXTURE_USER_INPUT[CONF_URL]
assert result["data"][CONF_USERNAME] == FIXTURE_USER_INPUT[CONF_USERNAME]
assert result["data"][CONF_PASSWORD] == FIXTURE_USER_INPUT[CONF_PASSWORD]
+
+
+async def test_ssdp(flow):
+ """Test SSDP discovery initiates config properly."""
+ url = "http://192.168.100.1/"
+ result = await flow.async_step_ssdp(
+ discovery_info={
+ ATTR_ST: "upnp:rootdevice",
+ ATTR_PORT: 60957,
+ ATTR_HOST: "192.168.100.1",
+ ATTR_MANUFACTURER: "Huawei",
+ ATTR_MANUFACTURERURL: "http://www.huawei.com/",
+ ATTR_MODEL_NAME: "Huawei router",
+ ATTR_MODEL_NUMBER: "12345678",
+ ATTR_NAME: "Mobile Wi-Fi",
+ ATTR_PRESENTATIONURL: url,
+ ATTR_SERIAL: "00000000",
+ ATTR_UDN: "uuid:XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX",
+ ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
+ }
+ )
+
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert flow.context[CONF_URL] == url
--
GitLab