From 4b85793c7c1b0444a5885d4996fecba3adff550f Mon Sep 17 00:00:00 2001 From: mr_miles Date: Mon, 6 Jul 2026 16:40:51 +0100 Subject: [PATCH 1/2] Handle missing paths in GetDeviceInfo TalkTalk (UK) branded F5364 does not recognise the ModelNumber attribute requested in GetDeviceInfo, which caused an exception because the response handler threw for any ActionError. As a result HA could not connect to the router to retrieve useful info, even though ModelNumber is not essential. Fixed by: - Lifting ActionError handling out of the low-level POST into a dedicated ActionErrorHandler, so callers can react appropriately (and multiple actions can be handled individually, addressing the previous TODO). - Adding a suppress_action_errors option so get_value(s)_by_xpath can ignore unknown-path errors and return None for missing values, while still raising genuine errors (auth, etc.). - Using that option in get_device_info's fallback so missing DeviceInfo attributes are tolerated. Existing error-raising behaviour is retained for all other calls, which now invoke ActionErrorHandler.throw_if_error explicitly. Tests cover the suppression behaviour of get_value(s)_by_xpath, the end-to-end get_device_info fallback for the reported F5364 case, and the ActionErrorHandler in isolation (throw_if_error / throw_if_error_at / from_error_description). Rebased onto upstream main, preserving the XPath-escaping fix (#476), get_logs, and the speed-test additions. [Rebased onto main 0b4d490: kept the XMO_INVALID_PATH_ERR mapping alongside XMO_UNKNOWN_PATH_ERR, and updated the auth-error tests for the one re-login retry introduced by #520 (max_tries=2).] Co-Authored-By: Claude Opus 4.8 (1M context) --- .../action_error_exception_handler.py | 99 ++++++++++++ sagemcom_api/client.py | 98 ++++++------ tests/conftest.py | 23 +++ .../device_info_fallback_partial.json | 127 +++++++++++++++ tests/fixtures/xpath_unknown_path_error.json | 22 +++ tests/fixtures/xpaths_mixed_errors.json | 43 +++++ tests/unit/test_action_error_handler.py | 142 +++++++++++++++++ tests/unit/test_client_basic.py | 150 ++++++++++++++++++ 8 files changed, 656 insertions(+), 48 deletions(-) create mode 100644 sagemcom_api/action_error_exception_handler.py create mode 100644 tests/fixtures/device_info_fallback_partial.json create mode 100644 tests/fixtures/xpath_unknown_path_error.json create mode 100644 tests/fixtures/xpaths_mixed_errors.json create mode 100644 tests/unit/test_action_error_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..5081bd5 --- /dev/null +++ b/sagemcom_api/action_error_exception_handler.py @@ -0,0 +1,99 @@ +"""Logic to spot and create ActionErrorExceptions.""" + +from .const import ( + XMO_ACCESS_RESTRICTION_ERR, + XMO_AUTHENTICATION_ERR, + XMO_INVALID_PATH_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_INVALID_PATH_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 in (XMO_INVALID_PATH_ERR, 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 edc4139..3de0d23 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -22,34 +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_PATH_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, LoginConnectionException, LoginRetryErrorException, LoginTimeoutException, - MaximumSessionCountException, - NonWritableParameterException, UnauthorizedException, UnknownException, UnknownPathException, @@ -261,37 +251,12 @@ async def __send(self, url, data): self._request_id = -1 raise InvalidSessionException(error) - # Error in one of the actions + # Error in one or more of the actions. Leave this to the layer + # above (via ActionErrorHandler), since a request may contain + # multiple actions and the caller may want to react per-action + # (e.g. suppress unknown-path errors for optional values). 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 in (XMO_INVALID_PATH_ERR, 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) + pass return result @@ -364,6 +329,8 @@ async def login(self): except (ClientConnectorError, ClientOSError) as exception: raise LoginConnectionException("Unable to connect to the device. Please check the host address.") from exception + ActionErrorHandler.throw_if_error(response) + data = self.__get_response(response) if data["id"] is not None and data["nonce"] is not None: @@ -377,7 +344,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 = "" @@ -409,13 +377,20 @@ async def get_encryption_method(self): return None - 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 :param options: optional options + :param suppress_action_errors: if True, return None instead of raising + when the path is unknown (other action errors are still raised) """ - result = await self.get_values_by_xpaths({"value": xpath}, options) + result = await self.get_values_by_xpaths({"value": xpath}, options, suppress_action_errors) return result["value"] @backoff.on_exception( @@ -429,11 +404,18 @@ async def get_value_by_xpath(self, xpath: str, options: dict | None = None) -> d max_tries=2, on_backoff=retry_login, ) - async def get_values_by_xpaths(self, xpaths: dict[str, str], options: dict | None = None) -> dict: + async def get_values_by_xpaths( + self, + xpaths: dict[str, str], + 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 :param options: optional options + :param suppress_action_errors: if True, unknown-path actions return None + instead of raising, while other action errors are still raised """ actions = [ { @@ -446,7 +428,16 @@ async def get_values_by_xpaths(self, xpaths: dict[str, str], options: dict | Non ] 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 @@ -489,6 +480,8 @@ async def set_values_by_xpaths(self, xpaths: dict[str, str], options: dict | Non ] response = await self.__api_request_async(actions, False) + ActionErrorHandler.throw_if_error(response) + return response @backoff.on_exception( @@ -516,7 +509,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 are returned as None when action errors are suppressed + suppress_action_errors=True, ) data["manufacturer"] = "Sagemcom" @@ -582,6 +577,8 @@ async def get_logs(self) -> str: } response = await self.__api_request_async([actions], False) + ActionErrorHandler.throw_if_error(response) + log_path = response["reply"]["actions"][0]["callbacks"][0]["parameters"]["uri"] log_uri = f"{self.protocol}://{self.host}{log_path}" @@ -599,6 +596,8 @@ async def reboot(self): } response = await self.__api_request_async([action], False, idempotent=False) + ActionErrorHandler.throw_if_error(response) + data = self.__get_response_value(response) return data @@ -613,7 +612,10 @@ async def run_speed_test(self, block_traffic: bool = False): "parameters": {"BlockTraffic": block_traffic}, } ] - return await self.__api_request_async(actions, False, idempotent=False) + response = await self.__api_request_async(actions, False, idempotent=False) + ActionErrorHandler.throw_if_error(response) + + return response async def get_speed_test_results(self) -> list[SpeedTestResult]: """Retrieve Speed Test results from Sagemcom F@st device.""" diff --git a/tests/conftest.py b/tests/conftest.py index edf39a9..2e7838a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -76,6 +76,29 @@ def xpath_invalid_path_response() -> dict[str, Any]: return load_fixture("xpath_invalid_path.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 device_info_fallback_partial_response() -> dict[str, Any]: + """Mock response for the get_device_info fallback where ModelNumber is unknown. + + Mirrors the TalkTalk F5364 case: individual DeviceInfo attributes are queried + after the aggregate Device/DeviceInfo path fails, and ModelNumber returns + XMO_UNKNOWN_PATH_ERR while the other attributes succeed. + """ + return load_fixture("device_info_fallback_partial.json") + + @pytest.fixture def mock_session_factory(): """Create a factory for mock aiohttp ClientSession. diff --git a/tests/fixtures/device_info_fallback_partial.json b/tests/fixtures/device_info_fallback_partial.json new file mode 100644 index 0000000..3f20eff --- /dev/null +++ b/tests/fixtures/device_info_fallback_partial.json @@ -0,0 +1,127 @@ +{ + "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": [] + }, + { + "uid": 3, + "id": 2, + "error": { + "code": 16777238, + "description": "XMO_NO_ERR" + }, + "callbacks": [ + { + "uid": 3, + "result": { + "code": 16777238, + "description": "XMO_NO_ERR" + }, + "xpath": "Device/DeviceInfo/ProductClass", + "parameters": { + "value": "F5364" + } + } + ] + }, + { + "uid": 4, + "id": 3, + "error": { + "code": 16777238, + "description": "XMO_NO_ERR" + }, + "callbacks": [ + { + "uid": 4, + "result": { + "code": 16777238, + "description": "XMO_NO_ERR" + }, + "xpath": "Device/DeviceInfo/ProductClass", + "parameters": { + "value": "F5364" + } + } + ] + }, + { + "uid": 5, + "id": 4, + "error": { + "code": 16777238, + "description": "XMO_NO_ERR" + }, + "callbacks": [ + { + "uid": 5, + "result": { + "code": 16777238, + "description": "XMO_NO_ERR" + }, + "xpath": "Device/DeviceInfo/SerialNumber", + "parameters": { + "value": "SN123456789" + } + } + ] + }, + { + "uid": 6, + "id": 5, + "error": { + "code": 16777238, + "description": "XMO_NO_ERR" + }, + "callbacks": [ + { + "uid": 6, + "result": { + "code": 16777238, + "description": "XMO_NO_ERR" + }, + "xpath": "Device/DeviceInfo/SoftwareVersion", + "parameters": { + "value": "1.2.3" + } + } + ] + } + ], + "events": [] + } +} 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_action_error_handler.py b/tests/unit/test_action_error_handler.py new file mode 100644 index 0000000..f643e82 --- /dev/null +++ b/tests/unit/test_action_error_handler.py @@ -0,0 +1,142 @@ +"""Unit tests for ActionErrorHandler.""" + +import pytest + +from sagemcom_api.action_error_exception_handler import ActionErrorHandler +from sagemcom_api.exceptions import ( + AccessRestrictionException, + AuthenticationException, + LoginRetryErrorException, + MaximumSessionCountException, + NonWritableParameterException, + UnknownException, + UnknownPathException, +) + + +def _response(request_desc, action_descs): + """Build a minimal JSON-req reply with the given request and action errors.""" + return { + "reply": { + "error": {"description": request_desc}, + "actions": [{"error": {"description": desc}} for desc in action_descs], + } + } + + +# --- throw_if_error -------------------------------------------------------- + + +def test_throw_if_error_noop_when_request_succeeded(): + """No exception is raised when the request-level error is not an action error.""" + response = _response("XMO_REQUEST_NO_ERR", ["XMO_UNKNOWN_PATH_ERR"]) + + # Must not raise even though an action carries an error description, because + # the request-level error indicates overall success. + ActionErrorHandler.throw_if_error(response) + + +def test_throw_if_error_noop_when_all_actions_ok(): + """No exception is raised when every action reports XMO_NO_ERR.""" + response = _response("XMO_REQUEST_ACTION_ERR", ["XMO_NO_ERR", "XMO_NO_ERR"]) + + ActionErrorHandler.throw_if_error(response) + + +def test_throw_if_error_raises_first_action_error(): + """The first erroring action determines the raised exception.""" + response = _response( + "XMO_REQUEST_ACTION_ERR", + ["XMO_NO_ERR", "XMO_UNKNOWN_PATH_ERR", "XMO_AUTHENTICATION_ERR"], + ) + + with pytest.raises(UnknownPathException): + ActionErrorHandler.throw_if_error(response) + + +def test_throw_if_error_ignore_unknown_path_skips_to_next_error(): + """Unknown-path errors are skipped and a later real error is still raised.""" + response = _response( + "XMO_REQUEST_ACTION_ERR", + ["XMO_UNKNOWN_PATH_ERR", "XMO_AUTHENTICATION_ERR"], + ) + + with pytest.raises(AuthenticationException): + ActionErrorHandler.throw_if_error(response, ignore_unknown_path=True) + + +def test_throw_if_error_ignore_unknown_path_noop_when_only_unknown_paths(): + """A response with only unknown-path errors does not raise when ignoring them.""" + response = _response( + "XMO_REQUEST_ACTION_ERR", + ["XMO_UNKNOWN_PATH_ERR", "XMO_UNKNOWN_PATH_ERR"], + ) + + ActionErrorHandler.throw_if_error(response, ignore_unknown_path=True) + + +# --- throw_if_error_at ----------------------------------------------------- + + +def test_throw_if_error_at_raises_for_indexed_action(): + """The exception for the action at the given index is raised.""" + response = _response( + "XMO_REQUEST_ACTION_ERR", + ["XMO_NO_ERR", "XMO_AUTHENTICATION_ERR"], + ) + + with pytest.raises(AuthenticationException): + ActionErrorHandler.throw_if_error_at(response, 1) + + +def test_throw_if_error_at_noop_when_action_ok(): + """No exception is raised when the indexed action reports XMO_NO_ERR.""" + response = _response("XMO_REQUEST_ACTION_ERR", ["XMO_NO_ERR"]) + + ActionErrorHandler.throw_if_error_at(response, 0) + + +def test_throw_if_error_at_noop_for_out_of_range_index(): + """An out-of-range index is tolerated rather than raising IndexError.""" + response = _response("XMO_REQUEST_ACTION_ERR", ["XMO_UNKNOWN_PATH_ERR"]) + + ActionErrorHandler.throw_if_error_at(response, 5) + + +def test_throw_if_error_at_ignore_unknown_path(): + """With ignore_unknown_path, an unknown-path error at the index is tolerated.""" + response = _response("XMO_REQUEST_ACTION_ERR", ["XMO_UNKNOWN_PATH_ERR"]) + + ActionErrorHandler.throw_if_error_at(response, 0, ignore_unknown_path=True) + + +def test_throw_if_error_at_still_raises_non_unknown_path_when_ignoring(): + """ignore_unknown_path must not suppress other error types.""" + response = _response("XMO_REQUEST_ACTION_ERR", ["XMO_AUTHENTICATION_ERR"]) + + with pytest.raises(AuthenticationException): + ActionErrorHandler.throw_if_error_at(response, 0, ignore_unknown_path=True) + + +# --- from_error_description ------------------------------------------------ + + +@pytest.mark.parametrize( + ("description", "expected"), + [ + ("XMO_AUTHENTICATION_ERR", AuthenticationException), + ("XMO_ACCESS_RESTRICTION_ERR", AccessRestrictionException), + ("XMO_NON_WRITABLE_PARAMETER_ERR", NonWritableParameterException), + ("XMO_UNKNOWN_PATH_ERR", UnknownPathException), + ("XMO_MAX_SESSION_COUNT_ERR", MaximumSessionCountException), + ("XMO_LOGIN_RETRY_ERR", LoginRetryErrorException), + ("SOMETHING_UNEXPECTED", UnknownException), + ], +) +def test_from_error_description_maps_to_exception(description, expected): + """Each known description maps to its exception; anything else is UnknownException.""" + action_error = {"description": description} + + exc = ActionErrorHandler.from_error_description(action_error, description) + + assert isinstance(exc, expected) diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index 18ce790..e71561a 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -173,3 +173,153 @@ 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.""" + # Auth errors trigger one re-login and retry (max_tries=2) before propagating. + mock_session = mock_session_factory([login_success_response, login_auth_error_response] * 2) + 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) + + assert mock_session.post.call_count == 4 + + +@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 +): + """Suppressed unknown-path actions return None while other values are preserved.""" + 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 +): + """A suppressed call still raises AuthenticationException instead of swallowing it.""" + # Auth errors trigger one re-login and retry (max_tries=2) before propagating. + mock_session = mock_session_factory([login_success_response, login_auth_error_response] * 2) + 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, + ) + + assert mock_session.post.call_count == 4 + + +@pytest.mark.asyncio +async def test_get_device_info_falls_back_when_aggregate_path_unknown( + mock_session_factory, + login_success_response, + xpath_unknown_path_error_response, + device_info_fallback_partial_response, +): + """A device missing the aggregate DeviceInfo path still yields a DeviceInfo. + + The aggregate probe raises UnknownPathException, triggering the per-attribute + fallback. There, the unknown ModelNumber is tolerated (returned as None) while + the remaining attributes are populated. + """ + mock_session = mock_session_factory( + [ + login_success_response, + xpath_unknown_path_error_response, + device_info_fallback_partial_response, + ] + ) + client = SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + session=mock_session, + ) + await client.login() + + info = await client.get_device_info() + + assert info.mac_address == "AA:BB:CC:DD:EE:FF" + assert info.model_name is None # ModelNumber was unknown, tolerated + assert info.model_number == "F5364" + assert info.serial_number == "SN123456789" + assert info.software_version == "1.2.3" + assert info.manufacturer == "Sagemcom" From 437f79d1f274fa44ce71331ba6bad5a67c847bf6 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Wed, 28 Oct 2026 00:28:44 +0000 Subject: [PATCH 2/2] Raise request errors by default and only ignore missing paths on request Follow-up to review feedback on #386: - Raise action errors centrally in __api_request_async instead of at each call site, so every request (login, logout, set, reboot, speed test, logs) keeps raising by default and new call sites cannot forget to. - Raise request-level XMO_AUTHENTICATION_ERR, XMO_ACCESS_RESTRICTION_ERR, XMO_MAX_SESSION_COUNT_ERR and XMO_LOGIN_RETRY_ERR before looking at the actions, so a placeholder unknown-path action cannot mask them. - Treat both XMO_INVALID_PATH_ERR and XMO_UNKNOWN_PATH_ERR as missing paths. - Rename suppress_action_errors to ignore_unknown_paths, since only missing paths are ignored, and drop the now unused throw_if_error_at helper. - Require the MAC address in the get_device_info fallback, as it is the gateway identity. - Cover request-level errors, invalid-path suppression, the #520 re-login retry, and action errors from login, set, reboot and speed test. --- .../action_error_exception_handler.py | 71 ++++---- sagemcom_api/client.py | 76 +++------ tests/unit/test_action_error_handler.py | 81 +++------ tests/unit/test_client_basic.py | 159 +++++++++++++++++- 4 files changed, 233 insertions(+), 154 deletions(-) diff --git a/sagemcom_api/action_error_exception_handler.py b/sagemcom_api/action_error_exception_handler.py index 5081bd5..f9a4486 100644 --- a/sagemcom_api/action_error_exception_handler.py +++ b/sagemcom_api/action_error_exception_handler.py @@ -14,6 +14,7 @@ from .exceptions import ( AccessRestrictionException, AuthenticationException, + BaseSagemcomException, LoginRetryErrorException, MaximumSessionCountException, NonWritableParameterException, @@ -21,60 +22,52 @@ UnknownPathException, ) +# Firmware differs in which of these it reports for a path that does not exist. +PATH_ERRORS = (XMO_INVALID_PATH_ERR, XMO_UNKNOWN_PATH_ERR) + +# These fail the whole request. Actions may still carry a placeholder such as +# XMO_UNKNOWN_PATH_ERR, which must not hide the real cause. +REQUEST_ERRORS = ( + XMO_AUTHENTICATION_ERR, + XMO_ACCESS_RESTRICTION_ERR, + XMO_MAX_SESSION_COUNT_ERR, + XMO_LOGIN_RETRY_ERR, +) + 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_INVALID_PATH_ERR, - XMO_UNKNOWN_PATH_ERR, - XMO_MAX_SESSION_COUNT_ERR, - XMO_LOGIN_RETRY_ERR, - ) + """Raise exceptions for errors reported in a JSON-req reply.""" @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. + def throw_if_error(response, ignore_unknown_paths: bool = False) -> None: + """Raise for a request-level error, otherwise for the first failed action. - :param ignore_unknown_path: if True, silently ignore UnknownPathException + :param ignore_unknown_paths: skip actions whose path does not exist, + while still raising any other error """ - if response["reply"]["error"]["description"] != XMO_REQUEST_ACTION_ERR: + request_error = response["reply"]["error"] + request_error_desc = request_error["description"] + + if request_error_desc in REQUEST_ERRORS: + raise ActionErrorHandler.from_error_description(request_error, request_error_desc) + + if request_error_desc != 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. + if action_error_desc == XMO_NO_ERR: + continue - :param ignore_unknown_path: if True, silently ignore UnknownPathException - """ - try: - action_error = response["reply"]["actions"][index]["error"] - except (KeyError, IndexError): - return + if ignore_unknown_paths and action_error_desc in PATH_ERRORS: + continue - 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 + raise ActionErrorHandler.from_error_description(action_error, action_error_desc) @staticmethod - def from_error_description(action_error, action_error_desc): + def from_error_description(action_error, action_error_desc) -> BaseSagemcomException: """Create the correct exception from an error, for the caller to throw.""" # pylint: disable=too-many-return-statements @@ -87,7 +80,7 @@ def from_error_description(action_error, action_error_desc): if action_error_desc == XMO_NON_WRITABLE_PARAMETER_ERR: return NonWritableParameterException(action_error) - if action_error_desc in (XMO_INVALID_PATH_ERR, XMO_UNKNOWN_PATH_ERR): + if action_error_desc in PATH_ERRORS: return UnknownPathException(action_error) if action_error_desc == XMO_MAX_SESSION_COUNT_ERR: diff --git a/sagemcom_api/client.py b/sagemcom_api/client.py index 3de0d23..3adf14d 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_REQUEST_ACTION_ERR, XMO_REQUEST_NO_ERR, ) from .enums import EncryptionMethod @@ -251,19 +250,13 @@ async def __send(self, url, data): self._request_id = -1 raise InvalidSessionException(error) - # Error in one or more of the actions. Leave this to the layer - # above (via ActionErrorHandler), since a request may contain - # multiple actions and the caller may want to react per-action - # (e.g. suppress unknown-path errors for optional values). - if error["description"] == XMO_REQUEST_ACTION_ERR: - pass - return result - async def __api_request_async(self, actions, priority=False, idempotent=True): + async def __api_request_async(self, actions, priority=False, idempotent=True, ignore_unknown_paths=False): """Build request to the internal JSON-req API. :param idempotent: set to False for actions that must not be resent after a dropped connection + :param ignore_unknown_paths: do not raise for actions whose path does not exist """ self.__generate_request_id() self.__generate_nonce() @@ -286,7 +279,6 @@ async def __api_request_async(self, actions, priority=False, idempotent=True): try: post = self.__post if idempotent else self.__post_once result = await post(api_host, form_data) - return result except ( ClientConnectorError, ClientOSError, @@ -294,6 +286,9 @@ async def __api_request_async(self, actions, priority=False, idempotent=True): ) as exception: raise ConnectionError(str(exception)) from exception + ActionErrorHandler.throw_if_error(result, ignore_unknown_paths=ignore_unknown_paths) + return result + async def login(self): """Login to the SagemCom F@st router using a username and password.""" actions = { @@ -329,8 +324,6 @@ async def login(self): except (ClientConnectorError, ClientOSError) as exception: raise LoginConnectionException("Unable to connect to the device. Please check the host address.") from exception - ActionErrorHandler.throw_if_error(response) - data = self.__get_response(response) if data["id"] is not None and data["nonce"] is not None: @@ -344,8 +337,7 @@ async def logout(self): """Log out of the Sagemcom F@st device.""" actions = {"id": 0, "method": "logOut"} - response = await self.__api_request_async([actions], False) - ActionErrorHandler.throw_if_error(response) + await self.__api_request_async([actions], False) self._session_id = -1 self._server_nonce = "" @@ -377,20 +369,15 @@ async def get_encryption_method(self): return None - async def get_value_by_xpath( - self, - xpath: str, - options: dict | None = None, - suppress_action_errors: bool = False, - ) -> Any: + async def get_value_by_xpath(self, xpath: str, options: dict | None = None, ignore_unknown_paths: bool = False) -> Any: """Retrieve raw value from router using XPath. :param xpath: path expression :param options: optional options - :param suppress_action_errors: if True, return None instead of raising - when the path is unknown (other action errors are still raised) + :param ignore_unknown_paths: return None instead of raising UnknownPathException + when the path does not exist; other errors are still raised """ - result = await self.get_values_by_xpaths({"value": xpath}, options, suppress_action_errors) + result = await self.get_values_by_xpaths({"value": xpath}, options, ignore_unknown_paths) return result["value"] @backoff.on_exception( @@ -404,18 +391,13 @@ async def get_value_by_xpath( max_tries=2, on_backoff=retry_login, ) - async def get_values_by_xpaths( - self, - xpaths: dict[str, str], - options: dict | None = None, - suppress_action_errors: bool = False, - ) -> dict: + async def get_values_by_xpaths(self, xpaths: dict[str, str], options: dict | None = None, ignore_unknown_paths: bool = False) -> dict: """Retrieve raw values from router using XPath. :param xpaths: Dict of key to xpath expression :param options: optional options - :param suppress_action_errors: if True, unknown-path actions return None - instead of raising, while other action errors are still raised + :param ignore_unknown_paths: return None for each path that does not exist instead + of raising UnknownPathException; other errors are still raised """ actions = [ { @@ -427,17 +409,8 @@ async def get_values_by_xpaths( for i, xpath in enumerate(xpaths.values()) ] - response = await self.__api_request_async(actions, False) - - 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)) - + response = await self.__api_request_async(actions, False, ignore_unknown_paths=ignore_unknown_paths) + values = [self.__get_response_value(response, i) for i in range(len(xpaths))] data = dict(zip(xpaths.keys(), values, strict=True)) return data @@ -480,8 +453,6 @@ async def set_values_by_xpaths(self, xpaths: dict[str, str], options: dict | Non ] response = await self.__api_request_async(actions, False) - ActionErrorHandler.throw_if_error(response) - return response @backoff.on_exception( @@ -510,11 +481,15 @@ async def get_device_info(self) -> DeviceInfo: "serial_number": "Device/DeviceInfo/SerialNumber", "software_version": "Device/DeviceInfo/SoftwareVersion", }, - # missing values are returned as None when action errors are suppressed - suppress_action_errors=True, + # Some firmware (e.g. TalkTalk F5364) lacks optional leaves such as ModelNumber. + ignore_unknown_paths=True, ) data["manufacturer"] = "Sagemcom" + # The MAC is the gateway's identity (DeviceInfo.id), so it cannot be optional. + if data["mac_address"] is None: + raise UnknownPathException("Device/DeviceInfo/MACAddress is not available") + return DeviceInfo(**data) @backoff.on_exception( @@ -577,8 +552,6 @@ async def get_logs(self) -> str: } response = await self.__api_request_async([actions], False) - ActionErrorHandler.throw_if_error(response) - log_path = response["reply"]["actions"][0]["callbacks"][0]["parameters"]["uri"] log_uri = f"{self.protocol}://{self.host}{log_path}" @@ -596,8 +569,6 @@ async def reboot(self): } response = await self.__api_request_async([action], False, idempotent=False) - ActionErrorHandler.throw_if_error(response) - data = self.__get_response_value(response) return data @@ -612,10 +583,7 @@ async def run_speed_test(self, block_traffic: bool = False): "parameters": {"BlockTraffic": block_traffic}, } ] - response = await self.__api_request_async(actions, False, idempotent=False) - ActionErrorHandler.throw_if_error(response) - - return response + return await self.__api_request_async(actions, False, idempotent=False) async def get_speed_test_results(self) -> list[SpeedTestResult]: """Retrieve Speed Test results from Sagemcom F@st device.""" diff --git a/tests/unit/test_action_error_handler.py b/tests/unit/test_action_error_handler.py index f643e82..206fdfe 100644 --- a/tests/unit/test_action_error_handler.py +++ b/tests/unit/test_action_error_handler.py @@ -27,9 +27,10 @@ def _response(request_desc, action_descs): # --- throw_if_error -------------------------------------------------------- -def test_throw_if_error_noop_when_request_succeeded(): - """No exception is raised when the request-level error is not an action error.""" - response = _response("XMO_REQUEST_NO_ERR", ["XMO_UNKNOWN_PATH_ERR"]) +@pytest.mark.parametrize("request_desc", ["XMO_REQUEST_NO_ERR", "Ok"]) +def test_throw_if_error_noop_when_request_succeeded(request_desc): + """No exception is raised when the request-level error indicates success.""" + response = _response(request_desc, ["XMO_UNKNOWN_PATH_ERR"]) # Must not raise even though an action carries an error description, because # the request-level error indicates overall success. @@ -54,68 +55,41 @@ def test_throw_if_error_raises_first_action_error(): ActionErrorHandler.throw_if_error(response) -def test_throw_if_error_ignore_unknown_path_skips_to_next_error(): +def test_throw_if_error_ignore_unknown_paths_skips_to_next_error(): """Unknown-path errors are skipped and a later real error is still raised.""" response = _response( "XMO_REQUEST_ACTION_ERR", - ["XMO_UNKNOWN_PATH_ERR", "XMO_AUTHENTICATION_ERR"], + ["XMO_UNKNOWN_PATH_ERR", "XMO_INVALID_PATH_ERR", "XMO_AUTHENTICATION_ERR"], ) with pytest.raises(AuthenticationException): - ActionErrorHandler.throw_if_error(response, ignore_unknown_path=True) + ActionErrorHandler.throw_if_error(response, ignore_unknown_paths=True) -def test_throw_if_error_ignore_unknown_path_noop_when_only_unknown_paths(): - """A response with only unknown-path errors does not raise when ignoring them.""" - response = _response( - "XMO_REQUEST_ACTION_ERR", - ["XMO_UNKNOWN_PATH_ERR", "XMO_UNKNOWN_PATH_ERR"], - ) - - ActionErrorHandler.throw_if_error(response, ignore_unknown_path=True) - - -# --- throw_if_error_at ----------------------------------------------------- - - -def test_throw_if_error_at_raises_for_indexed_action(): - """The exception for the action at the given index is raised.""" - response = _response( - "XMO_REQUEST_ACTION_ERR", - ["XMO_NO_ERR", "XMO_AUTHENTICATION_ERR"], - ) - - with pytest.raises(AuthenticationException): - ActionErrorHandler.throw_if_error_at(response, 1) - - -def test_throw_if_error_at_noop_when_action_ok(): - """No exception is raised when the indexed action reports XMO_NO_ERR.""" - response = _response("XMO_REQUEST_ACTION_ERR", ["XMO_NO_ERR"]) +@pytest.mark.parametrize("path_error", ["XMO_INVALID_PATH_ERR", "XMO_UNKNOWN_PATH_ERR"]) +def test_throw_if_error_ignore_unknown_paths_noop_when_only_path_errors(path_error): + """Both missing-path descriptions are tolerated when explicitly ignored.""" + response = _response("XMO_REQUEST_ACTION_ERR", ["XMO_NO_ERR", path_error]) - ActionErrorHandler.throw_if_error_at(response, 0) + ActionErrorHandler.throw_if_error(response, ignore_unknown_paths=True) -def test_throw_if_error_at_noop_for_out_of_range_index(): - """An out-of-range index is tolerated rather than raising IndexError.""" - response = _response("XMO_REQUEST_ACTION_ERR", ["XMO_UNKNOWN_PATH_ERR"]) - - ActionErrorHandler.throw_if_error_at(response, 5) - - -def test_throw_if_error_at_ignore_unknown_path(): - """With ignore_unknown_path, an unknown-path error at the index is tolerated.""" - response = _response("XMO_REQUEST_ACTION_ERR", ["XMO_UNKNOWN_PATH_ERR"]) - - ActionErrorHandler.throw_if_error_at(response, 0, ignore_unknown_path=True) - - -def test_throw_if_error_at_still_raises_non_unknown_path_when_ignoring(): - """ignore_unknown_path must not suppress other error types.""" - response = _response("XMO_REQUEST_ACTION_ERR", ["XMO_AUTHENTICATION_ERR"]) +@pytest.mark.parametrize( + ("request_desc", "expected"), + [ + ("XMO_AUTHENTICATION_ERR", AuthenticationException), + ("XMO_ACCESS_RESTRICTION_ERR", AccessRestrictionException), + ("XMO_MAX_SESSION_COUNT_ERR", MaximumSessionCountException), + ("XMO_LOGIN_RETRY_ERR", LoginRetryErrorException), + ], +) +@pytest.mark.parametrize("ignore_unknown_paths", [False, True]) +def test_throw_if_error_request_level_error_wins_over_action_placeholder(request_desc, expected, ignore_unknown_paths): + """A request-level error is raised even if the actions only report a missing path.""" + response = _response(request_desc, ["XMO_UNKNOWN_PATH_ERR"]) - with pytest.raises(AuthenticationException): - ActionErrorHandler.throw_if_error_at(response, 0, ignore_unknown_path=True) + with pytest.raises(expected): + ActionErrorHandler.throw_if_error(response, ignore_unknown_paths=ignore_unknown_paths) # --- from_error_description ------------------------------------------------ @@ -127,6 +101,7 @@ def test_throw_if_error_at_still_raises_non_unknown_path_when_ignoring(): ("XMO_AUTHENTICATION_ERR", AuthenticationException), ("XMO_ACCESS_RESTRICTION_ERR", AccessRestrictionException), ("XMO_NON_WRITABLE_PARAMETER_ERR", NonWritableParameterException), + ("XMO_INVALID_PATH_ERR", UnknownPathException), ("XMO_UNKNOWN_PATH_ERR", UnknownPathException), ("XMO_MAX_SESSION_COUNT_ERR", MaximumSessionCountException), ("XMO_LOGIN_RETRY_ERR", LoginRetryErrorException), diff --git a/tests/unit/test_client_basic.py b/tests/unit/test_client_basic.py index e71561a..7b24cd5 100644 --- a/tests/unit/test_client_basic.py +++ b/tests/unit/test_client_basic.py @@ -2,6 +2,7 @@ # pylint: disable=protected-access +from copy import deepcopy from unittest.mock import AsyncMock, MagicMock import pytest @@ -9,7 +10,13 @@ from sagemcom_api.client import SagemcomClient from sagemcom_api.enums import EncryptionMethod -from sagemcom_api.exceptions import AuthenticationException, UnknownPathException +from sagemcom_api.exceptions import ( + AuthenticationException, + InvalidSessionException, + LoginRetryErrorException, + NonWritableParameterException, + UnknownPathException, +) @pytest.mark.asyncio @@ -177,7 +184,7 @@ async def test_login_with_preconfigured_fixture(mock_client_sha512): @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.""" + """Test that ignore_unknown_paths=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", @@ -188,7 +195,7 @@ async def test_get_value_by_xpath_suppresses_unknown_path(mock_session_factory, ) await client.login() - result = await client.get_value_by_xpath("Device/NonExistent", suppress_action_errors=True) + result = await client.get_value_by_xpath("Device/NonExistent", ignore_unknown_paths=True) assert result is None @@ -197,7 +204,7 @@ async def test_get_value_by_xpath_suppresses_unknown_path(mock_session_factory, 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.""" + """Test that ignore_unknown_paths=False (default) raises UnknownPathException.""" mock_session = mock_session_factory([login_success_response, xpath_unknown_path_error_response]) client = SagemcomClient( host="192.168.1.1", @@ -216,7 +223,7 @@ async def test_get_value_by_xpath_raises_unknown_path_when_not_suppressed( 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.""" + """Test that ignore_unknown_paths=True still raises AuthenticationException.""" # Auth errors trigger one re-login and retry (max_tries=2) before propagating. mock_session = mock_session_factory([login_success_response, login_auth_error_response] * 2) client = SagemcomClient( @@ -229,7 +236,7 @@ async def test_get_value_by_xpath_still_raises_auth_error_when_suppressed( await client.login() with pytest.raises(AuthenticationException): - await client.get_value_by_xpath("Device/DeviceInfo", suppress_action_errors=True) + await client.get_value_by_xpath("Device/DeviceInfo", ignore_unknown_paths=True) assert mock_session.post.call_count == 4 @@ -254,7 +261,7 @@ async def test_get_values_by_xpaths_suppresses_unknown_path_per_action( "mac_address": "Device/DeviceInfo/MACAddress", "model_number": "Device/DeviceInfo/ModelNumber", }, - suppress_action_errors=True, + ignore_unknown_paths=True, ) assert result["mac_address"] == "AA:BB:CC:DD:EE:FF" @@ -280,7 +287,7 @@ async def test_get_values_by_xpaths_still_raises_auth_error_when_suppressed( with pytest.raises(AuthenticationException): await client.get_values_by_xpaths( {"mac_address": "Device/DeviceInfo/MACAddress"}, - suppress_action_errors=True, + ignore_unknown_paths=True, ) assert mock_session.post.call_count == 4 @@ -323,3 +330,139 @@ async def test_get_device_info_falls_back_when_aggregate_path_unknown( assert info.serial_number == "SN123456789" assert info.software_version == "1.2.3" assert info.manufacturer == "Sagemcom" + + +def _reply(request_error: str, action_errors: list[str]) -> dict: + """Build a JSON-req reply whose actions carry no values.""" + return { + "reply": { + "uid": 0, + "id": 1, + "error": {"code": 1, "description": request_error}, + "actions": [ + {"uid": i + 1, "id": i, "error": {"code": 1, "description": desc}, "callbacks": []} for i, desc in enumerate(action_errors) + ], + "events": [], + } + } + + +def _md5_client(session) -> SagemcomClient: + return SagemcomClient( + host="192.168.1.1", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + session=session, + ) + + +@pytest.mark.asyncio +async def test_get_values_by_xpaths_ignores_invalid_path_per_action( + mock_session_factory, login_success_response, xpaths_mixed_errors_response +): + """XMO_INVALID_PATH_ERR is treated as a missing path, like XMO_UNKNOWN_PATH_ERR.""" + mixed_response = deepcopy(xpaths_mixed_errors_response) + mixed_response["reply"]["actions"][1]["error"] = {"code": 16777242, "description": "XMO_INVALID_PATH_ERR"} + client = _md5_client(mock_session_factory([login_success_response, mixed_response])) + await client.login() + + result = await client.get_values_by_xpaths( + {"mac_address": "Device/DeviceInfo/MACAddress", "model_number": "Device/DeviceInfo/ModelNumber"}, + ignore_unknown_paths=True, + ) + + assert result == {"mac_address": "AA:BB:CC:DD:EE:FF", "model_number": None} + + +@pytest.mark.parametrize("ignore_unknown_paths", [False, True]) +@pytest.mark.asyncio +async def test_request_level_auth_error_is_not_masked_by_action_placeholder( + mock_session_factory, login_success_response, ignore_unknown_paths +): + """Request-level XMO_AUTHENTICATION_ERR raises, even when actions only report an unknown path.""" + auth_error = _reply("XMO_AUTHENTICATION_ERR", ["XMO_UNKNOWN_PATH_ERR"]) + mock_session = mock_session_factory([login_success_response, auth_error] * 2) + client = _md5_client(mock_session) + await client.login() + + with pytest.raises(AuthenticationException): + await client.get_value_by_xpath("Device/DeviceInfo/ModelNumber", ignore_unknown_paths=ignore_unknown_paths) + + # Re-login and retry once, as for action-level authentication errors. + assert mock_session.post.call_count == 4 + + +@pytest.mark.asyncio +async def test_invalid_session_relogs_in_and_retries( + mock_session_factory, login_success_response, login_invalid_session_response, xpath_value_response +): + """An invalidated session is transient: log in again and retry the request (#520).""" + mock_session = mock_session_factory( + [login_success_response, login_invalid_session_response, login_success_response, xpath_value_response] + ) + client = _md5_client(mock_session) + await client.login() + + value = await client.get_value_by_xpath("Device/WiFi/Radios/Radio[@uid='1']/Status", ignore_unknown_paths=True) + + assert value == "UP" + assert mock_session.post.call_count == 4 + + +@pytest.mark.asyncio +async def test_invalid_session_propagates_after_retry(mock_session_factory, login_success_response, login_invalid_session_response): + """A session that stays invalid surfaces as InvalidSessionException, not an auth failure.""" + mock_session = mock_session_factory([login_success_response, login_invalid_session_response] * 2) + client = _md5_client(mock_session) + await client.login() + + with pytest.raises(InvalidSessionException): + await client.get_value_by_xpath("Device/DeviceInfo") + + +@pytest.mark.asyncio +async def test_set_value_by_xpath_still_raises_action_errors(mock_session_factory): + """Requests other than getters keep raising action-level errors.""" + client = _md5_client(mock_session_factory([_reply("XMO_REQUEST_ACTION_ERR", ["XMO_NON_WRITABLE_PARAMETER_ERR"])])) + + with pytest.raises(NonWritableParameterException): + await client.set_value_by_xpath("Device/DeviceInfo/ModelNumber", "x") + + +@pytest.mark.parametrize("request_error", ["XMO_REQUEST_ACTION_ERR", "XMO_LOGIN_RETRY_ERR"]) +@pytest.mark.asyncio +async def test_login_raises_login_retry_error(mock_session_factory, request_error): + """A locked login raises LoginRetryErrorException, whether reported per action or per request.""" + client = _md5_client(mock_session_factory([_reply(request_error, ["XMO_LOGIN_RETRY_ERR"])])) + + with pytest.raises(LoginRetryErrorException): + await client.login() + + +@pytest.mark.asyncio +async def test_get_device_info_fallback_requires_mac_address( + mock_session_factory, + login_success_response, + xpath_unknown_path_error_response, + device_info_fallback_partial_response, +): + """Missing optional attributes are tolerated, but the MAC address is not.""" + fallback_response = deepcopy(device_info_fallback_partial_response) + fallback_response["reply"]["actions"][0]["error"] = {"code": 16777221, "description": "XMO_UNKNOWN_PATH_ERR"} + fallback_response["reply"]["actions"][0]["callbacks"] = [] + client = _md5_client(mock_session_factory([login_success_response, xpath_unknown_path_error_response, fallback_response])) + await client.login() + + with pytest.raises(UnknownPathException): + await client.get_device_info() + + +@pytest.mark.parametrize("call", ["reboot", "run_speed_test"]) +@pytest.mark.asyncio +async def test_non_getter_actions_raise_action_errors(mock_session_factory, call): + """Reboot and speed test surface action-level errors instead of returning them.""" + client = _md5_client(mock_session_factory([_reply("XMO_REQUEST_ACTION_ERR", ["XMO_AUTHENTICATION_ERR"])])) + + with pytest.raises(AuthenticationException): + await getattr(client, call)()