Handle missing paths in GetDeviceInfo

TalkTalk (UK) branded F5364 does not recognise the ModelNumber attribute
requested in GetDeviceInfo, which caused an exception because the response
handler threw for any ActionError. As a result HA could not connect to the
router to retrieve useful info, even though ModelNumber is not essential.

Fixed by:
 - Lifting ActionError handling out of the low-level POST into a dedicated
   ActionErrorHandler, so callers can react appropriately (and multiple
   actions can be handled individually, addressing the previous TODO).
 - Adding a suppress_action_errors option so get_value(s)_by_xpath can
   ignore unknown-path errors and return None for missing values, while
   still raising genuine errors (auth, etc.).
 - Using that option in get_device_info's fallback so missing DeviceInfo
   attributes are tolerated.

Existing error-raising behaviour is retained for all other calls, which now
invoke ActionErrorHandler.throw_if_error explicitly.

Tests cover the suppression behaviour of get_value(s)_by_xpath, the
end-to-end get_device_info fallback for the reported F5364 case, and the
ActionErrorHandler in isolation (throw_if_error / throw_if_error_at /
from_error_description).

Rebased onto upstream main, preserving the XPath-escaping fix (#476),
get_logs, and the speed-test additions.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mr-miles

sagemcom_api/action_error_exception_handler.py

@@ -0,0 +1,97 @@
+"""Logic to spot and create ActionErrorExceptions."""
+
+from .const import (
+    XMO_ACCESS_RESTRICTION_ERR,
+    XMO_AUTHENTICATION_ERR,
+    XMO_LOGIN_RETRY_ERR,
+    XMO_MAX_SESSION_COUNT_ERR,
+    XMO_NO_ERR,
+    XMO_NON_WRITABLE_PARAMETER_ERR,
+    XMO_REQUEST_ACTION_ERR,
+    XMO_UNKNOWN_PATH_ERR,
+)
+from .exceptions import (
+    AccessRestrictionException,
+    AuthenticationException,
+    LoginRetryErrorException,
+    MaximumSessionCountException,
+    NonWritableParameterException,
+    UnknownException,
+    UnknownPathException,
+)
+
+
+class ActionErrorHandler:
+    """Raised when a requested action has an error."""
+
+    KNOWN_EXCEPTIONS = (
+        XMO_AUTHENTICATION_ERR,
+        XMO_ACCESS_RESTRICTION_ERR,
+        XMO_NON_WRITABLE_PARAMETER_ERR,
+        XMO_UNKNOWN_PATH_ERR,
+        XMO_MAX_SESSION_COUNT_ERR,
+        XMO_LOGIN_RETRY_ERR,
+    )
+
+    @staticmethod
+    def throw_if_error(response, ignore_unknown_path: bool = False) -> None:
+        """Raise the first action-level error, or do nothing if all actions succeeded.
+
+        :param ignore_unknown_path: if True, silently ignore UnknownPathException
+        """
+        if response["reply"]["error"]["description"] != XMO_REQUEST_ACTION_ERR:
+            return
+
+        for action in response["reply"]["actions"]:
+            action_error = action["error"]
+            action_error_desc = action_error["description"]
+            if action_error_desc != XMO_NO_ERR:
+                exc = ActionErrorHandler.from_error_description(action_error, action_error_desc)
+                if ignore_unknown_path and isinstance(exc, UnknownPathException):
+                    continue
+                raise exc
+
+    @staticmethod
+    def throw_if_error_at(response, index: int, ignore_unknown_path: bool = False) -> None:
+        """Raise the error for a specific action, or do nothing if it succeeded.
+
+        :param ignore_unknown_path: if True, silently ignore UnknownPathException
+        """
+        try:
+            action_error = response["reply"]["actions"][index]["error"]
+        except (KeyError, IndexError):
+            return
+
+        action_error_desc = action_error["description"]
+        if action_error_desc == XMO_NO_ERR:
+            return
+
+        exc = ActionErrorHandler.from_error_description(action_error, action_error_desc)
+        if ignore_unknown_path and isinstance(exc, UnknownPathException):
+            return
+        raise exc
+
+    @staticmethod
+    def from_error_description(action_error, action_error_desc):
+        """Create the correct exception from an error, for the caller to throw."""
+        # pylint: disable=too-many-return-statements
+
+        if action_error_desc == XMO_AUTHENTICATION_ERR:
+            return AuthenticationException(action_error)
+
+        if action_error_desc == XMO_ACCESS_RESTRICTION_ERR:
+            return AccessRestrictionException(action_error)
+
+        if action_error_desc == XMO_NON_WRITABLE_PARAMETER_ERR:
+            return NonWritableParameterException(action_error)
+
+        if action_error_desc == XMO_UNKNOWN_PATH_ERR:
+            return UnknownPathException(action_error)
+
+        if action_error_desc == XMO_MAX_SESSION_COUNT_ERR:
+            return MaximumSessionCountException(action_error)
+
+        if action_error_desc == XMO_LOGIN_RETRY_ERR:
+            return LoginRetryErrorException(action_error)
+
+        return UnknownException(action_error)

sagemcom_api/client.py

@@ -22,33 +22,24 @@
     TCPConnector,
 )
 
+from .action_error_exception_handler import ActionErrorHandler
 from .const import (
     API_ENDPOINT,
     DEFAULT_TIMEOUT,
     DEFAULT_USER_AGENT,
     UINT_MAX,
-    XMO_ACCESS_RESTRICTION_ERR,
-    XMO_AUTHENTICATION_ERR,
     XMO_INVALID_SESSION_ERR,
-    XMO_LOGIN_RETRY_ERR,
-    XMO_MAX_SESSION_COUNT_ERR,
-    XMO_NO_ERR,
-    XMO_NON_WRITABLE_PARAMETER_ERR,
     XMO_REQUEST_ACTION_ERR,
     XMO_REQUEST_NO_ERR,
-    XMO_UNKNOWN_PATH_ERR,
 )
 from .enums import EncryptionMethod
 from .exceptions import (
-    AccessRestrictionException,
     AuthenticationException,
     BadRequestException,
     InvalidSessionException,
     LoginConnectionException,
     LoginRetryErrorException,
     LoginTimeoutException,
-    MaximumSessionCountException,
-    NonWritableParameterException,
     UnauthorizedException,
     UnknownException,
     UnknownPathException,
@@ -240,37 +231,12 @@ async def __post(self, url, data):
                 self._request_id = -1
                 raise InvalidSessionException(error)
 
-            # Error in one of the actions
+            # Error in one or more of the actions. Leave this to the layer
+            # above (via ActionErrorHandler), since a request may contain
+            # multiple actions and the caller may want to react per-action
+            # (e.g. suppress unknown-path errors for optional values).
             if error["description"] == XMO_REQUEST_ACTION_ERR:
-                # pylint:disable=fixme
-                # TODO How to support multiple actions + error handling?
-                actions = result["reply"]["actions"]
-                for action in actions:
-                    action_error = action["error"]
-                    action_error_desc = action_error["description"]
-
-                    if action_error_desc == XMO_NO_ERR:
-                        continue
-
-                    if action_error_desc == XMO_AUTHENTICATION_ERR:
-                        raise AuthenticationException(action_error)
-
-                    if action_error_desc == XMO_ACCESS_RESTRICTION_ERR:
-                        raise AccessRestrictionException(action_error)
-
-                    if action_error_desc == XMO_NON_WRITABLE_PARAMETER_ERR:
-                        raise NonWritableParameterException(action_error)
-
-                    if action_error_desc == XMO_UNKNOWN_PATH_ERR:
-                        raise UnknownPathException(action_error)
-
-                    if action_error_desc == XMO_MAX_SESSION_COUNT_ERR:
-                        raise MaximumSessionCountException(action_error)
-
-                    if action_error_desc == XMO_LOGIN_RETRY_ERR:
-                        raise LoginRetryErrorException(action_error)
-
-                    raise UnknownException(action_error)
+                pass
 
             return result
 
@@ -339,6 +305,8 @@ async def login(self):
         except (ClientConnectorError, ClientOSError) as exception:
             raise LoginConnectionException("Unable to connect to the device. Please check the host address.") from exception
 
+        ActionErrorHandler.throw_if_error(response)
+
         data = self.__get_response(response)
 
         if data["id"] is not None and data["nonce"] is not None:
@@ -352,7 +320,8 @@ async def logout(self):
         """Log out of the Sagemcom F@st device."""
         actions = {"id": 0, "method": "logOut"}
 
-        await self.__api_request_async([actions], False)
+        response = await self.__api_request_async([actions], False)
+        ActionErrorHandler.throw_if_error(response)
 
         self._session_id = -1
         self._server_nonce = ""
@@ -381,13 +350,20 @@ async def get_encryption_method(self):
 
         return None
 
-    async def get_value_by_xpath(self, xpath: str, options: dict | None = None) -> dict:
+    async def get_value_by_xpath(
+        self,
+        xpath: str,
+        options: dict | None = None,
+        suppress_action_errors: bool = False,
+    ) -> Any:
         """Retrieve raw value from router using XPath.
 
         :param xpath: path expression
         :param options: optional options
+        :param suppress_action_errors: if True, return None instead of raising
+            when the path is unknown (other action errors are still raised)
         """
-        result = await self.get_values_by_xpaths({"value": xpath}, options)
+        result = await self.get_values_by_xpaths({"value": xpath}, options, suppress_action_errors)
         return result["value"]
 
     @backoff.on_exception(
@@ -401,11 +377,18 @@ async def get_value_by_xpath(self, xpath: str, options: dict | None = None) -> d
         max_tries=1,
         on_backoff=retry_login,
     )
-    async def get_values_by_xpaths(self, xpaths: dict[str, str], options: dict | None = None) -> dict:
+    async def get_values_by_xpaths(
+        self,
+        xpaths: dict[str, str],
+        options: dict | None = None,
+        suppress_action_errors: bool = False,
+    ) -> dict:
         """Retrieve raw values from router using XPath.
 
         :param xpaths: Dict of key to xpath expression
         :param options: optional options
+        :param suppress_action_errors: if True, unknown-path actions return None
+            instead of raising, while other action errors are still raised
         """
         actions = [
             {
@@ -418,7 +401,16 @@ async def get_values_by_xpaths(self, xpaths: dict[str, str], options: dict | Non
         ]
 
         response = await self.__api_request_async(actions, False)
-        values = [self.__get_response_value(response, i) for i in range(len(xpaths))]
+
+        if not suppress_action_errors:
+            ActionErrorHandler.throw_if_error(response)
+            values = [self.__get_response_value(response, i) for i in range(len(xpaths))]
+        else:
+            values = []
+            for i in range(len(xpaths)):
+                ActionErrorHandler.throw_if_error_at(response, i, ignore_unknown_path=True)
+                values.append(self.__get_response_value(response, i))
+
         data = dict(zip(xpaths.keys(), values, strict=True))
 
         return data
@@ -461,6 +453,8 @@ async def set_values_by_xpaths(self, xpaths: dict[str, str], options: dict | Non
         ]
 
         response = await self.__api_request_async(actions, False)
+        ActionErrorHandler.throw_if_error(response)
+
         return response
 
     @backoff.on_exception(
@@ -488,7 +482,9 @@ async def get_device_info(self) -> DeviceInfo:
                     "product_class": "Device/DeviceInfo/ProductClass",
                     "serial_number": "Device/DeviceInfo/SerialNumber",
                     "software_version": "Device/DeviceInfo/SoftwareVersion",
-                }
+                },
+                # missing values are returned as None when action errors are suppressed
+                suppress_action_errors=True,
             )
             data["manufacturer"] = "Sagemcom"
 
@@ -554,6 +550,8 @@ async def get_logs(self) -> str:
         }
 
         response = await self.__api_request_async([actions], False)
+        ActionErrorHandler.throw_if_error(response)
+
         log_path = response["reply"]["actions"][0]["callbacks"][0]["parameters"]["uri"]
 
         log_uri = f"{self.protocol}://{self.host}{log_path}"
@@ -571,6 +569,8 @@ async def reboot(self):
         }
 
         response = await self.__api_request_async([action], False)
+        ActionErrorHandler.throw_if_error(response)
+
         data = self.__get_response_value(response)
 
         return data
@@ -585,7 +585,10 @@ async def run_speed_test(self, block_traffic: bool = False):
                 "parameters": {"BlockTraffic": block_traffic},
             }
         ]
-        return await self.__api_request_async(actions, False)
+        response = await self.__api_request_async(actions, False)
+        ActionErrorHandler.throw_if_error(response)
+
+        return response
 
     async def get_speed_test_results(self) -> list[SpeedTestResult]:
         """Retrieve Speed Test results from Sagemcom F@st device."""

tests/conftest.py

@@ -64,6 +64,29 @@ def xpath_value_response() -> dict[str, Any]:
     return load_fixture("xpath_value.json")
 
 
+@pytest.fixture
+def xpath_unknown_path_error_response() -> dict[str, Any]:
+    """Mock response for XPath query that returns XMO_UNKNOWN_PATH_ERR."""
+    return load_fixture("xpath_unknown_path_error.json")
+
+
+@pytest.fixture
+def xpaths_mixed_errors_response() -> dict[str, Any]:
+    """Mock response for multi-XPath query with one success and one unknown-path error."""
+    return load_fixture("xpaths_mixed_errors.json")
+
+
+@pytest.fixture
+def device_info_fallback_partial_response() -> dict[str, Any]:
+    """Mock response for the get_device_info fallback where ModelNumber is unknown.
+
+    Mirrors the TalkTalk F5364 case: individual DeviceInfo attributes are queried
+    after the aggregate Device/DeviceInfo path fails, and ModelNumber returns
+    XMO_UNKNOWN_PATH_ERR while the other attributes succeed.
+    """
+    return load_fixture("device_info_fallback_partial.json")
+
+
 @pytest.fixture
 def mock_session_factory():
     """Create a factory for mock aiohttp ClientSession.