diff --git a/sagemcom_api/action_error_exception_handler.py b/sagemcom_api/action_error_exception_handler.py new file mode 100644 index 0000000..f9a4486 --- /dev/null +++ b/sagemcom_api/action_error_exception_handler.py @@ -0,0 +1,92 @@ +"""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, + BaseSagemcomException, + LoginRetryErrorException, + MaximumSessionCountException, + NonWritableParameterException, + UnknownException, + 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: + """Raise exceptions for errors reported in a JSON-req reply.""" + + @staticmethod + 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_paths: skip actions whose path does not exist, + while still raising any other error + """ + 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: + continue + + if ignore_unknown_paths and action_error_desc in PATH_ERRORS: + continue + + raise ActionErrorHandler.from_error_description(action_error, action_error_desc) + + @staticmethod + 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 + + 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 PATH_ERRORS: + 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..3adf14d 100644 --- a/sagemcom_api/client.py +++ b/sagemcom_api/client.py @@ -22,34 +22,23 @@ TCPConnector, ) +from .action_error_exception_handler import ActionErrorHandler from .const import ( API_ENDPOINT, DEFAULT_TIMEOUT, DEFAULT_USER_AGENT, UINT_MAX, - XMO_ACCESS_RESTRICTION_ERR, - XMO_AUTHENTICATION_ERR, - XMO_INVALID_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,44 +250,13 @@ async def __send(self, url, data): self._request_id = -1 raise InvalidSessionException(error) - # 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 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) - 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() @@ -321,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, @@ -329,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 = { @@ -409,13 +369,15 @@ 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, ignore_unknown_paths: bool = False) -> Any: """Retrieve raw value from router using XPath. :param xpath: path expression :param options: optional options + :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) + result = await self.get_values_by_xpaths({"value": xpath}, options, ignore_unknown_paths) return result["value"] @backoff.on_exception( @@ -429,11 +391,13 @@ 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, 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 ignore_unknown_paths: return None for each path that does not exist instead + of raising UnknownPathException; other errors are still raised """ actions = [ { @@ -445,7 +409,7 @@ async def get_values_by_xpaths(self, xpaths: dict[str, str], options: dict | Non for i, xpath in enumerate(xpaths.values()) ] - response = await self.__api_request_async(actions, False) + 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)) @@ -516,10 +480,16 @@ async def get_device_info(self) -> DeviceInfo: "product_class": "Device/DeviceInfo/ProductClass", "serial_number": "Device/DeviceInfo/SerialNumber", "software_version": "Device/DeviceInfo/SoftwareVersion", - } + }, + # 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( 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..206fdfe --- /dev/null +++ b/tests/unit/test_action_error_handler.py @@ -0,0 +1,117 @@ +"""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 -------------------------------------------------------- + + +@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. + 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_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_INVALID_PATH_ERR", "XMO_AUTHENTICATION_ERR"], + ) + + with pytest.raises(AuthenticationException): + ActionErrorHandler.throw_if_error(response, ignore_unknown_paths=True) + + +@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(response, ignore_unknown_paths=True) + + +@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(expected): + ActionErrorHandler.throw_if_error(response, ignore_unknown_paths=ignore_unknown_paths) + + +# --- from_error_description ------------------------------------------------ + + +@pytest.mark.parametrize( + ("description", "expected"), + [ + ("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), + ("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..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 @@ -173,3 +180,289 @@ 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 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", + username="admin", + password="admin", + authentication_method=EncryptionMethod.MD5, + session=mock_session, + ) + await client.login() + + result = await client.get_value_by_xpath("Device/NonExistent", ignore_unknown_paths=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 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", + 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 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( + 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", ignore_unknown_paths=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", + }, + ignore_unknown_paths=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"}, + ignore_unknown_paths=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" + + +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)()