Handle missing paths in GetDeviceInfo

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,32 +22,23 @@
     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,
@@ -239,37 +230,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,11 +295,14 @@ async def login(self):
 
         try:
             response = await self.__api_request_async([actions], True)
+
         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."
             ) from exception
 
+        ActionErrorHandler.throw_if_error(response)
+
         data = self.__get_response(response)
 
         if data["id"] is not None and data["nonce"] is not None:
@@ -349,7 +316,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 = ""
@@ -389,7 +357,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,
+    ) -> Any:
         """Retrieve raw value from router using XPath.
 
         :param xpath: path expression
@@ -403,6 +376,8 @@ async def get_value_by_xpath(self, xpath: str, options: dict | None = None) -> d
         }
 
         response = await self.__api_request_async([actions], False)
+        ActionErrorHandler.throw_if_error(response, ignore_unknown_path=suppress_action_errors)
+
         data = self.__get_response_value(response)
 
         return data
@@ -418,7 +393,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,7 +415,16 @@ async def get_values_by_xpaths(self, xpaths, options: dict | None = None) -> dic
         ]
 
         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
@@ -467,6 +456,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_error(response)
 
         return response
 
@@ -495,7 +485,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 returned as None when action errors are suppressed
+                suppress_action_errors=True,
             )
             data["manufacturer"] = "Sagemcom"
 
@@ -562,6 +554,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

tests/conftest.py

@@ -64,6 +64,18 @@ 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 mock_session_factory():
     """Create a factory for mock aiohttp ClientSession.

tests/fixtures/xpath_unknown_path_error.json

@@ -0,0 +1,22 @@
+{
+  "reply": {
+    "uid": 0,
+    "id": 1,
+    "error": {
+      "code": 16777236,
+      "description": "XMO_REQUEST_ACTION_ERR"
+    },
+    "actions": [
+      {
+        "uid": 1,
+        "id": 0,
+        "error": {
+          "code": 16777221,
+          "description": "XMO_UNKNOWN_PATH_ERR"
+        },
+        "callbacks": []
+      }
+    ],
+    "events": []
+  }
+}

tests/fixtures/xpaths_mixed_errors.json

@@ -0,0 +1,43 @@
+{
+  "reply": {
+    "uid": 0,
+    "id": 1,
+    "error": {
+      "code": 16777236,
+      "description": "XMO_REQUEST_ACTION_ERR"
+    },
+    "actions": [
+      {
+        "uid": 1,
+        "id": 0,
+        "error": {
+          "code": 16777238,
+          "description": "XMO_NO_ERR"
+        },
+        "callbacks": [
+          {
+            "uid": 1,
+            "result": {
+              "code": 16777238,
+              "description": "XMO_NO_ERR"
+            },
+            "xpath": "Device/DeviceInfo/MACAddress",
+            "parameters": {
+              "value": "AA:BB:CC:DD:EE:FF"
+            }
+          }
+        ]
+      },
+      {
+        "uid": 2,
+        "id": 1,
+        "error": {
+          "code": 16777221,
+          "description": "XMO_UNKNOWN_PATH_ERR"
+        },
+        "callbacks": []
+      }
+    ],
+    "events": []
+  }
+}