diff --git a/src/demetriek/cloud.py b/src/demetriek/cloud.py index 7d340428..2f875dcb 100644 --- a/src/demetriek/cloud.py +++ b/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. diff --git a/src/demetriek/device.py b/src/demetriek/device.py index 569d568d..4d48ba42 100644 --- a/src/demetriek/device.py +++ b/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.""" diff --git a/tests/test_cloud.py b/tests/test_cloud.py index 7f73d75f..6daec32c 100644 --- a/tests/test_cloud.py +++ b/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( diff --git a/tests/test_device.py b/tests/test_device.py index fb058086..ebc29521 100644 --- a/tests/test_device.py +++ b/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"