Use dataclass for ZeroconfServiceInfo

homeassistant/components/bosch_shc/config_flow.py

@@ -186,7 +186,7 @@ async def async_step_zeroconf(
         self, discovery_info: zeroconf.ZeroconfServiceInfo
     ) -> FlowResult:
         """Handle zeroconf discovery."""
-        if not discovery_info.get(zeroconf.ATTR_NAME, "").startswith("Bosch SHC"):
+        if not discovery_info.name.startswith("Bosch SHC"):
             return self.async_abort(reason="not_bosch_shc")
 
         try:

homeassistant/components/modern_forms/config_flow.py

@@ -1,7 +1,7 @@
 """Config flow for Modern Forms."""
 from __future__ import annotations
 
-from typing import Any, cast
+from typing import Any
 
 from aiomodernforms import ModernFormsConnectionError, ModernFormsDevice
 import voluptuous as vol
@@ -43,7 +43,7 @@ async def async_step_zeroconf(
         )
 
         # Prepare configuration flow
-        return await self._handle_config_flow(cast(dict, discovery_info), True)
+        return await self._handle_config_flow({}, True)
 
     async def async_step_zeroconf_confirm(
         self, user_input: dict[str, Any] | None = None

homeassistant/components/nut/config_flow.py

@@ -107,7 +107,7 @@ async def async_step_user(self, user_input=None):
                 user_input.update(
                     {
                         CONF_HOST: self.discovery_info[CONF_HOST],
-                        CONF_PORT: self.discovery_info.get(CONF_PORT, DEFAULT_PORT),
+                        CONF_PORT: self.discovery_info.port or DEFAULT_PORT,
                     }
                 )
             info, errors = await self._async_validate_or_error(user_input)

homeassistant/components/shelly/config_flow.py

@@ -201,9 +201,7 @@ async def async_step_zeroconf(
         self._abort_if_unique_id_configured({CONF_HOST: host})
         self.host = host
 
-        self.context["title_placeholders"] = {
-            "name": discovery_info.get("name", "").split(".")[0]
-        }
+        self.context["title_placeholders"] = {"name": discovery_info.name.split(".")[0]}
 
         if get_info_auth(self.info):
             return await self.async_step_credentials()

homeassistant/components/wled/config_flow.py

@@ -1,7 +1,7 @@
 """Config flow to configure the WLED integration."""
 from __future__ import annotations
 
-from typing import Any, cast
+from typing import Any
 
 import voluptuous as vol
 from wled import WLED, WLEDConnectionError
@@ -57,7 +57,7 @@ async def async_step_zeroconf(
         )
 
         # Prepare configuration flow
-        return await self._handle_config_flow(cast(dict, discovery_info), True)
+        return await self._handle_config_flow({}, True)
 
     async def async_step_zeroconf_confirm(
         self, user_input: dict[str, Any] | None = None

homeassistant/components/zeroconf/__init__.py

@@ -3,12 +3,13 @@
 
 import asyncio
 from contextlib import suppress
+from dataclasses import dataclass
 import fnmatch
 from ipaddress import IPv4Address, IPv6Address, ip_address
 import logging
 import socket
 import sys
-from typing import Any, Final, TypedDict, cast
+from typing import Any, Final, cast
 
 import voluptuous as vol
 from zeroconf import InterfaceChoice, IPVersion, ServiceStateChange
@@ -25,8 +26,10 @@
     __version__,
 )
 from homeassistant.core import Event, HomeAssistant, callback
+from homeassistant.data_entry_flow import BaseServiceInfo
 from homeassistant.helpers import discovery_flow
 import homeassistant.helpers.config_validation as cv
+from homeassistant.helpers.frame import report
 from homeassistant.helpers.network import NoURLAvailableError, get_url
 from homeassistant.helpers.typing import ConfigType
 from homeassistant.loader import async_get_homekit, async_get_zeroconf, bind_hass
@@ -89,7 +92,8 @@
 )
 
 
-class ZeroconfServiceInfo(TypedDict):
+@dataclass
+class ZeroconfServiceInfo(BaseServiceInfo):
     """Prepared info from mDNS entries."""
 
     host: str
@@ -99,6 +103,25 @@ class ZeroconfServiceInfo(TypedDict):
     name: str
     properties: dict[str, Any]
 
+    # Used to prevent log flooding. To be removed in 2022.6
+    _warning_logged: bool = False
+
+    def __getitem__(self, name: str) -> Any:
+        """
+        Allow property access by name for compatibility reason.
+
+        Deprecated, and will be removed in version 2022.6.
+        """
+        if not self._warning_logged:
+            report(
+                f"accessed discovery_info['{name}'] instead of discovery_info.{name}; this will fail in version 2022.6",
+                exclude_integrations={"zeroconf"},
+                error_if_core=False,
+                level=logging.DEBUG,
+            )
+            self._warning_logged = True
+        return getattr(self, name)
+
 
 @bind_hass
 async def async_get_instance(hass: HomeAssistant) -> HaZeroconf:
@@ -360,7 +383,7 @@ async def _process_service_update(
 
         # If we can handle it as a HomeKit discovery, we do that here.
         if service_type in HOMEKIT_TYPES:
-            props = info[ATTR_PROPERTIES]
+            props = info.properties
             if domain := async_get_homekit_discovery_domain(self.homekit_models, props):
                 discovery_flow.async_create_flow(
                     self.hass, domain, {"source": config_entries.SOURCE_HOMEKIT}, info
@@ -382,25 +405,23 @@ async def _process_service_update(
                     # likely bad homekit data
                     return
 
-        if ATTR_NAME in info:
-            lowercase_name: str | None = info[ATTR_NAME].lower()
+        if info.name:
+            lowercase_name: str | None = info.name.lower()
         else:
             lowercase_name = None
 
-        if "macaddress" in info[ATTR_PROPERTIES]:
-            uppercase_mac: str | None = info[ATTR_PROPERTIES]["macaddress"].upper()
+        if "macaddress" in info.properties:
+            uppercase_mac: str | None = info.properties["macaddress"].upper()
         else:
             uppercase_mac = None
 
-        if "manufacturer" in info[ATTR_PROPERTIES]:
-            lowercase_manufacturer: str | None = info[ATTR_PROPERTIES][
-                "manufacturer"
-            ].lower()
+        if "manufacturer" in info.properties:
+            lowercase_manufacturer: str | None = info.properties["manufacturer"].lower()
         else:
             lowercase_manufacturer = None
 
-        if "model" in info[ATTR_PROPERTIES]:
-            lowercase_model: str | None = info[ATTR_PROPERTIES]["model"].lower()
+        if "model" in info.properties:
+            lowercase_model: str | None = info.properties["model"].lower()
         else:
             lowercase_model = None
 

tests/components/homekit_controller/test_config_flow.py

@@ -162,8 +162,8 @@ def get_device_discovery_info(
         del result["properties"]["c#"]
 
     if upper_case_props:
-        result["properties"] = {
-            key.upper(): val for (key, val) in result["properties"].items()
+        result.properties = {
+            key.upper(): val for (key, val) in result.properties.items()
         }
 
     return result

tests/components/ipp/test_config_flow.py

@@ -1,4 +1,5 @@
 """Tests for the IPP config flow."""
+import dataclasses
 from unittest.mock import patch
 
 from homeassistant.components import zeroconf
@@ -40,7 +41,7 @@ async def test_show_zeroconf_form(
     """Test that the zeroconf confirmation form is served."""
     mock_connection(aioclient_mock)
 
-    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},
@@ -76,7 +77,7 @@ async def test_zeroconf_connection_error(
     """Test we abort zeroconf flow on IPP connection error."""
     mock_connection(aioclient_mock, conn_error=True)
 
-    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},
@@ -93,7 +94,7 @@ async def test_zeroconf_confirm_connection_error(
     """Test we abort zeroconf flow on IPP connection error."""
     mock_connection(aioclient_mock, conn_error=True)
 
-    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={"source": SOURCE_ZEROCONF}, data=discovery_info
     )
@@ -126,7 +127,7 @@ async def test_zeroconf_connection_upgrade_required(
     """Test we abort zeroconf flow on IPP connection error."""
     mock_connection(aioclient_mock, conn_upgrade_error=True)
 
-    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},
@@ -160,7 +161,7 @@ async def test_zeroconf_parse_error(
     """Test we abort zeroconf flow on IPP parse error."""
     mock_connection(aioclient_mock, parse_error=True)
 
-    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},
@@ -194,7 +195,7 @@ async def test_zeroconf_ipp_error(
     """Test we abort zeroconf flow on IPP error."""
     mock_connection(aioclient_mock, ipp_error=True)
 
-    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},
@@ -228,7 +229,7 @@ async def test_zeroconf_ipp_version_error(
     """Test we abort zeroconf flow on IPP version not supported error."""
     mock_connection(aioclient_mock, version_not_supported=True)
 
-    discovery_info = {**MOCK_ZEROCONF_IPP_SERVICE_INFO}
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},
@@ -262,7 +263,7 @@ async def test_zeroconf_device_exists_abort(
     """Test we abort zeroconf flow if printer already configured."""
     await init_integration(hass, aioclient_mock, skip_setup=True)
 
-    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},
@@ -279,13 +280,12 @@ async def test_zeroconf_with_uuid_device_exists_abort(
     """Test we abort zeroconf flow if printer already configured."""
     await init_integration(hass, aioclient_mock, skip_setup=True)
 
-    discovery_info = {
-        **MOCK_ZEROCONF_IPP_SERVICE_INFO,
-        "properties": {
-            **MOCK_ZEROCONF_IPP_SERVICE_INFO[zeroconf.ATTR_PROPERTIES],
-            "UUID": "cfe92100-67c4-11d4-a45f-f8d027761251",
-        },
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
+    discovery_info.properties = {
+        **MOCK_ZEROCONF_IPP_SERVICE_INFO[zeroconf.ATTR_PROPERTIES],
+        "UUID": "cfe92100-67c4-11d4-a45f-f8d027761251",
     }
+
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},
@@ -302,12 +302,10 @@ async def test_zeroconf_empty_unique_id(
     """Test zeroconf flow if printer lacks (empty) unique identification."""
     mock_connection(aioclient_mock, no_unique_id=True)
 
-    discovery_info = {
-        **MOCK_ZEROCONF_IPP_SERVICE_INFO,
-        "properties": {
-            **MOCK_ZEROCONF_IPP_SERVICE_INFO[zeroconf.ATTR_PROPERTIES],
-            "UUID": "",
-        },
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
+    discovery_info.properties = {
+        **MOCK_ZEROCONF_IPP_SERVICE_INFO[zeroconf.ATTR_PROPERTIES],
+        "UUID": "",
     }
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
@@ -324,7 +322,7 @@ async def test_zeroconf_no_unique_id(
     """Test zeroconf flow if printer lacks unique identification."""
     mock_connection(aioclient_mock, no_unique_id=True)
 
-    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},
@@ -371,7 +369,7 @@ async def test_full_zeroconf_flow_implementation(
     """Test the full manual user flow from start to finish."""
     mock_connection(aioclient_mock)
 
-    discovery_info = MOCK_ZEROCONF_IPP_SERVICE_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPP_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},
@@ -405,7 +403,7 @@ async def test_full_zeroconf_tls_flow_implementation(
     """Test the full manual user flow from start to finish."""
     mock_connection(aioclient_mock, ssl=True)
 
-    discovery_info = MOCK_ZEROCONF_IPPS_SERVICE_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_ZEROCONF_IPPS_SERVICE_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={"source": SOURCE_ZEROCONF},

tests/components/lookin/test_config_flow.py

@@ -1,12 +1,12 @@
 """Define tests for the lookin config flow."""
 from __future__ import annotations
 
+import dataclasses
 from unittest.mock import patch
 
 from aiolookin import NoUsableService
 
 from homeassistant import config_entries
-from homeassistant.components import zeroconf
 from homeassistant.components.lookin.const import DOMAIN
 from homeassistant.const import CONF_HOST
 from homeassistant.core import HomeAssistant
@@ -138,8 +138,8 @@ async def test_discovered_zeroconf(hass):
     assert mock_async_setup_entry.called
 
     entry = hass.config_entries.async_entries(DOMAIN)[0]
-    zc_data_new_ip = ZEROCONF_DATA.copy()
-    zc_data_new_ip[zeroconf.ATTR_HOST] = "127.0.0.2"
+    zc_data_new_ip = dataclasses.replace(ZEROCONF_DATA)
+    zc_data_new_ip.host = "127.0.0.2"
 
     with _patch_get_info(), patch(
         f"{MODULE}.async_setup_entry", return_value=True

tests/components/roku/test_config_flow.py

@@ -1,4 +1,5 @@
 """Test the Roku config flow."""
+import dataclasses
 from unittest.mock import patch
 
 from homeassistant.components.roku.const import DOMAIN
@@ -136,7 +137,7 @@ async def test_homekit_cannot_connect(
         error=True,
     )
 
-    discovery_info = MOCK_HOMEKIT_DISCOVERY_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_HOMEKIT_DISCOVERY_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN,
         context={CONF_SOURCE: SOURCE_HOMEKIT},
@@ -151,7 +152,7 @@ async def test_homekit_unknown_error(
     hass: HomeAssistant, aioclient_mock: AiohttpClientMocker
 ) -> None:
     """Test we abort homekit flow on unknown error."""
-    discovery_info = MOCK_HOMEKIT_DISCOVERY_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_HOMEKIT_DISCOVERY_INFO)
     with patch(
         "homeassistant.components.roku.config_flow.Roku.update",
         side_effect=Exception,
@@ -172,7 +173,7 @@ async def test_homekit_discovery(
     """Test the homekit discovery flow."""
     mock_connection(aioclient_mock, device="rokutv", host=HOMEKIT_HOST)
 
-    discovery_info = MOCK_HOMEKIT_DISCOVERY_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_HOMEKIT_DISCOVERY_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={CONF_SOURCE: SOURCE_HOMEKIT}, data=discovery_info
     )
@@ -200,7 +201,7 @@ async def test_homekit_discovery(
     assert len(mock_setup_entry.mock_calls) == 1
 
     # test abort on existing host
-    discovery_info = MOCK_HOMEKIT_DISCOVERY_INFO.copy()
+    discovery_info = dataclasses.replace(MOCK_HOMEKIT_DISCOVERY_INFO)
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={CONF_SOURCE: SOURCE_HOMEKIT}, data=discovery_info
     )