Fix changed port on Netgear R7000 (#64012)

* fix changed port on Netgear R7000

* fix styling

* update tests

homeassistant/components/netgear/config_flow.py

@@ -23,7 +23,9 @@ from .const import (
     DEFAULT_NAME,
     DOMAIN,
     MODELS_PORT_80,
+    MODELS_PORT_5555,
     PORT_80,
+    PORT_5555,
 )
 from .errors import CannotLoginException
 from .router import get_api
@@ -126,16 +128,17 @@ class NetgearFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
         device_url = urlparse(discovery_info.ssdp_location)
         if device_url.hostname:
             updated_data[CONF_HOST] = device_url.hostname
-        if device_url.scheme == "https":
-            updated_data[CONF_SSL] = True
-        else:
-            updated_data[CONF_SSL] = False
 
         _LOGGER.debug("Netgear ssdp discovery info: %s", discovery_info)
 
         await self.async_set_unique_id(discovery_info.upnp[ssdp.ATTR_UPNP_SERIAL])
         self._abort_if_unique_id_configured(updates=updated_data)
 
+        if device_url.scheme == "https":
+            updated_data[CONF_SSL] = True
+        else:
+            updated_data[CONF_SSL] = False
+
         updated_data[CONF_PORT] = DEFAULT_PORT
         for model in MODELS_PORT_80:
             if discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NUMBER, "").startswith(
@@ -144,6 +147,14 @@ class NetgearFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
                 model
             ):
                 updated_data[CONF_PORT] = PORT_80
+        for model in MODELS_PORT_5555:
+            if discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NUMBER, "").startswith(
+                model
+            ) or discovery_info.upnp.get(ssdp.ATTR_UPNP_MODEL_NAME, "").startswith(
+                model
+            ):
+                updated_data[CONF_PORT] = PORT_5555
+                updated_data[CONF_SSL] = True
 
         self.placeholders.update(updated_data)
         self.discovered = True

homeassistant/components/netgear/const.py

@@ -32,6 +32,10 @@ MODELS_PORT_80 = [
     "SXS",
 ]
 PORT_80 = 80
+MODELS_PORT_5555 = [
+    "R7000",
+]
+PORT_5555 = 5555
 # update method V2 models
 MODELS_V2 = [
     "Orbi",

tests/components/netgear/test_config_flow.py

@@ -6,7 +6,13 @@ import pytest
 
 from homeassistant import data_entry_flow
 from homeassistant.components import ssdp
-from homeassistant.components.netgear.const import CONF_CONSIDER_HOME, DOMAIN, PORT_80
+from homeassistant.components.netgear.const import (
+    CONF_CONSIDER_HOME,
+    DOMAIN,
+    MODELS_PORT_5555,
+    PORT_80,
+    PORT_5555,
+)
 from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
 from homeassistant.const import (
     CONF_HOST,
@@ -19,6 +25,7 @@ from homeassistant.const import (
 from tests.common import MockConfigEntry
 
 URL = "http://routerlogin.net"
+URL_SSL = "https://routerlogin.net"
 SERIAL = "5ER1AL0000001"
 
 ROUTER_INFOS = {
@@ -43,6 +50,7 @@ ROUTER_INFOS = {
     "DeviceModeCapability": "0;1",
 }
 TITLE = f"{ROUTER_INFOS['ModelName']} - {ROUTER_INFOS['DeviceName']}"
+TITLE_INCOMPLETE = ROUTER_INFOS["ModelName"]
 
 HOST = "10.0.0.1"
 SERIAL_2 = "5ER1AL0000002"
@@ -64,6 +72,18 @@ def mock_controller_service():
         yield service_mock
 
 
+@pytest.fixture(name="service_incomplete")
+def mock_controller_service_incomplete():
+    """Mock a successful service."""
+    router_infos = ROUTER_INFOS.copy()
+    router_infos.pop("DeviceName")
+    with patch(
+        "homeassistant.components.netgear.async_setup_entry", return_value=True
+    ), patch("homeassistant.components.netgear.router.Netgear") as service_mock:
+        service_mock.return_value.get_info = Mock(return_value=router_infos)
+        yield service_mock
+
+
 @pytest.fixture(name="service_failed")
 def mock_controller_service_failed():
     """Mock a failed service."""
@@ -102,6 +122,59 @@ async def test_user(hass, service):
     assert result["data"][CONF_PASSWORD] == PASSWORD
 
 
+async def test_user_connect_error(hass, service_failed):
+    """Test user step with connection failure."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == "user"
+
+    # Have to provide all config
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {
+            CONF_HOST: HOST,
+            CONF_PORT: PORT,
+            CONF_SSL: SSL,
+            CONF_USERNAME: USERNAME,
+            CONF_PASSWORD: PASSWORD,
+        },
+    )
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {"base": "config"}
+
+
+async def test_user_incomplete_info(hass, service_incomplete):
+    """Test user step with incomplete device info."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == "user"
+
+    # Have to provide all config
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {
+            CONF_HOST: HOST,
+            CONF_PORT: PORT,
+            CONF_SSL: SSL,
+            CONF_USERNAME: USERNAME,
+            CONF_PASSWORD: PASSWORD,
+        },
+    )
+    assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result["result"].unique_id == SERIAL
+    assert result["title"] == TITLE_INCOMPLETE
+    assert result["data"].get(CONF_HOST) == HOST
+    assert result["data"].get(CONF_PORT) == PORT
+    assert result["data"].get(CONF_SSL) == SSL
+    assert result["data"].get(CONF_USERNAME) == USERNAME
+    assert result["data"][CONF_PASSWORD] == PASSWORD
+
+
 async def test_abort_if_already_setup(hass, service):
     """Test we abort if the router is already setup."""
     MockConfigEntry(
@@ -183,6 +256,38 @@ async def test_ssdp(hass, service):
     assert result["data"][CONF_PASSWORD] == PASSWORD
 
 
+async def test_ssdp_port_5555(hass, service):
+    """Test ssdp step with port 5555."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_SSDP},
+        data=ssdp.SsdpServiceInfo(
+            ssdp_usn="mock_usn",
+            ssdp_st="mock_st",
+            ssdp_location=SSDP_URL_SLL,
+            upnp={
+                ssdp.ATTR_UPNP_MODEL_NUMBER: MODELS_PORT_5555[0],
+                ssdp.ATTR_UPNP_PRESENTATION_URL: URL_SSL,
+                ssdp.ATTR_UPNP_SERIAL: SERIAL,
+            },
+        ),
+    )
+    assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"], {CONF_PASSWORD: PASSWORD}
+    )
+    assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+    assert result["result"].unique_id == SERIAL
+    assert result["title"] == TITLE
+    assert result["data"].get(CONF_HOST) == HOST
+    assert result["data"].get(CONF_PORT) == PORT_5555
+    assert result["data"].get(CONF_SSL) is True
+    assert result["data"].get(CONF_USERNAME) == DEFAULT_USER
+    assert result["data"][CONF_PASSWORD] == PASSWORD
+
+
 async def test_options_flow(hass, service):
     """Test specifying non default settings using options flow."""
     config_entry = MockConfigEntry(