Handle invalid UTF-8 in router responses

sagemcom_api/client.py

@@ -228,7 +228,11 @@ async def __post(self, url, data):
                 result = await response.text()
                 raise UnknownException(result)
 
-            result = await response.json()
+            try:
+                result = await response.json()
+            except UnicodeDecodeError:
+                # Some router firmware emits otherwise valid JSON containing non-UTF-8 bytes.
+                result = json.loads((await response.read()).decode("utf-8", errors="replace"))
             error = self.__get_response_error(result)
 
             # No errors

tests/unit/test_client_basic.py

@@ -2,13 +2,39 @@
 
 # pylint: disable=protected-access
 
+from unittest.mock import AsyncMock, MagicMock
+
 import pytest
+from aiohttp import ClientSession
 
 from sagemcom_api.client import SagemcomClient
 from sagemcom_api.enums import EncryptionMethod
 from sagemcom_api.exceptions import AuthenticationException, UnknownPathException
 
 
+@pytest.mark.asyncio
+async def test_response_with_invalid_utf8_is_parsed():
+    """Test malformed UTF-8 in a router response does not discard valid JSON."""
+    response_body = b'{"reply":{"error":{"description":"XMO_REQUEST_NO_ERR"},"value":"invalid \x8e byte"}}'
+    invalid_byte = response_body.index(b"\x8e")
+    decode_error = UnicodeDecodeError("utf-8", response_body, invalid_byte, invalid_byte + 1, "invalid start byte")
+
+    response = AsyncMock()
+    response.status = 200
+    response.json = AsyncMock(side_effect=decode_error)
+    response.read = AsyncMock(return_value=response_body)
+    response.__aenter__ = AsyncMock(return_value=response)
+    response.__aexit__ = AsyncMock(return_value=None)
+
+    session = MagicMock(spec=ClientSession)
+    session.post.return_value = response
+    client = SagemcomClient(host="192.168.1.1", username="admin", password="admin", session=session)
+
+    result = await client._SagemcomClient__post("http://192.168.1.1/ws", {})
+
+    assert result["reply"]["value"] == "invalid � byte"
+
+
 @pytest.mark.asyncio
 async def test_login_success(mock_session_factory, login_success_response):
     """Test successful login with mocked session.