From 476a59d248df5f2f933affd70d42af4c8bf1175e Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Tue, 16 Nov 2021 12:19:50 +0100 Subject: [PATCH] Adjust async_step_dhcp signature for strict typing (#59751) Co-authored-by: epenet --- homeassistant/components/axis/config_flow.py | 11 ++++---- homeassistant/components/dhcp/__init__.py | 25 +++++++++++++------ .../components/flux_led/config_flow.py | 10 ++++---- .../components/goalzero/config_flow.py | 11 ++++---- .../components/guardian/config_flow.py | 8 +++--- .../components/samsungtv/config_flow.py | 9 +++---- .../components/tplink/config_flow.py | 6 ++--- homeassistant/config_entries.py | 5 ++-- homeassistant/helpers/config_entry_flow.py | 12 +++++++-- 9 files changed, 55 insertions(+), 42 deletions(-) diff --git a/homeassistant/components/axis/config_flow.py b/homeassistant/components/axis/config_flow.py index d1e834e7bb6..eb4fd9296fc 100644 --- a/homeassistant/components/axis/config_flow.py +++ b/homeassistant/components/axis/config_flow.py @@ -6,8 +6,7 @@ from urllib.parse import urlsplit import voluptuous as vol from homeassistant import config_entries -from homeassistant.components import zeroconf -from homeassistant.components.dhcp import HOSTNAME, IP_ADDRESS, MAC_ADDRESS +from homeassistant.components import dhcp, zeroconf from homeassistant.config_entries import SOURCE_IGNORE from homeassistant.const import ( CONF_HOST, @@ -153,13 +152,13 @@ class AxisFlowHandler(config_entries.ConfigFlow, domain=AXIS_DOMAIN): return await self.async_step_user() - async def async_step_dhcp(self, discovery_info: dict): + async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult: """Prepare configuration for a DHCP discovered Axis device.""" return await self._process_discovered_device( { - CONF_HOST: discovery_info[IP_ADDRESS], - CONF_MAC: format_mac(discovery_info.get(MAC_ADDRESS, "")), - CONF_NAME: discovery_info.get(HOSTNAME), + CONF_HOST: discovery_info[dhcp.IP_ADDRESS], + CONF_MAC: format_mac(discovery_info[dhcp.MAC_ADDRESS]), + CONF_NAME: discovery_info[dhcp.HOSTNAME], CONF_PORT: DEFAULT_PORT, } ) diff --git a/homeassistant/components/dhcp/__init__.py b/homeassistant/components/dhcp/__init__.py index d52b30ccfb2..be6aa5072d8 100644 --- a/homeassistant/components/dhcp/__init__.py +++ b/homeassistant/components/dhcp/__init__.py @@ -6,6 +6,7 @@ from ipaddress import ip_address as make_ip_address import logging import os import threading +from typing import Final, TypedDict from aiodiscover import DiscoverHosts from aiodiscover.discovery import ( @@ -45,15 +46,23 @@ from homeassistant.util.network import is_invalid, is_link_local, is_loopback FILTER = "udp and (port 67 or 68)" REQUESTED_ADDR = "requested_addr" MESSAGE_TYPE = "message-type" -HOSTNAME = "hostname" -MAC_ADDRESS = "macaddress" -IP_ADDRESS = "ip" +HOSTNAME: Final = "hostname" +MAC_ADDRESS: Final = "macaddress" +IP_ADDRESS: Final = "ip" DHCP_REQUEST = 3 SCAN_INTERVAL = timedelta(minutes=60) _LOGGER = logging.getLogger(__name__) +class DhcpServiceInfo(TypedDict): + """Prepared info from dhcp entries.""" + + ip: str + hostname: str + macaddress: str + + async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: """Set up the dhcp component.""" @@ -150,11 +159,11 @@ class WatcherBase: self.hass, entry["domain"], {"source": config_entries.SOURCE_DHCP}, - { - IP_ADDRESS: ip_address, - HOSTNAME: lowercase_hostname, - MAC_ADDRESS: data[MAC_ADDRESS], - }, + DhcpServiceInfo( + ip=ip_address, + hostname=lowercase_hostname, + macaddress=data[MAC_ADDRESS], + ), ) diff --git a/homeassistant/components/flux_led/config_flow.py b/homeassistant/components/flux_led/config_flow.py index 306dbc2c25e..0a059abaf34 100644 --- a/homeassistant/components/flux_led/config_flow.py +++ b/homeassistant/components/flux_led/config_flow.py @@ -7,7 +7,7 @@ from typing import Any, Final import voluptuous as vol from homeassistant import config_entries -from homeassistant.components.dhcp import HOSTNAME, IP_ADDRESS, MAC_ADDRESS +from homeassistant.components import dhcp from homeassistant.const import CONF_HOST, CONF_MAC, CONF_MODE, CONF_NAME, CONF_PROTOCOL from homeassistant.core import callback from homeassistant.data_entry_flow import FlowResult @@ -82,12 +82,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): }, ) - async def async_step_dhcp(self, discovery_info: DiscoveryInfoType) -> FlowResult: + async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult: """Handle discovery via dhcp.""" self._discovered_device = { - FLUX_HOST: discovery_info[IP_ADDRESS], - FLUX_MODEL: discovery_info[HOSTNAME], - FLUX_MAC: discovery_info[MAC_ADDRESS].replace(":", ""), + FLUX_HOST: discovery_info[dhcp.IP_ADDRESS], + FLUX_MODEL: discovery_info[dhcp.HOSTNAME], + FLUX_MAC: discovery_info[dhcp.MAC_ADDRESS].replace(":", ""), } return await self._async_handle_discovery() diff --git a/homeassistant/components/goalzero/config_flow.py b/homeassistant/components/goalzero/config_flow.py index f192c71cbf8..4b02fe1e83a 100644 --- a/homeassistant/components/goalzero/config_flow.py +++ b/homeassistant/components/goalzero/config_flow.py @@ -8,12 +8,11 @@ from goalzero import Yeti, exceptions import voluptuous as vol from homeassistant import config_entries -from homeassistant.components.dhcp import IP_ADDRESS, MAC_ADDRESS +from homeassistant.components import dhcp from homeassistant.const import CONF_HOST, CONF_NAME from homeassistant.data_entry_flow import FlowResult from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.device_registry import format_mac -from homeassistant.helpers.typing import DiscoveryInfoType from .const import DEFAULT_NAME, DOMAIN @@ -27,13 +26,13 @@ class GoalZeroFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): def __init__(self) -> None: """Initialize a Goal Zero Yeti flow.""" - self.ip_address = None + self.ip_address: str | None = None - async def async_step_dhcp(self, discovery_info: DiscoveryInfoType) -> FlowResult: + async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult: """Handle dhcp discovery.""" - self.ip_address = discovery_info[IP_ADDRESS] + self.ip_address = discovery_info[dhcp.IP_ADDRESS] - await self.async_set_unique_id(discovery_info[MAC_ADDRESS]) + await self.async_set_unique_id(discovery_info[dhcp.MAC_ADDRESS]) self._abort_if_unique_id_configured(updates={CONF_HOST: self.ip_address}) self._async_abort_entries_match({CONF_HOST: self.ip_address}) diff --git a/homeassistant/components/guardian/config_flow.py b/homeassistant/components/guardian/config_flow.py index 2aa00f56d99..78a4cbd90ea 100644 --- a/homeassistant/components/guardian/config_flow.py +++ b/homeassistant/components/guardian/config_flow.py @@ -8,12 +8,10 @@ from aioguardian.errors import GuardianError import voluptuous as vol from homeassistant import config_entries -from homeassistant.components import zeroconf -from homeassistant.components.dhcp import IP_ADDRESS +from homeassistant.components import dhcp, zeroconf from homeassistant.const import CONF_IP_ADDRESS, CONF_PORT from homeassistant.core import HomeAssistant, callback from homeassistant.data_entry_flow import FlowResult -from homeassistant.helpers.typing import DiscoveryInfoType from .const import CONF_UID, DOMAIN, LOGGER @@ -99,10 +97,10 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): title=info[CONF_UID], data={CONF_UID: info["uid"], **user_input} ) - async def async_step_dhcp(self, discovery_info: DiscoveryInfoType) -> FlowResult: + async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult: """Handle the configuration via dhcp.""" self.discovery_info = { - CONF_IP_ADDRESS: discovery_info[IP_ADDRESS], + CONF_IP_ADDRESS: discovery_info[dhcp.IP_ADDRESS], CONF_PORT: DEFAULT_PORT, } return await self._async_handle_discovery() diff --git a/homeassistant/components/samsungtv/config_flow.py b/homeassistant/components/samsungtv/config_flow.py index 1525d037cd9..f476d1dc87d 100644 --- a/homeassistant/components/samsungtv/config_flow.py +++ b/homeassistant/components/samsungtv/config_flow.py @@ -10,8 +10,7 @@ import getmac import voluptuous as vol from homeassistant import config_entries, data_entry_flow -from homeassistant.components import zeroconf -from homeassistant.components.dhcp import IP_ADDRESS, MAC_ADDRESS +from homeassistant.components import dhcp, zeroconf from homeassistant.components.ssdp import ( ATTR_SSDP_LOCATION, ATTR_UPNP_MANUFACTURER, @@ -285,12 +284,12 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return await self.async_step_confirm() async def async_step_dhcp( - self, discovery_info: DiscoveryInfoType + self, discovery_info: dhcp.DhcpServiceInfo ) -> data_entry_flow.FlowResult: """Handle a flow initialized by dhcp discovery.""" LOGGER.debug("Samsung device found via DHCP: %s", discovery_info) - self._mac = discovery_info[MAC_ADDRESS] - self._host = discovery_info[IP_ADDRESS] + self._mac = discovery_info[dhcp.MAC_ADDRESS] + self._host = discovery_info[dhcp.IP_ADDRESS] await self._async_start_discovery_with_mac_address() await self._async_set_device_unique_id() self.context["title_placeholders"] = {"device": self._title} diff --git a/homeassistant/components/tplink/config_flow.py b/homeassistant/components/tplink/config_flow.py index 8abc00d7fdb..f07d8887b85 100644 --- a/homeassistant/components/tplink/config_flow.py +++ b/homeassistant/components/tplink/config_flow.py @@ -9,7 +9,7 @@ from kasa.discover import Discover import voluptuous as vol from homeassistant import config_entries -from homeassistant.components.dhcp import IP_ADDRESS, MAC_ADDRESS +from homeassistant.components import dhcp from homeassistant.const import CONF_DEVICE, CONF_HOST, CONF_MAC, CONF_NAME from homeassistant.core import callback from homeassistant.data_entry_flow import FlowResult @@ -32,10 +32,10 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): self._discovered_devices: dict[str, SmartDevice] = {} self._discovered_device: SmartDevice | None = None - async def async_step_dhcp(self, discovery_info: DiscoveryInfoType) -> FlowResult: + async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult: """Handle discovery via dhcp.""" return await self._async_handle_discovery( - discovery_info[IP_ADDRESS], discovery_info[MAC_ADDRESS] + discovery_info[dhcp.IP_ADDRESS], discovery_info[dhcp.MAC_ADDRESS] ) async def async_step_discovery( diff --git a/homeassistant/config_entries.py b/homeassistant/config_entries.py index d9c0faf2f71..85215a5962f 100644 --- a/homeassistant/config_entries.py +++ b/homeassistant/config_entries.py @@ -32,6 +32,7 @@ from homeassistant.util.decorator import Registry import homeassistant.util.uuid as uuid_util if TYPE_CHECKING: + from homeassistant.components.dhcp import DhcpServiceInfo from homeassistant.components.zeroconf import ZeroconfServiceInfo _LOGGER = logging.getLogger(__name__) @@ -1378,10 +1379,10 @@ class ConfigFlow(data_entry_flow.FlowHandler): return await self.async_step_discovery(cast(dict, discovery_info)) async def async_step_dhcp( - self, discovery_info: DiscoveryInfoType + self, discovery_info: DhcpServiceInfo ) -> data_entry_flow.FlowResult: """Handle a flow initialized by DHCP discovery.""" - return await self.async_step_discovery(discovery_info) + return await self.async_step_discovery(cast(dict, discovery_info)) async def async_step_usb( self, discovery_info: DiscoveryInfoType diff --git a/homeassistant/helpers/config_entry_flow.py b/homeassistant/helpers/config_entry_flow.py index 4a312b5e01f..75e3d128435 100644 --- a/homeassistant/helpers/config_entry_flow.py +++ b/homeassistant/helpers/config_entry_flow.py @@ -5,7 +5,7 @@ import logging from typing import Any, Awaitable, Callable, Union from homeassistant import config_entries -from homeassistant.components import zeroconf +from homeassistant.components import dhcp, zeroconf from homeassistant.core import HomeAssistant from homeassistant.data_entry_flow import FlowResult from homeassistant.helpers.typing import UNDEFINED, DiscoveryInfoType, UndefinedType @@ -82,6 +82,15 @@ class DiscoveryFlowHandler(config_entries.ConfigFlow): return await self.async_step_confirm() + async def async_step_dhcp(self, discovery_info: dhcp.DhcpServiceInfo) -> FlowResult: + """Handle a flow initialized by dhcp discovery.""" + if self._async_in_progress() or self._async_current_entries(): + return self.async_abort(reason="single_instance_allowed") + + await self.async_set_unique_id(self._domain) + + return await self.async_step_confirm() + async def async_step_homekit( self, discovery_info: zeroconf.ZeroconfServiceInfo ) -> FlowResult: @@ -106,7 +115,6 @@ class DiscoveryFlowHandler(config_entries.ConfigFlow): async_step_ssdp = async_step_discovery async_step_mqtt = async_step_discovery - async_step_dhcp = async_step_discovery async def async_step_import(self, _: dict[str, Any] | None) -> FlowResult: """Handle a flow initialized by import."""