From 5e1433cc7aebfbaa21cd367ae36517bf37babbca Mon Sep 17 00:00:00 2001 From: mr_miles Date: Mon, 6 Jul 2026 16:40:51 +0100 Subject: [PATCH] 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. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../action_error_exception_handler.py | 97 ++++++++++++ sagemcom_api/client.py | 97 ++++++------ 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 | 146 +++++++++++++++++- 8 files changed, 649 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..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 3a16e64..ed0cfde 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -22,33 +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_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, @@ -240,37 +231,12 @@ async def __post(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 == 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 @@ -339,6 +305,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: @@ -352,7 +320,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 = "" @@ -381,13 +350,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( @@ -401,11 +377,18 @@ 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: 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 = [ { @@ -418,7 +401,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 @@ -461,6 +453,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( @@ -488,7 +482,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" @@ -554,6 +550,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}" @@ -571,6 +569,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 @@ -585,7 +585,10 @@ async def run_speed_test(self, block_traffic: bool = False): "parameters": {"BlockTraffic": block_traffic}, } ] - return await self.__api_request_async(actions, False) + response = await self.__api_request_async(actions, 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 031abd4..bfed260 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -64,6 +64,29 @@ 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 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 d7ba7db..52ce052 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,147 @@ 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 +): + """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.""" + 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, + ) + + +@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"