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"