Improve Huawei LTE SSDP inclusion (#85572)
* Probe Huawei LTE API for device support on SSDP match More or less as expected, the loosening of SSDP/UPnP data matches done in #81643 started to yield false positives, as in #85402. Coming up with robust matches solely based on the SSDP/UPnP data still does not seem possible, so keep the matches as loose as they were made, but additionally invoke a probe request on the API to determine if the device looks like a supported one. * Probe only after unique id checks Prevents throwaway probes for discoveries already in progress. * Fix SSDP result URL test, add missing assert on itpull/85788/head
parent
255a8362a1
commit
c625051665
|
@ -250,6 +250,24 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||||
await self.async_set_unique_id(unique_id)
|
await self.async_set_unique_id(unique_id)
|
||||||
self._abort_if_unique_id_configured(updates={CONF_URL: url})
|
self._abort_if_unique_id_configured(updates={CONF_URL: url})
|
||||||
|
|
||||||
|
def _is_supported_device() -> bool:
|
||||||
|
"""
|
||||||
|
See if we are looking at a possibly supported device.
|
||||||
|
|
||||||
|
Matching solely on SSDP data does not yield reliable enough results.
|
||||||
|
"""
|
||||||
|
try:
|
||||||
|
with Connection(url=url, timeout=CONNECTION_TIMEOUT) as conn:
|
||||||
|
basic_info = Client(conn).device.basic_information()
|
||||||
|
except ResponseErrorException: # API compatible error
|
||||||
|
return True
|
||||||
|
except Exception: # API incompatible error # pylint: disable=broad-except
|
||||||
|
return False
|
||||||
|
return isinstance(basic_info, dict) # Crude content check
|
||||||
|
|
||||||
|
if not await self.hass.async_add_executor_job(_is_supported_device):
|
||||||
|
return self.async_abort(reason="unsupported_device")
|
||||||
|
|
||||||
self.context.update(
|
self.context.update(
|
||||||
{
|
{
|
||||||
"title_placeholders": {
|
"title_placeholders": {
|
||||||
|
|
|
@ -1,7 +1,8 @@
|
||||||
{
|
{
|
||||||
"config": {
|
"config": {
|
||||||
"abort": {
|
"abort": {
|
||||||
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
|
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
|
||||||
|
"unsupported_device": "Unsupported device"
|
||||||
},
|
},
|
||||||
"error": {
|
"error": {
|
||||||
"connection_timeout": "Connection timeout",
|
"connection_timeout": "Connection timeout",
|
||||||
|
|
|
@ -1,8 +1,8 @@
|
||||||
{
|
{
|
||||||
"config": {
|
"config": {
|
||||||
"abort": {
|
"abort": {
|
||||||
"not_huawei_lte": "Not a Huawei LTE device",
|
"reauth_successful": "Re-authentication was successful",
|
||||||
"reauth_successful": "Re-authentication was successful"
|
"unsupported_device": "Unsupported device"
|
||||||
},
|
},
|
||||||
"error": {
|
"error": {
|
||||||
"connection_timeout": "Connection timeout",
|
"connection_timeout": "Connection timeout",
|
||||||
|
|
|
@ -211,9 +211,14 @@ async def test_success(hass, login_requests_mock):
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
("upnp_data", "expected_result"),
|
("requests_mock_request_kwargs", "upnp_data", "expected_result"),
|
||||||
(
|
(
|
||||||
(
|
(
|
||||||
|
{
|
||||||
|
"method": ANY,
|
||||||
|
"url": f"{FIXTURE_USER_INPUT[CONF_URL]}api/device/basic_information",
|
||||||
|
"text": "<response><devicename>Mock device</devicename></response>",
|
||||||
|
},
|
||||||
{
|
{
|
||||||
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Mobile Wi-Fi",
|
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Mobile Wi-Fi",
|
||||||
ssdp.ATTR_UPNP_SERIAL: "00000000",
|
ssdp.ATTR_UPNP_SERIAL: "00000000",
|
||||||
|
@ -225,6 +230,11 @@ async def test_success(hass, login_requests_mock):
|
||||||
},
|
},
|
||||||
),
|
),
|
||||||
(
|
(
|
||||||
|
{
|
||||||
|
"method": ANY,
|
||||||
|
"url": f"{FIXTURE_USER_INPUT[CONF_URL]}api/device/basic_information",
|
||||||
|
"text": "<error><code>100002</code><message/></error>",
|
||||||
|
},
|
||||||
{
|
{
|
||||||
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Mobile Wi-Fi",
|
ssdp.ATTR_UPNP_FRIENDLY_NAME: "Mobile Wi-Fi",
|
||||||
# No ssdp.ATTR_UPNP_SERIAL
|
# No ssdp.ATTR_UPNP_SERIAL
|
||||||
|
@ -235,19 +245,36 @@ async def test_success(hass, login_requests_mock):
|
||||||
"errors": {},
|
"errors": {},
|
||||||
},
|
},
|
||||||
),
|
),
|
||||||
|
(
|
||||||
|
{
|
||||||
|
"method": ANY,
|
||||||
|
"url": f"{FIXTURE_USER_INPUT[CONF_URL]}api/device/basic_information",
|
||||||
|
"exc": Exception("Something unexpected"),
|
||||||
|
},
|
||||||
|
{
|
||||||
|
# Does not matter
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"type": data_entry_flow.FlowResultType.ABORT,
|
||||||
|
"reason": "unsupported_device",
|
||||||
|
},
|
||||||
|
),
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
async def test_ssdp(hass, upnp_data, expected_result):
|
async def test_ssdp(
|
||||||
|
hass, login_requests_mock, requests_mock_request_kwargs, upnp_data, expected_result
|
||||||
|
):
|
||||||
"""Test SSDP discovery initiates config properly."""
|
"""Test SSDP discovery initiates config properly."""
|
||||||
url = "http://192.168.100.1/"
|
url = FIXTURE_USER_INPUT[CONF_URL][:-1] # strip trailing slash for appending port
|
||||||
context = {"source": config_entries.SOURCE_SSDP}
|
context = {"source": config_entries.SOURCE_SSDP}
|
||||||
|
login_requests_mock.request(**requests_mock_request_kwargs)
|
||||||
result = await hass.config_entries.flow.async_init(
|
result = await hass.config_entries.flow.async_init(
|
||||||
DOMAIN,
|
DOMAIN,
|
||||||
context=context,
|
context=context,
|
||||||
data=ssdp.SsdpServiceInfo(
|
data=ssdp.SsdpServiceInfo(
|
||||||
ssdp_usn="mock_usn",
|
ssdp_usn="mock_usn",
|
||||||
ssdp_st="upnp:rootdevice",
|
ssdp_st="upnp:rootdevice",
|
||||||
ssdp_location="http://192.168.100.1:60957/rootDesc.xml",
|
ssdp_location=f"{url}:60957/rootDesc.xml",
|
||||||
upnp={
|
upnp={
|
||||||
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
|
ssdp.ATTR_UPNP_DEVICE_TYPE: "urn:schemas-upnp-org:device:InternetGatewayDevice:1",
|
||||||
ssdp.ATTR_UPNP_MANUFACTURER: "Huawei",
|
ssdp.ATTR_UPNP_MANUFACTURER: "Huawei",
|
||||||
|
@ -264,7 +291,7 @@ async def test_ssdp(hass, upnp_data, expected_result):
|
||||||
for k, v in expected_result.items():
|
for k, v in expected_result.items():
|
||||||
assert result[k] == v
|
assert result[k] == v
|
||||||
if result.get("data_schema"):
|
if result.get("data_schema"):
|
||||||
result["data_schema"]({})[CONF_URL] == url
|
assert result["data_schema"]({})[CONF_URL] == url + "/"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.parametrize(
|
@pytest.mark.parametrize(
|
||||||
|
|
Loading…
Reference in New Issue