Handle non-UTF-8 and invalid JSON responses from WLED device

src/wled/exceptions.py

@@ -9,6 +9,10 @@ class WLEDEmptyResponseError(Exception):
     """WLED empty API response exception."""
 
 
+class WLEDInvalidResponseError(WLEDError):
+    """WLED invalid API response exception."""
+
+
 class WLEDConnectionError(WLEDError):
     """WLED connection exception."""
 

src/wled/wled.py

@@ -21,6 +21,7 @@
     WLEDConnectionTimeoutError,
     WLEDEmptyResponseError,
     WLEDError,
+    WLEDInvalidResponseError,
     WLEDUpgradeError,
 )
 from .models import Device, Playlist, Preset, Releases
@@ -233,10 +234,20 @@ async def request(
                     {"message": contents.decode("utf8")},
                 )
 
-            response_data = await response.text()
+            try:
+                response_data = await response.text()
+            except UnicodeDecodeError as exception:
+                msg = f"Received a non-UTF-8 response from request: {method} {uri}"
+                raise WLEDInvalidResponseError(msg) from exception
             if "application/json" in content_type:
-                response_data = orjson.loads(response_data)
-
+                try:
+                    response_data = orjson.loads(response_data)
+                except orjson.JSONDecodeError as exception:
+                    msg = (
+                        "Received an invalid JSON response "
+                        f"from request: {method} {uri}"
+                    )
+                    raise WLEDInvalidResponseError(msg) from exception
         except TimeoutError as exception:
             msg = f"Timeout occurred while connecting to WLED device at {self.host}"
             raise WLEDConnectionTimeoutError(msg) from exception

tests/conftest.py

@@ -7,6 +7,7 @@
 
 import aiohttp
 import pytest
+import pytest_asyncio
 from aioresponses import aioresponses
 
 from wled import WLED
@@ -28,18 +29,26 @@ def full_device_data() -> dict[str, Any]:
     return data
 
 
-def mock_json_and_presets(mocked: aioresponses) -> None:
+def mock_json_and_presets(
+    mocked: aioresponses,
+    wled_data: dict[str, Any] | None = None,
+    presets_data: dict[str, Any] | None = None,
+) -> None:
     """Register the two GET endpoints that WLED.update() calls."""
+    if wled_data is None:
+        wled_data = load_fixture_json("wled")
     mocked.get(
         "http://example.com/json",
         status=200,
-        body=json.dumps(load_fixture_json("wled")),
+        body=json.dumps(wled_data),
         content_type="application/json",
     )
+    if presets_data is None:
+        presets_data = load_fixture_json("presets")
     mocked.get(
         "http://example.com/presets.json",
         status=200,
-        body=json.dumps(load_fixture_json("presets")),
+        body=json.dumps(presets_data),
         content_type="application/json",
     )
 
@@ -57,8 +66,16 @@ def responses() -> Generator[aioresponses, None, None]:
         yield mocker
 
 
-@pytest.fixture
+@pytest_asyncio.fixture
+async def session() -> AsyncGenerator[aiohttp.ClientSession, None]:
+    """Yield a new aiohttp client session."""
+    async with aiohttp.ClientSession() as aio_session:
+        yield aio_session
+
+
+@pytest_asyncio.fixture
 async def wled() -> AsyncGenerator[WLED, None]:
     """Yield a WLED client wired to example.com with default settings."""
-    async with aiohttp.ClientSession() as session:
-        yield WLED("example.com", session=session)
+    wled_instance = WLED("example.com")
+    yield wled_instance
+    await wled_instance.close()