Raise a LaMetricError when a response does not fit its model

src/demetriek/cloud.py

@@ -6,10 +6,12 @@
 import socket
 from dataclasses import dataclass, field
 from http import HTTPStatus
-from typing import Any, Self
+from typing import Any, Self, TypeVar
 
 import aiohttp
 from aiohttp import hdrs
+from mashumaro.exceptions import MissingField
+from mashumaro.mixins.orjson import DataClassORJSONMixin
 from tenacity import (
     retry,
     retry_if_exception_type,
@@ -27,6 +29,8 @@
 )
 from .models import CloudDevice, User
 
+_ModelT = TypeVar("_ModelT", bound=DataClassORJSONMixin)
+
 
 @dataclass
 class LaMetricCloud:
@@ -133,6 +137,33 @@ async def _request(
             msg = "Error occurred while communicating with the LaMetric cloud"
             raise LaMetricConnectionError(msg) from exception
 
+    def _parse(self, model: type[_ModelT], data: Any) -> _ModelT:
+        """Parse a response of the LaMetric cloud into a model.
+
+        Args:
+        ----
+            model: The model to parse the response into.
+            data: The JSON decoded response.
+
+        Returns:
+        -------
+            The model, filled with the response.
+
+        Raises:
+        ------
+            LaMetricError: The response does not fit the model, for example
+                because the cloud API changed.
+
+        """
+        try:
+            return model.from_dict(data)
+        except (MissingField, ValueError) as exception:
+            msg = (
+                "The LaMetric cloud answered with data this library does not"
+                f" understand: {exception}"
+            )
+            raise LaMetricError(msg) from exception
+
     async def current_user(self) -> User:
         """Get LaMetric user information.
 
@@ -142,7 +173,7 @@ async def current_user(self) -> User:
 
         """
         response = await self._request("/api/v2/users/me")
-        return User.from_dict(response)
+        return self._parse(User, response)
 
     async def devices(self) -> list[CloudDevice]:
         """Get LaMetric devices from the cloud.
@@ -153,7 +184,7 @@ async def devices(self) -> list[CloudDevice]:
 
         """
         response = await self._request("/api/v2/users/me/devices")
-        return [CloudDevice.from_dict(cloud_device) for cloud_device in response]
+        return [self._parse(CloudDevice, cloud_device) for cloud_device in response]
 
     async def device(self, device_id: int) -> CloudDevice:
         """Get a LaMetric device from the cloud.
@@ -168,7 +199,7 @@ async def device(self, device_id: int) -> CloudDevice:
 
         """
         response = await self._request(f"/api/v2/users/me/devices/{device_id}")
-        return CloudDevice.from_dict(response)
+        return self._parse(CloudDevice, response)
 
     async def rename_device(self, device_id: int, *, name: str) -> None:
         """Rename a LaMetric device in the cloud.

src/demetriek/device.py

@@ -7,12 +7,13 @@
 import socket
 from dataclasses import dataclass, field
 from http import HTTPStatus
-from typing import TYPE_CHECKING, Any, Self
+from typing import TYPE_CHECKING, Any, Self, TypeVar
 
 import aiohttp
 from aiohttp import hdrs
 from aiohttp.helpers import BasicAuth
-from mashumaro.exceptions import InvalidFieldValue, MissingField
+from mashumaro.exceptions import MissingField
+from mashumaro.mixins.orjson import DataClassORJSONMixin
 from tenacity import (
     RetryCallState,
     retry,
@@ -53,6 +54,8 @@
 
 _LOGGER = logging.getLogger(__package__)
 
+_ModelT = TypeVar("_ModelT", bound=DataClassORJSONMixin)
+
 
 def _can_retry(retry_state: RetryCallState) -> bool:
     """Tell whether a failed request to the device is safe to repeat.
@@ -188,6 +191,33 @@ async def _request(
             )
             raise LaMetricConnectionError(msg) from exception
 
+    def _parse(self, model: type[_ModelT], data: Any) -> _ModelT:
+        """Parse a response of the LaMetric device into a model.
+
+        Args:
+        ----
+            model: The model to parse the response into.
+            data: The JSON decoded response.
+
+        Returns:
+        -------
+            The model, filled with the response.
+
+        Raises:
+        ------
+            LaMetricError: The response does not fit the model, for example
+                because a firmware update changed it.
+
+        """
+        try:
+            return model.from_dict(data)
+        except (MissingField, ValueError) as exception:
+            msg = (
+                f"The LaMetric device at {self.host} answered with data this"
+                f" library does not understand: {exception}"
+            )
+            raise LaMetricError(msg) from exception
+
     async def api(self) -> API:
         """Get the API version and the endpoints the device supports.
 
@@ -197,7 +227,7 @@ async def api(self) -> API:
 
         """
         response = await self._request("/api/v2")
-        return API.from_dict(response)
+        return self._parse(API, response)
 
     async def device(self) -> Device:
         """Get LaMetric device information.
@@ -209,13 +239,15 @@ async def device(self) -> Device:
         """
         response = await self._request("/api/v2/device")
 
-        response["wifi"].update(
-            mac=response["wifi"].get("address", response["wifi"].get("mac")),
-            ssid=response["wifi"].get("essid", response["wifi"].get("ssid")),
-            rssi=response["wifi"].get("strength", response["wifi"].get("rssi")),
-        )
+        # Leave a missing Wi-Fi block to the parsing below, which reports it.
+        if isinstance(wifi := response.get("wifi"), dict):
+            wifi.update(
+                mac=wifi.get("address", wifi.get("mac")),
+                ssid=wifi.get("essid", wifi.get("ssid")),
+                rssi=wifi.get("strength", wifi.get("rssi")),
+            )
 
-        return Device.from_dict(response)
+        return self._parse(Device, response)
 
     async def set_device_mode(self, *, mode: DeviceMode) -> None:
         """Set the mode of the LaMetric device.
@@ -331,10 +363,10 @@ async def display(  # noqa: PLR0913 # pylint: disable=too-many-arguments
                 method=hdrs.METH_PUT,
                 data=data,
             )
-            return Display.from_dict(response["success"]["data"])
+            return self._parse(Display, response["success"]["data"])
 
         response = await self._request("/api/v2/device/display")
-        return Display.from_dict(response)
+        return self._parse(Display, response)
 
     async def audio(self, *, volume: int | None = None) -> Audio:
         """Get or set LaMetric device audio information.
@@ -360,10 +392,10 @@ async def audio(self, *, volume: int | None = None) -> Audio:
                 method=hdrs.METH_PUT,
                 data=data,
             )
-            return Audio.from_dict(response["success"]["data"])
+            return self._parse(Audio, response["success"]["data"])
 
         data = await self._request("/api/v2/device/audio")
-        return Audio.from_dict(data)
+        return self._parse(Audio, data)
 
     async def bluetooth(
         self,
@@ -403,7 +435,7 @@ async def bluetooth(
         # The Bluetooth endpoint calls the address "mac", the device endpoint
         # calls it "address". Only fill it in when the device left it out.
         response.setdefault("address", response.get("mac"))
-        return Bluetooth.from_dict(response)
+        return self._parse(Bluetooth, response)
 
     async def wifi(self) -> Wifi:
         """Get LaMetric device Wi-Fi information.
@@ -418,7 +450,7 @@ async def wifi(self) -> Wifi:
         # Only fill them in when the device left them out.
         data.setdefault("ip", data.get("ipv4"))
         data.setdefault("rssi", data.get("signal_strength"))
-        return Wifi.from_dict(data)
+        return self._parse(Wifi, data)
 
     async def apps(self) -> dict[str, App]:
         """Get the apps installed on LaMetric Time.
@@ -429,7 +461,7 @@ async def apps(self) -> dict[str, App]:
 
         """
         response = await self._request("/api/v2/device/apps")
-        return {package: App.from_dict(app) for package, app in response.items()}
+        return {package: self._parse(App, app) for package, app in response.items()}
 
     async def app(self, *, package: str) -> App:
         """Get a single app installed on LaMetric Time.
@@ -448,7 +480,7 @@ async def app(self, *, package: str) -> App:
 
         """
         response = await self._request(f"/api/v2/device/apps/{package}")
-        return App.from_dict(response)
+        return self._parse(App, response)
 
     async def activate_widget(self, *, package: str, widget_id: str) -> None:
         """Show a specific widget of an app on LaMetric Time.
@@ -621,7 +653,7 @@ async def notification(self, *, notification_id: int) -> Notification:
         response = await self._request(
             f"/api/v2/device/notifications/{notification_id}",
         )
-        return Notification.from_dict(response)
+        return self._parse(Notification, response)
 
     async def notification_current(self) -> Notification | None:
         """Get the current notification.
@@ -633,7 +665,7 @@ async def notification_current(self) -> Notification | None:
 
         """
         if data := await self._request("/api/v2/device/notifications/current"):
-            return Notification.from_dict(data)
+            return self._parse(Notification, data)
         return None
 
     async def notification_queue(self) -> list[Notification]:
@@ -653,8 +685,8 @@ async def notification_queue(self) -> list[Notification]:
         notifications: list[Notification] = []
         for notification in data:
             try:
-                notifications.append(Notification.from_dict(notification))
-            except (InvalidFieldValue, MissingField):
+                notifications.append(self._parse(Notification, notification))
+            except LaMetricError:
                 _LOGGER.warning(
                     "Skipping notification %s, its format is not supported",
                     notification.get("id"),
@@ -673,7 +705,7 @@ async def stream(self) -> Stream:
 
         """
         response = await self._request("/api/v2/device/stream")
-        return Stream.from_dict(response)
+        return self._parse(Stream, response)
 
     async def stream_start(
         self,
@@ -721,7 +753,7 @@ async def stream_start(
 
         # The canvas settings come back nested, flatten them into the session.
         data = response["success"]["data"]
-        return StreamSession.from_dict({**data.pop("canvas"), **data})
+        return self._parse(StreamSession, {**data.pop("canvas"), **data})
 
     async def stream_stop(self) -> None:
         """Stop the stream, so the device returns to normal operation."""

tests/test_cloud.py

@@ -183,6 +183,18 @@ async def test_get_current_user(responses: aioresponses, cloud: LaMetricCloud) -
     assert User.from_dict(user.to_dict()) == user
 
 
+async def test_get_current_user_unexpected_data(
+    responses: aioresponses, cloud: LaMetricCloud
+) -> None:
+    """Test data the library does not understand raises a LaMetricError."""
+    responses.get(f"{CLOUD_URL}/api/v2/users/me", status=200, body='{"id": 1}')
+
+    with pytest.raises(
+        LaMetricError, match='Field "apps_count" of type int is missing'
+    ):
+        await cloud.current_user()
+
+
 async def test_get_devices(responses: aioresponses, cloud: LaMetricCloud) -> None:
     """Test getting devices from the logged in account."""
     responses.get(

tests/test_device.py

@@ -18,6 +18,7 @@
     Goal,
     GoalData,
     LaMetricDevice,
+    LaMetricError,
     Model,
     Notification,
     NotificationIconType,
@@ -68,6 +69,34 @@ async def test_get_device(
     assert Device.from_dict(result.to_dict()) == result
 
 
+async def test_get_device_unexpected_data(
+    responses: aioresponses, device: LaMetricDevice
+) -> None:
+    """Test data the library does not understand raises a LaMetricError.
+
+    So a firmware update that changes a response is reported with what did
+    not fit, rather than with an error of the parsing library underneath.
+    """
+    responses.get(f"{DEVICE_URL}/api/v2/device", status=200, body='{"id": "1"}')
+
+    with pytest.raises(
+        LaMetricError, match=r"The LaMetric device at 127\.0\.0\.2"
+    ) as error:
+        await device.device()
+
+    assert 'Field "display" of type Display is missing in Device' in str(error.value)
+
+
+async def test_api_not_an_object(
+    responses: aioresponses, device: LaMetricDevice
+) -> None:
+    """Test a response that is not an object at all raises a LaMetricError."""
+    responses.get(f"{DEVICE_URL}/api/v2", status=200, body="[]")
+
+    with pytest.raises(LaMetricError, match="should be a dict instance"):
+        await device.api()
+
+
 async def test_notify(responses: aioresponses, device: LaMetricDevice) -> None:
     """Test sending notification serialization."""
     url = f"{DEVICE_URL}/api/v2/device/notifications"