Support Huawei LTE SSDP discovery

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

homeassistant/components/huawei_lte/config_flow.py

@@ -19,6 +19,7 @@
 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 @@ async def _async_show_user_form(self, user_input=None, errors=None):
                     (
                         (
                             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 @@ async def async_step_import(self, user_input=None):
         """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 @@ async def async_step_user(self, user_input=None):
                 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 @@ def get_router_title(conn: Connection) -> str:
 
         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."""

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"

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

homeassistant/generated/ssdp.py

@@ -16,6 +16,12 @@
             "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"

tests/components/huawei_lte/test_config_flow.py

@@ -10,6 +10,21 @@
 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 @@
 }
 
 
-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,34 +123,56 @@ 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
     assert result["step_id"] == "user"
     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",
+        }
+    )
+
+    print(result)
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == "user"
+    assert flow.context[CONF_URL] == url