Handle missing paths in GetDeviceInfo

sagemcom_api/action_error_exception_handler.py

@@ -0,0 +1,80 @@
+"""Logic to spot and create ActionErrorExceptions."""
+
+from sagemcom_api.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 sagemcom_api.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(response):
+        """For anywhere that needs the old single-exception behaviour."""
+        if response["reply"]["error"]["description"] != XMO_REQUEST_ACTION_ERR:
+            return
+
+        actions = response["reply"]["actions"]
+        for action in actions:
+            action_error = action["error"]
+            action_error_desc = action_error["description"]
+
+            if action_error_desc == XMO_NO_ERR:
+                continue
+
+            raise ActionErrorHandler.from_error_description(action_error, action_error_desc)
+
+    @staticmethod
+    def is_unknown_exception(desc):
+        """True/false if the ActionError is one of our known types."""
+        return False if desc == XMO_NO_ERR else desc not in ActionErrorHandler.KNOWN_EXCEPTIONS
+
+    @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,32 +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,
     LoginRetryErrorException,
     LoginTimeoutException,
-    MaximumSessionCountException,
-    NonWritableParameterException,
     UnauthorizedException,
     UnknownException,
     UnknownPathException,
@@ -191,8 +183,19 @@ def __get_response(self, response, index=0):
 
         return value
 
-    def __get_response_value(self, response, index=0):
+    def __get_response_value(self, response, index=0, throw_on_action_error: bool = False):
         """Retrieve response value from value."""
+        if throw_on_action_error:
+            try:
+                error = response["reply"]["actions"][index]["error"]
+            except (KeyError, IndexError):
+                error = None
+
+            if error is not None:
+                error_description = error["description"]
+                if error_description != XMO_NO_ERR:
+                    raise ActionErrorHandler.from_error_description(error, error_description)
+
         try:
             value = self.__get_response(response, index)["value"]
         except KeyError:
@@ -239,37 +242,10 @@ async def __post(self, url, data):
                 self._request_id = -1
                 raise InvalidSessionException(error)
 
-            # Error in one of the actions
+            # Unknown error in one of the actions
             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)
+                # leave this to the layer above as there may be multiple actions
+                pass
 
             return result
 
@@ -331,6 +307,8 @@ async def login(self):
 
         try:
             response = await self.__api_request_async([actions], True)
+            ActionErrorHandler.throw_if(response)
+
         except TimeoutError as exception:
             raise LoginTimeoutException(
                 "Login request timed-out. This could be caused by using the wrong encryption method, or using a (non) SSL connection."
@@ -349,7 +327,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(response)
 
         self._session_id = -1
         self._server_nonce = ""
@@ -389,7 +368,12 @@ async def get_encryption_method(self):
         max_tries=1,
         on_backoff=retry_login,
     )
-    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,
+    ) -> dict:
         """Retrieve raw value from router using XPath.
 
         :param xpath: path expression
@@ -403,6 +387,9 @@ async def get_value_by_xpath(self, xpath: str, options: dict | None = None) -> d
         }
 
         response = await self.__api_request_async([actions], False)
+        if not suppress_action_errors:
+            ActionErrorHandler.throw_if(response)
+
         data = self.__get_response_value(response)
 
         return data
@@ -418,7 +405,12 @@ 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, options: dict | None = None) -> dict:
+    async def get_values_by_xpaths(
+        self,
+        xpaths,
+        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
@@ -435,6 +427,9 @@ async def get_values_by_xpaths(self, xpaths, options: dict | None = None) -> dic
         ]
 
         response = await self.__api_request_async(actions, False)
+        if not suppress_action_errors:
+            ActionErrorHandler.throw_if(response)
+
         values = [self.__get_response_value(response, i) for i in range(len(xpaths))]
         data = dict(zip(xpaths.keys(), values, strict=True))
 
@@ -467,6 +462,7 @@ async def set_value_by_xpath(self, xpath: str, value: str, options: dict | None
         }
 
         response = await self.__api_request_async([actions], False)
+        ActionErrorHandler.throw_if(response)
 
         return response
 
@@ -495,7 +491,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 converted to empty string
+                suppress_action_errors=True,
             )
             data["manufacturer"] = "Sagemcom"
 
@@ -562,6 +560,8 @@ async def reboot(self):
         }
 
         response = await self.__api_request_async([action], False)
+        ActionErrorHandler.throw_if(response)
+
         data = self.__get_response_value(response)
 
         return data