Move action-error handling to a separate function

Lift behaviour to higher level so caller can specify the behaviour

Specifically to avoid throwing exceptions on missing path errors when getting DeviceInfo

Author: mr-miles

sagemcom_api/action_error_exception_handler.py

@@ -0,0 +1,86 @@
+"""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_NON_WRITABLE_PARAMETER_ERR,
+    XMO_NO_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
+        """
+
+        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,25 @@
 )
 import backoff
 import humps
+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,
@@ -201,8 +193,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 = 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:
@@ -251,37 +254,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
 
@@ -344,6 +320,8 @@ async def login(self):
 
         try:
             response = await self.__api_request_async([actions], True)
+            ActionErrorHandler.throw_if(response)
+
         except asyncio.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."
@@ -362,7 +340,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 = ""
@@ -404,7 +383,7 @@ 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 = False) -> dict:
         """
         Retrieve raw value from router using XPath.
 
@@ -419,6 +398,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
@@ -434,7 +416,7 @@ 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 = False) -> dict:
         """
         Retrieve raw values from router using XPath.
 
@@ -452,6 +434,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))
 
@@ -487,6 +472,7 @@ async def set_value_by_xpath(
         }
 
         response = await self.__api_request_async([actions], False)
+        ActionErrorHandler.throw_if(response)
 
         return response
 
@@ -515,7 +501,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"
 
@@ -584,6 +572,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