From 19d71b1ce3de4b4ab8d597eed4a24a92570196d6 Mon Sep 17 00:00:00 2001 From: mr_miles Date: Sat, 1 Feb 2025 19:21:55 +0000 Subject: [PATCH 1/3] 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 --- .../action_error_exception_handler.py | 87 ++++++++++++++++++ sagemcom_api/client.py | 89 ++++++++++--------- 2 files changed, 133 insertions(+), 43 deletions(-) create mode 100644 sagemcom_api/action_error_exception_handler.py diff --git a/sagemcom_api/action_error_exception_handler.py b/sagemcom_api/action_error_exception_handler.py new file mode 100644 index 0000000..a70fdb5 --- /dev/null +++ b/sagemcom_api/action_error_exception_handler.py @@ -0,0 +1,87 @@ +"""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 + """ + # 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..cc0719c 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import hashlib import json import math @@ -21,33 +22,25 @@ ServerDisconnectedError, 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 +184,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 +243,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,7 +308,9 @@ async def login(self): try: response = await self.__api_request_async([actions], True) - except TimeoutError as exception: + 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." ) from exception @@ -349,7 +328,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 = "" @@ -390,6 +370,12 @@ async def get_encryption_method(self): 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 +389,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 @@ -419,6 +408,12 @@ async def get_value_by_xpath(self, xpath: str, options: dict | None = None) -> d 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 +430,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 +465,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 +494,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 +563,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 From a966be12a5f4539393fc913c8dffc91c64bddef7 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Mon, 6 Apr 2026 22:43:03 +0000 Subject: [PATCH 2/3] Fix post-rebase lint and formatting issues --- .../action_error_exception_handler.py | 25 +++++++------------ sagemcom_api/client.py | 9 +++---- 2 files changed, 12 insertions(+), 22 deletions(-) diff --git a/sagemcom_api/action_error_exception_handler.py b/sagemcom_api/action_error_exception_handler.py index a70fdb5..3323e8c 100644 --- a/sagemcom_api/action_error_exception_handler.py +++ b/sagemcom_api/action_error_exception_handler.py @@ -1,14 +1,14 @@ -"""Logic to spot and create ActionErrorExceptions""" +"""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_NON_WRITABLE_PARAMETER_ERR, XMO_REQUEST_ACTION_ERR, - XMO_UNKNOWN_PATH_ERR + XMO_UNKNOWN_PATH_ERR, ) from sagemcom_api.exceptions import ( AccessRestrictionException, @@ -17,12 +17,12 @@ MaximumSessionCountException, NonWritableParameterException, UnknownException, - UnknownPathException + UnknownPathException, ) class ActionErrorHandler: - """Raised when a requested action has an error""" + """Raised when a requested action has an error.""" KNOWN_EXCEPTIONS = ( XMO_AUTHENTICATION_ERR, @@ -30,19 +30,17 @@ class ActionErrorHandler: XMO_NON_WRITABLE_PARAMETER_ERR, XMO_UNKNOWN_PATH_ERR, XMO_MAX_SESSION_COUNT_ERR, - XMO_LOGIN_RETRY_ERR + XMO_LOGIN_RETRY_ERR, ) @staticmethod def throw_if(response): - """For anywhere that needs the old single-exception behaviour""" - + """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"] @@ -53,17 +51,12 @@ def throw_if(response): @staticmethod def is_unknown_exception(desc): - """ - True/false if the ActionError is one of our known types - """ - + """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 - """ + """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: diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index cc0719c..1513aa8 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -2,7 +2,6 @@ from __future__ import annotations -import asyncio import hashlib import json import math @@ -22,8 +21,8 @@ ServerDisconnectedError, TCPConnector, ) -from .action_error_exception_handler import ActionErrorHandler +from .action_error_exception_handler import ActionErrorHandler from .const import ( API_ENDPOINT, DEFAULT_TIMEOUT, @@ -310,7 +309,7 @@ async def login(self): response = await self.__api_request_async([actions], True) ActionErrorHandler.throw_if(response) - except asyncio.TimeoutError as exception: + 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 @@ -369,7 +368,6 @@ 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, @@ -407,7 +405,6 @@ async def get_value_by_xpath( 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, @@ -496,7 +493,7 @@ async def get_device_info(self) -> DeviceInfo: "software_version": "Device/DeviceInfo/SoftwareVersion", }, # missing values converted to empty string - suppress_action_errors=True + suppress_action_errors=True, ) data["manufacturer"] = "Sagemcom" From 20e7f314d34afdfad461993ca6f3b1789e781ad3 Mon Sep 17 00:00:00 2001 From: mr_miles Date: Wed, 8 Apr 2026 17:28:40 +0100 Subject: [PATCH 3/3] Address Copilot PR review comments on action-error handling - Fix relative imports in action_error_exception_handler.py (was absolute) - Remove dead code: is_unknown_exception() was never called - Fix get_value_by_xpath return type annotation: dict -> Any - Fix inaccurate comment in get_device_info (values are None, not empty string) - Replace throw_if() raise/catch pattern with throw_if_error() and throw_if_error_at() which raise directly, eliminating expensive exception construction and catching at call sites - Both methods accept ignore_unknown_path=False so callers can suppress UnknownPathException without swallowing auth/access errors - get_values_by_xpaths suppress path now does per-action error checking so non-unknown-path errors (e.g. auth failures) still propagate - Add unit tests covering suppress_action_errors behaviour: unknown-path suppression returns None, auth errors still raise when suppressed Co-Authored-By: Claude Sonnet 4.6 --- .../action_error_exception_handler.py | 43 +++++-- sagemcom_api/client.py | 40 +++---- tests/conftest.py | 12 ++ tests/fixtures/xpath_unknown_path_error.json | 22 ++++ tests/fixtures/xpaths_mixed_errors.json | 43 +++++++ tests/unit/test_client_basic.py | 111 +++++++++++++++++- 6 files changed, 234 insertions(+), 37 deletions(-) create mode 100644 tests/fixtures/xpath_unknown_path_error.json create mode 100644 tests/fixtures/xpaths_mixed_errors.json diff --git a/sagemcom_api/action_error_exception_handler.py b/sagemcom_api/action_error_exception_handler.py index 3323e8c..cf84693 100644 --- a/sagemcom_api/action_error_exception_handler.py +++ b/sagemcom_api/action_error_exception_handler.py @@ -1,6 +1,6 @@ """Logic to spot and create ActionErrorExceptions.""" -from sagemcom_api.const import ( +from .const import ( XMO_ACCESS_RESTRICTION_ERR, XMO_AUTHENTICATION_ERR, XMO_LOGIN_RETRY_ERR, @@ -10,7 +10,7 @@ XMO_REQUEST_ACTION_ERR, XMO_UNKNOWN_PATH_ERR, ) -from sagemcom_api.exceptions import ( +from .exceptions import ( AccessRestrictionException, AuthenticationException, LoginRetryErrorException, @@ -34,25 +34,42 @@ class ActionErrorHandler: ) @staticmethod - def throw_if(response): - """For anywhere that needs the old single-exception behaviour.""" + 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 - actions = response["reply"]["actions"] - for action in actions: + 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 - if action_error_desc == XMO_NO_ERR: - continue + @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 - raise ActionErrorHandler.from_error_description(action_error, action_error_desc) + action_error_desc = action_error["description"] + if action_error_desc == XMO_NO_ERR: + return - @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 + 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): diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 1513aa8..d995b18 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -29,7 +29,6 @@ DEFAULT_USER_AGENT, UINT_MAX, XMO_INVALID_SESSION_ERR, - XMO_NO_ERR, XMO_REQUEST_ACTION_ERR, XMO_REQUEST_NO_ERR, ) @@ -183,19 +182,8 @@ def __get_response(self, response, index=0): return value - def __get_response_value(self, response, index=0, throw_on_action_error: bool = False): + def __get_response_value(self, response, index=0): """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: @@ -307,13 +295,14 @@ 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." ) from exception + ActionErrorHandler.throw_if_error(response) + data = self.__get_response(response) if data["id"] is not None and data["nonce"] is not None: @@ -328,7 +317,7 @@ async def logout(self): actions = {"id": 0, "method": "logOut"} response = await self.__api_request_async([actions], False) - ActionErrorHandler.throw_if(response) + ActionErrorHandler.throw_if_error(response) self._session_id = -1 self._server_nonce = "" @@ -373,7 +362,7 @@ async def get_value_by_xpath( xpath: str, options: dict | None = None, suppress_action_errors: bool = False, - ) -> dict: + ) -> Any: """Retrieve raw value from router using XPath. :param xpath: path expression @@ -387,8 +376,7 @@ async def get_value_by_xpath( } response = await self.__api_request_async([actions], False) - if not suppress_action_errors: - ActionErrorHandler.throw_if(response) + ActionErrorHandler.throw_if_error(response, ignore_unknown_path=suppress_action_errors) data = self.__get_response_value(response) @@ -427,10 +415,16 @@ async def get_values_by_xpaths( ] response = await self.__api_request_async(actions, False) + if not suppress_action_errors: - ActionErrorHandler.throw_if(response) + 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)) - values = [self.__get_response_value(response, i) for i in range(len(xpaths))] data = dict(zip(xpaths.keys(), values, strict=True)) return data @@ -462,7 +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(response) + ActionErrorHandler.throw_if_error(response) return response @@ -492,7 +486,7 @@ async def get_device_info(self) -> DeviceInfo: "serial_number": "Device/DeviceInfo/SerialNumber", "software_version": "Device/DeviceInfo/SoftwareVersion", }, - # missing values converted to empty string + # missing values returned as None when action errors are suppressed suppress_action_errors=True, ) data["manufacturer"] = "Sagemcom" @@ -560,7 +554,7 @@ async def reboot(self): } response = await self.__api_request_async([action], False) - ActionErrorHandler.throw_if(response) + ActionErrorHandler.throw_if_error(response) data = self.__get_response_value(response) 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, + )