diff --git a/sagemcom_api/action_error_exception_handler.py b/sagemcom_api/action_error_exception_handler.py new file mode 100644 index 0000000..cf84693 --- /dev/null +++ b/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) diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 3fb83ec..d995b18 100644 --- a/sagemcom_api/client.py +++ b/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 diff --git a/tests/conftest.py b/tests/conftest.py index 031abd4..e5855f6 100644 --- a/tests/conftest.py +++ b/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. diff --git a/tests/fixtures/xpath_unknown_path_error.json b/tests/fixtures/xpath_unknown_path_error.json new file mode 100644 index 0000000..744b233 --- /dev/null +++ b/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": [] + } +} diff --git a/tests/fixtures/xpaths_mixed_errors.json b/tests/fixtures/xpaths_mixed_errors.json new file mode 100644 index 0000000..614ec77 --- /dev/null +++ b/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": [] + } +} diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index d7ba7db..74cc8b9 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -6,7 +6,7 @@ from sagemcom_api.client import SagemcomClient from sagemcom_api.enums import EncryptionMethod -from sagemcom_api.exceptions import AuthenticationException +from sagemcom_api.exceptions import AuthenticationException, UnknownPathException @pytest.mark.asyncio @@ -123,3 +123,112 @@ async def test_login_with_preconfigured_fixture(mock_client_sha512): assert client.authentication_method == EncryptionMethod.SHA512 assert client._session_id == 12345 assert client._server_nonce == "abcdef1234567890" + + +@pytest.mark.asyncio +async def test_get_value_by_xpath_suppresses_unknown_path( + mock_session_factory, login_success_response, xpath_unknown_path_error_response +): + """Test that suppress_action_errors=True returns None for UnknownPathException.""" + mock_session = mock_session_factory([login_success_response, xpath_unknown_path_error_response]) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + session=mock_session, + ) + await client.login() + + result = await client.get_value_by_xpath("Device/NonExistent", suppress_action_errors=True) + + assert result is None + + +@pytest.mark.asyncio +async def test_get_value_by_xpath_raises_unknown_path_when_not_suppressed( + mock_session_factory, login_success_response, xpath_unknown_path_error_response +): + """Test that suppress_action_errors=False (default) raises UnknownPathException.""" + mock_session = mock_session_factory([login_success_response, xpath_unknown_path_error_response]) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + session=mock_session, + ) + await client.login() + + with pytest.raises(UnknownPathException): + await client.get_value_by_xpath("Device/NonExistent") + + +@pytest.mark.asyncio +async def test_get_value_by_xpath_still_raises_auth_error_when_suppressed( + mock_session_factory, login_success_response, login_auth_error_response +): + """Test that suppress_action_errors=True still raises AuthenticationException.""" + mock_session = mock_session_factory([login_success_response, login_auth_error_response]) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + session=mock_session, + ) + await client.login() + + with pytest.raises(AuthenticationException): + await client.get_value_by_xpath("Device/DeviceInfo", suppress_action_errors=True) + + +@pytest.mark.asyncio +async def test_get_values_by_xpaths_suppresses_unknown_path_per_action( + mock_session_factory, login_success_response, xpaths_mixed_errors_response +): + """Test that get_values_by_xpaths with suppress_action_errors=True returns None for + unknown-path actions while preserving successful values from other actions.""" + mock_session = mock_session_factory([login_success_response, xpaths_mixed_errors_response]) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + session=mock_session, + ) + await client.login() + + result = await client.get_values_by_xpaths( + { + "mac_address": "Device/DeviceInfo/MACAddress", + "model_number": "Device/DeviceInfo/ModelNumber", + }, + suppress_action_errors=True, + ) + + assert result["mac_address"] == "AA:BB:CC:DD:EE:FF" + assert result["model_number"] is None + + +@pytest.mark.asyncio +async def test_get_values_by_xpaths_still_raises_auth_error_when_suppressed( + mock_session_factory, login_success_response, login_auth_error_response +): + """Test that get_values_by_xpaths with suppress_action_errors=True still raises + AuthenticationException instead of silently swallowing it.""" + mock_session = mock_session_factory([login_success_response, login_auth_error_response]) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + session=mock_session, + ) + await client.login() + + with pytest.raises(AuthenticationException): + await client.get_values_by_xpaths( + {"mac_address": "Device/DeviceInfo/MACAddress"}, + suppress_action_errors=True, + )