Improve type hints in fritz config flow (#130511)
* Improve type hints in fritz config flow * Improve coverage * Apply suggestions from code review Co-authored-by: Michael <35783820+mib1185@users.noreply.github.com> --------- Co-authored-by: Michael <35783820+mib1185@users.noreply.github.com>
This commit is contained in:
parent
ab11b84678
commit
8300afc00d
2 changed files with 30 additions and 8 deletions
|
@ -57,6 +57,8 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN):
|
||||||
|
|
||||||
VERSION = 1
|
VERSION = 1
|
||||||
|
|
||||||
|
_host: str
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
@callback
|
@callback
|
||||||
def async_get_options_flow(
|
def async_get_options_flow(
|
||||||
|
@ -67,7 +69,6 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN):
|
||||||
|
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
"""Initialize FRITZ!Box Tools flow."""
|
"""Initialize FRITZ!Box Tools flow."""
|
||||||
self._host: str | None = None
|
|
||||||
self._name: str = ""
|
self._name: str = ""
|
||||||
self._password: str = ""
|
self._password: str = ""
|
||||||
self._use_tls: bool = False
|
self._use_tls: bool = False
|
||||||
|
@ -112,7 +113,6 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN):
|
||||||
|
|
||||||
async def async_check_configured_entry(self) -> ConfigEntry | None:
|
async def async_check_configured_entry(self) -> ConfigEntry | None:
|
||||||
"""Check if entry is configured."""
|
"""Check if entry is configured."""
|
||||||
assert self._host
|
|
||||||
current_host = await self.hass.async_add_executor_job(
|
current_host = await self.hass.async_add_executor_job(
|
||||||
socket.gethostbyname, self._host
|
socket.gethostbyname, self._host
|
||||||
)
|
)
|
||||||
|
@ -154,15 +154,17 @@ class FritzBoxToolsFlowHandler(ConfigFlow, domain=DOMAIN):
|
||||||
) -> ConfigFlowResult:
|
) -> ConfigFlowResult:
|
||||||
"""Handle a flow initialized by discovery."""
|
"""Handle a flow initialized by discovery."""
|
||||||
ssdp_location: ParseResult = urlparse(discovery_info.ssdp_location or "")
|
ssdp_location: ParseResult = urlparse(discovery_info.ssdp_location or "")
|
||||||
self._host = ssdp_location.hostname
|
host = ssdp_location.hostname
|
||||||
|
if not host or ipaddress.ip_address(host).is_link_local:
|
||||||
|
return self.async_abort(reason="ignore_ip6_link_local")
|
||||||
|
|
||||||
|
self._host = host
|
||||||
self._name = (
|
self._name = (
|
||||||
discovery_info.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
|
discovery_info.upnp.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
|
||||||
or discovery_info.upnp[ssdp.ATTR_UPNP_MODEL_NAME]
|
or discovery_info.upnp[ssdp.ATTR_UPNP_MODEL_NAME]
|
||||||
)
|
)
|
||||||
|
|
||||||
if not self._host or ipaddress.ip_address(self._host).is_link_local:
|
uuid: str | None
|
||||||
return self.async_abort(reason="ignore_ip6_link_local")
|
|
||||||
|
|
||||||
if uuid := discovery_info.upnp.get(ssdp.ATTR_UPNP_UDN):
|
if uuid := discovery_info.upnp.get(ssdp.ATTR_UPNP_UDN):
|
||||||
if uuid.startswith("uuid:"):
|
if uuid.startswith("uuid:"):
|
||||||
uuid = uuid[5:]
|
uuid = uuid[5:]
|
||||||
|
|
|
@ -10,6 +10,7 @@ from fritzconnection.core.exceptions import (
|
||||||
)
|
)
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
from homeassistant.components import ssdp
|
||||||
from homeassistant.components.device_tracker import (
|
from homeassistant.components.device_tracker import (
|
||||||
CONF_CONSIDER_HOME,
|
CONF_CONSIDER_HOME,
|
||||||
DEFAULT_CONSIDER_HOME,
|
DEFAULT_CONSIDER_HOME,
|
||||||
|
@ -22,7 +23,6 @@ from homeassistant.components.fritz.const import (
|
||||||
ERROR_UNKNOWN,
|
ERROR_UNKNOWN,
|
||||||
FRITZ_AUTH_EXCEPTIONS,
|
FRITZ_AUTH_EXCEPTIONS,
|
||||||
)
|
)
|
||||||
from homeassistant.components.ssdp import ATTR_UPNP_UDN
|
|
||||||
from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
|
from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
|
||||||
from homeassistant.const import (
|
from homeassistant.const import (
|
||||||
CONF_HOST,
|
CONF_HOST,
|
||||||
|
@ -644,7 +644,7 @@ async def test_ssdp_already_in_progress_host(
|
||||||
|
|
||||||
MOCK_NO_UNIQUE_ID = dataclasses.replace(MOCK_SSDP_DATA)
|
MOCK_NO_UNIQUE_ID = dataclasses.replace(MOCK_SSDP_DATA)
|
||||||
MOCK_NO_UNIQUE_ID.upnp = MOCK_NO_UNIQUE_ID.upnp.copy()
|
MOCK_NO_UNIQUE_ID.upnp = MOCK_NO_UNIQUE_ID.upnp.copy()
|
||||||
del MOCK_NO_UNIQUE_ID.upnp[ATTR_UPNP_UDN]
|
del MOCK_NO_UNIQUE_ID.upnp[ssdp.ATTR_UPNP_UDN]
|
||||||
result = await hass.config_entries.flow.async_init(
|
result = await hass.config_entries.flow.async_init(
|
||||||
DOMAIN, context={"source": SOURCE_SSDP}, data=MOCK_NO_UNIQUE_ID
|
DOMAIN, context={"source": SOURCE_SSDP}, data=MOCK_NO_UNIQUE_ID
|
||||||
)
|
)
|
||||||
|
@ -737,3 +737,23 @@ async def test_options_flow(hass: HomeAssistant) -> None:
|
||||||
CONF_OLD_DISCOVERY: False,
|
CONF_OLD_DISCOVERY: False,
|
||||||
CONF_CONSIDER_HOME: 37,
|
CONF_CONSIDER_HOME: 37,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
async def test_ssdp_ipv6_link_local(hass: HomeAssistant) -> None:
|
||||||
|
"""Test ignoring ipv6-link-local while ssdp discovery."""
|
||||||
|
|
||||||
|
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="https://[fe80::1ff:fe23:4567:890a]:12345/test",
|
||||||
|
upnp={
|
||||||
|
ssdp.ATTR_UPNP_FRIENDLY_NAME: "fake_name",
|
||||||
|
ssdp.ATTR_UPNP_UDN: "uuid:only-a-test",
|
||||||
|
},
|
||||||
|
),
|
||||||
|
)
|
||||||
|
assert result["type"] is FlowResultType.ABORT
|
||||||
|
assert result["reason"] == "ignore_ip6_link_local"
|
||||||
|
|
Loading…
Add table
Reference in a new issue