Improve integration startup in AVM Fritz!Tools (#96269)

homeassistant/components/fritz/common.py

@@ -129,13 +129,34 @@ class Interface(TypedDict):
     type: str
 
 
-class HostInfo(TypedDict):
+HostAttributes = TypedDict(
-    """FRITZ!Box host info class."""
+    "HostAttributes",
-
+    {
-    mac: str
+        "Index": int,
-    name: str
+        "IPAddress": str,
-    ip: str
+        "MACAddress": str,
-    status: bool
+        "Active": bool,
+        "HostName": str,
+        "InterfaceType": str,
+        "X_AVM-DE_Port": int,
+        "X_AVM-DE_Speed": int,
+        "X_AVM-DE_UpdateAvailable": bool,
+        "X_AVM-DE_UpdateSuccessful": str,
+        "X_AVM-DE_InfoURL": str | None,
+        "X_AVM-DE_MACAddressList": str | None,
+        "X_AVM-DE_Model": str | None,
+        "X_AVM-DE_URL": str | None,
+        "X_AVM-DE_Guest": bool,
+        "X_AVM-DE_RequestClient": str,
+        "X_AVM-DE_VPN": bool,
+        "X_AVM-DE_WANAccess": str,
+        "X_AVM-DE_Disallow": bool,
+        "X_AVM-DE_IsMeshable": str,
+        "X_AVM-DE_Priority": str,
+        "X_AVM-DE_FriendlyName": str,
+        "X_AVM-DE_FriendlyNameIsWriteable": str,
+    },
+)
 
 
 class UpdateCoordinatorDataType(TypedDict):
@@ -353,11 +374,11 @@ class FritzBoxTools(
         """Event specific per FRITZ!Box entry to signal updates in devices."""
         return f"{DOMAIN}-device-update-{self._unique_id}"
 
-    async def _async_update_hosts_info(self) -> list[HostInfo]:
+    async def _async_update_hosts_info(self) -> list[HostAttributes]:
         """Retrieve latest hosts information from the FRITZ!Box."""
         try:
             return await self.hass.async_add_executor_job(
-                self.fritz_hosts.get_hosts_info
+                self.fritz_hosts.get_hosts_attributes
             )
         except Exception as ex:  # pylint: disable=[broad-except]
             if not self.hass.is_stopping:
@@ -392,29 +413,6 @@ class FritzBoxTools(
             return {int(item["DeflectionId"]): item for item in items}
         return {}
 
-    async def _async_get_wan_access(self, ip_address: str) -> bool | None:
-        """Get WAN access rule for given IP address."""
-        try:
-            wan_access = await self.hass.async_add_executor_job(
-                partial(
-                    self.connection.call_action,
-                    "X_AVM-DE_HostFilter:1",
-                    "GetWANAccessByIP",
-                    NewIPv4Address=ip_address,
-                )
-            )
-            return not wan_access.get("NewDisallow")
-        except FRITZ_EXCEPTIONS as ex:
-            _LOGGER.debug(
-                (
-                    "could not get WAN access rule for client device with IP '%s',"
-                    " error: %s"
-                ),
-                ip_address,
-                ex,
-            )
-            return None
-
     def manage_device_info(
         self, dev_info: Device, dev_mac: str, consider_home: bool
     ) -> bool:
@@ -462,17 +460,17 @@ class FritzBoxTools(
         new_device = False
         hosts = {}
         for host in await self._async_update_hosts_info():
-            if not host.get("mac"):
+            if not host.get("MACAddress"):
                 continue
 
-            hosts[host["mac"]] = Device(
+            hosts[host["MACAddress"]] = Device(
-                name=host["name"],
+                name=host["HostName"],
-                connected=host["status"],
+                connected=host["Active"],
                 connected_to="",
                 connection_type="",
-                ip_address=host["ip"],
+                ip_address=host["IPAddress"],
                 ssid=None,
-                wan_access=None,
+                wan_access="granted" in host["X_AVM-DE_WANAccess"],
             )
 
         if not self.fritz_status.device_has_mesh_support or (
@@ -484,8 +482,6 @@ class FritzBoxTools(
             )
             self.mesh_role = MeshRoles.NONE
             for mac, info in hosts.items():
-                if info.ip_address:
-                    info.wan_access = await self._async_get_wan_access(info.ip_address)
                 if self.manage_device_info(info, mac, consider_home):
                     new_device = True
             await self.async_send_signal_device_update(new_device)
@@ -535,11 +531,6 @@ class FritzBoxTools(
 
                 dev_info: Device = hosts[dev_mac]
 
-                if dev_info.ip_address:
-                    dev_info.wan_access = await self._async_get_wan_access(
-                        dev_info.ip_address
-                    )
-
                 for link in interf["node_links"]:
                     intf = mesh_intf.get(link["node_interface_1_uid"])
                     if intf is not None:
@@ -583,7 +574,7 @@ class FritzBoxTools(
     ) -> None:
         """Trigger device trackers cleanup."""
         device_hosts_list = await self.hass.async_add_executor_job(
-            self.fritz_hosts.get_hosts_info
+            self.fritz_hosts.get_hosts_attributes
         )
         entity_reg: er.EntityRegistry = er.async_get(self.hass)
 
@@ -600,8 +591,8 @@ class FritzBoxTools(
         device_hosts_macs = set()
         device_hosts_names = set()
         for device in device_hosts_list:
-            device_hosts_macs.add(device["mac"])
+            device_hosts_macs.add(device["MACAddress"])
-            device_hosts_names.add(device["name"])
+            device_hosts_names.add(device["HostName"])
 
         for entry in ha_entity_reg_list:
             if entry.original_name is None:

tests/components/fritz/conftest.py

@@ -6,7 +6,12 @@ from fritzconnection.core.processor import Service
 from fritzconnection.lib.fritzhosts import FritzHosts
 import pytest
 
-from .const import MOCK_FB_SERVICES, MOCK_MESH_DATA, MOCK_MODELNAME
+from .const import (
+    MOCK_FB_SERVICES,
+    MOCK_HOST_ATTRIBUTES_DATA,
+    MOCK_MESH_DATA,
+    MOCK_MODELNAME,
+)
 
 LOGGER = logging.getLogger(__name__)
 
@@ -75,6 +80,10 @@ class FritzHostMock(FritzHosts):
         """Retrurn mocked mesh data."""
         return MOCK_MESH_DATA
 
+    def get_hosts_attributes(self):
+        """Retrurn mocked host attributes data."""
+        return MOCK_HOST_ATTRIBUTES_DATA
+
 
 @pytest.fixture(name="fc_data")
 def fc_data_mock():

tests/components/fritz/const.py

@@ -52,27 +52,8 @@ MOCK_FB_SERVICES: dict[str, dict] = {
         },
     },
     "Hosts1": {
-        "GetGenericHostEntry": [
-            {
-                "NewIPAddress": MOCK_IPS["fritz.box"],
-                "NewAddressSource": "Static",
-                "NewLeaseTimeRemaining": 0,
-                "NewMACAddress": MOCK_MESH_MASTER_MAC,
-                "NewInterfaceType": "",
-                "NewActive": True,
-                "NewHostName": "fritz.box",
-            },
-            {
-                "NewIPAddress": MOCK_IPS["printer"],
-                "NewAddressSource": "DHCP",
-                "NewLeaseTimeRemaining": 0,
-                "NewMACAddress": "AA:BB:CC:00:11:22",
-                "NewInterfaceType": "Ethernet",
-                "NewActive": True,
-                "NewHostName": "printer",
-            },
-        ],
         "X_AVM-DE_GetMeshListPath": {},
+        "X_AVM-DE_GetHostListPath": {},
     },
     "LANEthernetInterfaceConfig1": {
         "GetStatistics": {
@@ -783,6 +764,58 @@ MOCK_MESH_DATA = {
     ],
 }
 
+MOCK_HOST_ATTRIBUTES_DATA = [
+    {
+        "Index": 1,
+        "IPAddress": MOCK_IPS["printer"],
+        "MACAddress": "AA:BB:CC:00:11:22",
+        "Active": True,
+        "HostName": "printer",
+        "InterfaceType": "Ethernet",
+        "X_AVM-DE_Port": 1,
+        "X_AVM-DE_Speed": 1000,
+        "X_AVM-DE_UpdateAvailable": False,
+        "X_AVM-DE_UpdateSuccessful": "unknown",
+        "X_AVM-DE_InfoURL": None,
+        "X_AVM-DE_MACAddressList": None,
+        "X_AVM-DE_Model": None,
+        "X_AVM-DE_URL": f"http://{MOCK_IPS['printer']}",
+        "X_AVM-DE_Guest": False,
+        "X_AVM-DE_RequestClient": "0",
+        "X_AVM-DE_VPN": False,
+        "X_AVM-DE_WANAccess": "granted",
+        "X_AVM-DE_Disallow": False,
+        "X_AVM-DE_IsMeshable": "0",
+        "X_AVM-DE_Priority": "0",
+        "X_AVM-DE_FriendlyName": "printer",
+        "X_AVM-DE_FriendlyNameIsWriteable": "1",
+    },
+    {
+        "Index": 2,
+        "IPAddress": MOCK_IPS["fritz.box"],
+        "MACAddress": MOCK_MESH_MASTER_MAC,
+        "Active": True,
+        "HostName": "fritz.box",
+        "InterfaceType": None,
+        "X_AVM-DE_Port": 0,
+        "X_AVM-DE_Speed": 0,
+        "X_AVM-DE_UpdateAvailable": False,
+        "X_AVM-DE_UpdateSuccessful": "unknown",
+        "X_AVM-DE_InfoURL": None,
+        "X_AVM-DE_MACAddressList": f"{MOCK_MESH_MASTER_MAC},{MOCK_MESH_MASTER_WIFI1_MAC}",
+        "X_AVM-DE_Model": None,
+        "X_AVM-DE_URL": f"http://{MOCK_IPS['fritz.box']}",
+        "X_AVM-DE_Guest": False,
+        "X_AVM-DE_RequestClient": "0",
+        "X_AVM-DE_VPN": False,
+        "X_AVM-DE_WANAccess": "granted",
+        "X_AVM-DE_Disallow": False,
+        "X_AVM-DE_IsMeshable": "1",
+        "X_AVM-DE_Priority": "0",
+        "X_AVM-DE_FriendlyName": "fritz.box",
+        "X_AVM-DE_FriendlyNameIsWriteable": "0",
+    },
+]
 
 MOCK_USER_DATA = MOCK_CONFIG[DOMAIN][CONF_DEVICES][0]
 MOCK_DEVICE_INFO = {