Fritz: introduce wrapper.py - part 1 (sensor) (#63682)

* Revert "Improve availability for Shelly Valve"

This reverts commit d8cbd81.

* Fritz: introduce api.py - part 1 (sensor)

* Revert

* coverage

* Apply review comments

* Better naming

* Coverage

* fix naming

* Avoid "ignore[no-any-return]"

.coveragerc

@@ -378,6 +378,7 @@ omit =
     homeassistant/components/fritz/sensor.py
     homeassistant/components/fritz/services.py
     homeassistant/components/fritz/switch.py
+    homeassistant/components/fritz/wrapper.py
     homeassistant/components/fritzbox_callmonitor/__init__.py
     homeassistant/components/fritzbox_callmonitor/const.py
     homeassistant/components/fritzbox_callmonitor/base.py

homeassistant/components/fritz/sensor.py

@@ -7,13 +7,7 @@
 import logging
 from typing import Any, Literal
 
-from fritzconnection.core.exceptions import (
-    FritzActionError,
-    FritzActionFailedError,
-    FritzConnectionException,
-    FritzInternalError,
-    FritzServiceError,
-)
+from fritzconnection.core.exceptions import FritzConnectionException
 from fritzconnection.lib.fritzstatus import FritzStatus
 
 from homeassistant.components.sensor import (
@@ -36,6 +30,7 @@
 
 from .common import FritzBoxBaseEntity, FritzBoxTools
 from .const import DOMAIN, DSL_CONNECTION, UPTIME_DEVIATION, MeshRoles
+from .wrapper import AvmWrapper
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -282,22 +277,12 @@ async def async_setup_entry(
     """Set up entry."""
     _LOGGER.debug("Setting up FRITZ!Box sensors")
     avm_device: FritzBoxTools = hass.data[DOMAIN][entry.entry_id]
+    avm_wrapper = AvmWrapper(avm_device)
 
     dsl: bool = False
-    try:
-        dslinterface = await hass.async_add_executor_job(
-            avm_device.connection.call_action,
-            "WANDSLInterfaceConfig:1",
-            "GetInfo",
-        )
+    dslinterface = await avm_wrapper.get_wan_dsl_interface_config()
+    if dslinterface:
         dsl = dslinterface["NewEnable"]
-    except (
-        FritzInternalError,
-        FritzActionError,
-        FritzActionFailedError,
-        FritzServiceError,
-    ):
-        pass
 
     entities = [
         FritzBoxSensor(avm_device, entry.title, description)

homeassistant/components/fritz/wrapper.py

@@ -0,0 +1,98 @@
+"""AVM FRITZ!Box API wrapper."""
+from __future__ import annotations
+
+from functools import partial
+import logging
+from typing import Any
+
+from fritzconnection.core.exceptions import (
+    FritzActionError,
+    FritzActionFailedError,
+    FritzConnectionException,
+    FritzLookUpError,
+    FritzSecurityError,
+    FritzServiceError,
+)
+
+from .common import FritzBoxTools
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class AvmWrapper:
+    """Setup AVM wrapper for API calls."""
+
+    def __init__(self, avm_device: FritzBoxTools) -> None:
+        """Init wrapper API class."""
+
+        self._avm_device = avm_device
+
+    def _service_call_action(
+        self,
+        service_name: str,
+        service_suffix: str,
+        action_name: str,
+        **kwargs: Any,
+    ) -> dict | None:
+        """Return service details."""
+
+        if (
+            f"{service_name}{service_suffix}"
+            not in self._avm_device.connection.services
+        ):
+            return None
+
+        try:
+            result: dict = self._avm_device.connection.call_action(
+                f"{service_name}:{service_suffix}",
+                action_name,
+                **kwargs,
+            )
+            return result
+        except FritzSecurityError:
+            _LOGGER.error(
+                "Authorization Error: Please check the provided credentials and verify that you can log into the web interface",
+                exc_info=True,
+            )
+        except (
+            FritzActionError,
+            FritzActionFailedError,
+            FritzServiceError,
+            FritzLookUpError,
+        ):
+            _LOGGER.error(
+                "Service/Action Error: cannot execute service %s with action %s",
+                service_name,
+                action_name,
+                exc_info=True,
+            )
+        except FritzConnectionException:
+            _LOGGER.error(
+                "Connection Error: Please check the device is properly configured for remote login",
+                exc_info=True,
+            )
+        return None
+
+    async def _async_service_call_action(
+        self, service_name: str, service_suffix: str, action_name: str, **kwargs: Any
+    ) -> dict[str, Any] | None:
+        """Make call_action async."""
+
+        return await self._avm_device.hass.async_add_executor_job(
+            partial(
+                self._service_call_action,
+                service_name,
+                service_suffix,
+                action_name,
+                **kwargs,
+            )
+        )
+
+    async def get_wan_dsl_interface_config(self) -> dict[str, Any] | None:
+        """Call WANDSLInterfaceConfig service."""
+
+        return await self._async_service_call_action(
+            "WANDSLInterfaceConfig",
+            "1",
+            "GetInfo",
+        )