Add support for catching authentication failures (#257)

src/demetriek/__init__.py

@@ -15,6 +15,7 @@
 )
 from .device import LaMetricDevice
 from .exceptions import (
+    LaMetricAuthenticationError,
     LaMetricConnectionError,
     LaMetricConnectionTimeoutError,
     LaMetricError,
@@ -52,6 +53,7 @@
     "Goal",
     "GoalData",
     "LaMetricCloud",
+    "LaMetricAuthenticationError",
     "LaMetricConnectionError",
     "LaMetricConnectionTimeoutError",
     "LaMetricDevice",

src/demetriek/device.py

@@ -16,6 +16,7 @@
 
 from .const import BrightnessMode
 from .exceptions import (
+    LaMetricAuthenticationError,
     LaMetricConnectionError,
     LaMetricConnectionTimeoutError,
     LaMetricError,
@@ -58,6 +59,7 @@ async def _request(
             LaMetric device.
 
         Raises:
+            LaMetricAuthenticationError: If the API key is invalid.
             LaMetricConnectionError: An error occurred while communication with
                 the LaMetric device.
             LaMetricConnectionTimeoutError: A timeout occurred while communicating
@@ -91,6 +93,14 @@ async def _request(
             raise LaMetricConnectionTimeoutError(
                 f"Timeout occurred while connecting to the LaMetric device at {self.host}"
             ) from exception
+        except aiohttp.ClientResponseError as exception:
+            if exception.status in [401, 403]:
+                raise LaMetricAuthenticationError(
+                    f"Authentication to the LaMetric device at {self.host} failed"
+                ) from exception
+            raise LaMetricError(
+                f"Error occurred while connecting to the LaMetric device at {self.host}"
+            ) from exception
         except (aiohttp.ClientError, socket.gaierror) as exception:
             raise LaMetricConnectionError(
                 f"Error occurred while communicating with the LaMetric device at {self.host}"

src/demetriek/exceptions.py

@@ -9,5 +9,9 @@ class LaMetricConnectionError(LaMetricError):
     """LaMetric connection exception."""
 
 
+class LaMetricAuthenticationError(LaMetricError):
+    """LaMetric authentication exception."""
+
+
 class LaMetricConnectionTimeoutError(LaMetricConnectionError):
     """LaMetric connection Timeout exception."""

tests/test_lametric.py

@@ -6,7 +6,12 @@
 import pytest
 from aresponses import Response, ResponsesMockServer
 
-from demetriek import LaMetricConnectionError, LaMetricDevice, LaMetricError
+from demetriek import (
+    LaMetricAuthenticationError,
+    LaMetricConnectionError,
+    LaMetricDevice,
+    LaMetricError,
+)
 
 
 @pytest.mark.asyncio
@@ -174,3 +179,20 @@ async def test_no_json_response(aresponses: ResponsesMockServer) -> None:
         demetriek = LaMetricDevice("127.0.0.2", api_key="abc", session=session)
         with pytest.raises(LaMetricError):
             assert await demetriek._request("/")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", {401, 403})
+async def test_http_error401(aresponses: ResponsesMockServer, status: int) -> None:
+    """Test HTTP 401 response handling."""
+    aresponses.add(
+        "127.0.0.2:4343",
+        "/",
+        "GET",
+        aresponses.Response(text="Access denied!", status=status),
+    )
+
+    async with aiohttp.ClientSession() as session:
+        demetriek = LaMetricDevice("127.0.0.2", api_key="abc", session=session)
+        with pytest.raises(LaMetricAuthenticationError):
+            assert await demetriek._request("/")