evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Support Huawei LTE SSDP discovery (#28214)

Browse source 
* 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
This commit is contained in:
Ville Skyttä 2019-11-04 19:56:49 +02:00 • committed by GitHub
parent f3ea44cd92
commit 6a7b5657ac
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 116 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
homeassistant/components/huawei_lte/.translations/en.json
View file

@ -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",

50
homeassistant/components/huawei_lte/config_flow.py
View file

@ -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."""

6
homeassistant/components/huawei_lte/manifest.json
View file

@ -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"

4
homeassistant/components/huawei_lte/strings.json
View file

@ -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",

6
homeassistant/generated/ssdp.py
View file

@ -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"

73
tests/components/huawei_lte/test_config_flow.py
View file

@ -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