Auth robustness: serialised refresh, every HTTP status mapped, token_update_callback

pyproject.toml

@@ -30,7 +30,6 @@ dependencies = [
     "mashumaro>=3.20",
     "orjson>=3.10.12",
     "yarl>=1.6.0",
-    "aioresponses>=0.7.7,<0.8",
     "pyjwt>=2.11.0",
 ]
 
@@ -43,6 +42,7 @@ Changelog = "https://github.com/erwindouna/python-tado/releases"
 
 [dependency-groups]
 dev = [
+    "aioresponses>=0.7.7,<0.8",
     "aresponses>=3.0.0",
     "codespell>=2.2.6",
     "covdefaults>=2.3.0",
@@ -60,6 +60,12 @@ dev = [
     "deptry>=0.24.0",
 ]
 
+[tool.uv]
+# aioresponses does not support aiohttp 3.14 yet (pnuckowski/aioresponses#289), which
+# breaks the test suite; keep the development environment on aiohttp 3.13 until it does.
+# This only affects uv.lock, not the published requirements.
+constraint-dependencies = ["aiohttp<3.14"]
+
 [tool.hatch.build.targets.sdist]
 include = ["src/tadoasync"]
 

src/tadoasync/__init__.py

@@ -5,6 +5,7 @@
     TadoBadRequestError,
     TadoConnectionError,
     TadoError,
+    TadoRateLimitError,
     TadoReadingError,
     TadoServerError,
 )
@@ -16,6 +17,7 @@
     "TadoBadRequestError",
     "TadoConnectionError",
     "TadoError",
+    "TadoRateLimitError",
     "TadoReadingError",
     "TadoServerError",
 ]

src/tadoasync/exceptions.py

@@ -17,6 +17,10 @@ class TadoServerError(TadoError):
     """Tado server exception."""
 
 
+class TadoRateLimitError(TadoError):
+    """Tado API rate limit (daily quota) exception."""
+
+
 class TadoBadRequestError(TadoError):
     """Tado bad request exception."""
 

src/tadoasync/tadoasync.py

@@ -48,7 +48,9 @@
     TadoConnectionError,
     TadoError,
     TadoForbiddenError,
+    TadoRateLimitError,
     TadoReadingError,
+    TadoServerError,
 )
 from tadoasync.models_v3 import (
     Capabilities,
@@ -64,6 +66,8 @@
 from tadoasync.unifier import get_unifier_from_generation
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from tadoasync import models_unified as unified_models
 
 CLIENT_ID = "1bb50063-6b0c-4d11-bd99-387f4a91cc46"
@@ -98,9 +102,17 @@ def __init__(
         debug: bool | None = None,  # noqa: FBT001
         session: ClientSession | None = None,
         request_timeout: int = 10,
+        token_update_callback: Callable[[str], None] | None = None,
     ) -> None:
-        """Initialize the Tado object."""
+        """Initialize the Tado object.
+
+        token_update_callback, if given, is called with the new refresh token
+        every time tado issues one (device login and every refresh). tado refresh
+        tokens are single use, so a caller that stores the token should save it
+        here rather than only after its own calls succeed.
+        """
         self._refresh_token = refresh_token
+        self._token_update_callback = token_update_callback
         self._debug: bool = debug or False
         self._session = session
         self._request_timeout = request_timeout
@@ -113,6 +125,8 @@ def __init__(
 
         self._access_token: str | None = None
         self._token_expiry: float | None = None
+        # tado refresh tokens are single use: only one refresh may run at a time
+        self._refresh_lock = asyncio.Lock()
         self._access_headers: dict[str, str] | None = None
         self._home_id: int | None = None
         self._me: GetMe | None = None
@@ -284,7 +298,7 @@ async def _check_device_activation(self) -> bool:
             response = await request.json()
             self._access_token = response["access_token"]
             self._token_expiry = time.time() + float(response["expires_in"])
-            self._refresh_token = response["refresh_token"]
+            self._set_refresh_token(response["refresh_token"])
 
             self._set_home_id_from_access_token()
             return True
@@ -345,7 +359,7 @@ async def login(self) -> None:
         response = await request.json()
         self._access_token = response["access_token"]
         self._token_expiry = time.time() + float(response["expires_in"])
-        self._refresh_token = response["refresh_token"]
+        self._set_refresh_token(response["refresh_token"])
 
         self._set_home_id_from_access_token()
 
@@ -369,39 +383,59 @@ async def check_request_status(
         self, response_error: ClientResponseError, *, login: bool = False
     ) -> None:
         """Check the status of the request and raise the proper exception if needed."""
-        status_error_mapping = {
-            500: TadoError(
-                "Error "
-                + str(response_error.status)
-                + " connecting to Tado. Response body: "
-                + response_error.message
-            ),
-            401: TadoAuthenticationError(
-                "Authentication error connecting to Tado. Response body: "
-                + response_error.message
-            ),
-            403: TadoForbiddenError(
-                "Forbidden error connecting to Tado. Response body: "
-                + response_error.message
-            ),
-        }
-
-        status_error_mapping[400] = TadoBadRequestError(
-            "Bad request to Tado. Response body: " + response_error.message
-        )
-        if login:
-            status_error_mapping[400] = TadoAuthenticationError(
-                "Authentication error connecting to Tado. Response body: "
-                + response_error.message
+        status = response_error.status
+        body = response_error.message
+        if status == 401 or (status == 400 and login):
+            raise TadoAuthenticationError(
+                "Authentication error connecting to Tado. Response body: " + body
             )
-
-        raise status_error_mapping[response_error.status]
+        if status == 400:
+            raise TadoBadRequestError("Bad request to Tado. Response body: " + body)
+        if status == 403:
+            raise TadoForbiddenError(
+                "Forbidden error connecting to Tado. Response body: " + body
+            )
+        if status == 429:
+            raise TadoRateLimitError(
+                "Tado API rate limit reached. Response body: " + body
+            )
+        if status >= 500:
+            raise TadoServerError(
+                "Error " + str(status) + " connecting to Tado. Response body: " + body
+            )
+        raise TadoError(
+            "Unexpected status " + str(status) + " from Tado. Response body: " + body
+        )
 
     async def _refresh_auth(self) -> None:
         """Refresh the authentication token."""
-        if self._token_expiry is not None and time.time() < self._token_expiry - 30:
+        if self._token_is_fresh():
             return
 
+        # Several requests can find the token expired at the same time. Only the
+        # first one refreshes; the others wait and then use its new token instead
+        # of sending the refresh token it has just used up (tado answers 400).
+        async with self._refresh_lock:
+            if self._token_is_fresh():
+                return
+            await self._refresh_auth_locked()
+
+    def _set_refresh_token(self, refresh_token: str) -> None:
+        """Store a newly issued refresh token and hand it to the callback."""
+        self._refresh_token = refresh_token
+        if self._token_update_callback is None:
+            return
+        try:
+            self._token_update_callback(refresh_token)
+        except Exception:  # pylint: disable=broad-exception-caught
+            _LOGGER.exception("Token update callback failed")
+
+    def _token_is_fresh(self) -> bool:
+        """Return True while the access token is valid for at least 30 more seconds."""
+        return self._token_expiry is not None and time.time() < self._token_expiry - 30
+
+    async def _refresh_auth_locked(self) -> None:
+        """Refresh the authentication token; the caller holds the refresh lock."""
         data = {
             "client_id": CLIENT_ID,
             "grant_type": "refresh_token",
@@ -420,12 +454,13 @@ async def _refresh_auth(self) -> None:
                 "Timeout occurred while connecting to Tado."
             ) from err
         except ClientResponseError as err:
-            await self.check_request_status(err)
+            # 400 here is invalid_grant: tado refused the refresh token
+            await self.check_request_status(err, login=True)
 
         response = await request.json()
         self._access_token = response["access_token"]
         self._token_expiry = time.time() + float(response["expires_in"])
-        self._refresh_token = response["refresh_token"]
+        self._set_refresh_token(response["refresh_token"])
 
         _LOGGER.debug("Tado token refreshed")
 

tests/conftest.py

@@ -43,7 +43,7 @@ async def client() -> AsyncGenerator[Tado]:
 @pytest.fixture(autouse=True)
 def _tado_oauth(responses: aioresponses) -> None:
     """Mock the Tado token URL."""
-    auth_token = load_fixture("auth_token.txt")
+    auth_token = load_fixture("auth_token.txt").strip()
 
     responses.post(
         TADO_DEVICE_AUTH_URL,

tests/test_tado.py

@@ -23,7 +23,10 @@
     TadoBadRequestError,
     TadoConnectionError,
     TadoError,
+    TadoForbiddenError,
+    TadoRateLimitError,
     TadoReadingError,
+    TadoServerError,
 )
 from tadoasync.tadoasync import DEVICE_AUTH_URL, DeviceActivationStatus
 
@@ -34,7 +37,7 @@
 if TYPE_CHECKING:
     from syrupy import SnapshotAssertion
 
-AUTH_TOKEN = load_fixture("auth_token.txt")
+AUTH_TOKEN = load_fixture("auth_token.txt").strip()
 
 
 async def test_create_session(
@@ -302,10 +305,123 @@ async def mock_post(*args: Any, **kwargs: Any) -> ClientResponse:  # noqa: ARG00
         python_tado._access_token = "old_test_access_token"
         python_tado._token_expiry = time.time() - 10  # make sure the token is expired
         python_tado._refresh_token = "old_test_refresh_token"
-        with pytest.raises(TadoBadRequestError):
+        # a 400 from the token endpoint means the refresh token was refused
+        with pytest.raises(TadoAuthenticationError):
             await python_tado._refresh_auth()
 
 
+@pytest.mark.parametrize(
+    ("status", "login", "expected"),
+    [
+        (400, False, TadoBadRequestError),
+        (400, True, TadoAuthenticationError),
+        (401, False, TadoAuthenticationError),
+        (403, False, TadoForbiddenError),
+        (404, False, TadoError),
+        (429, False, TadoRateLimitError),
+        (500, False, TadoServerError),
+        (502, False, TadoServerError),
+        (503, False, TadoServerError),
+    ],
+)
+async def test_check_request_status_maps_every_status(
+    python_tado: Tado, status: int, *, login: bool, expected: type[TadoError]
+) -> None:
+    """Every HTTP error maps to a Tado exception, never a KeyError."""
+    error = ClientResponseError(
+        MagicMock(spec=RequestInfo), (), status=status, message="Error message"
+    )
+    with pytest.raises(expected) as exc_info:
+        await python_tado.check_request_status(error, login=login)
+    assert "Error message" in str(exc_info.value)
+
+
+async def test_parallel_requests_refresh_token_once(python_tado: Tado) -> None:
+    """Parallel requests at token expiry refresh once and reuse the result.
+
+    tado refresh tokens are single use. Without serialisation every coroutine
+    sends the same refresh token and all but the first get HTTP 400.
+    """
+    valid = {"rotating_0"}
+    calls = {"refresh": 0, "rejected": 0}
+
+    async def mock_post(*args: Any, **kwargs: Any) -> ClientResponse:  # noqa: ARG001 # pylint: disable=unused-argument
+        sent = kwargs["data"]["refresh_token"]
+        await asyncio.sleep(0.01)  # a real round trip; lets the others catch up
+        calls["refresh"] += 1
+        response = MagicMock(spec=ClientResponse)
+        if sent not in valid:
+            calls["rejected"] += 1
+            response.raise_for_status.side_effect = ClientResponseError(
+                MagicMock(spec=RequestInfo), (), status=400, message="invalid_grant"
+            )
+            return response
+        valid.discard(sent)
+        new = f"rotating_{calls['refresh']}"
+        valid.add(new)
+        response.json = AsyncMock(
+            return_value={
+                "access_token": AUTH_TOKEN,
+                "expires_in": 600,
+                "refresh_token": new,
+            }
+        )
+        return response
+
+    with patch("aiohttp.ClientSession.post", new=mock_post):
+        python_tado._token_expiry = time.time() - 10
+        python_tado._refresh_token = "rotating_0"
+        results = await asyncio.gather(
+            *(python_tado._refresh_auth() for _ in range(6)), return_exceptions=True
+        )
+
+    assert [r for r in results if isinstance(r, Exception)] == []
+    assert calls == {"refresh": 1, "rejected": 0}
+    assert python_tado.refresh_token == "rotating_1"
+
+
+async def test_token_update_callback_gets_every_new_token(
+    responses: aioresponses,
+) -> None:
+    """The callback sees the token from the device login and from each refresh."""
+    seen: list[str] = []
+    async with aiohttp.ClientSession() as session:
+        tado = Tado(session=session, token_update_callback=seen.append)
+        await tado.async_init()
+        await tado.device_activation()
+        assert seen == ["test_refresh_token"]
+
+        responses.post(
+            TADO_TOKEN_URL,
+            status=200,
+            payload={
+                "access_token": AUTH_TOKEN,
+                "expires_in": 600,
+                "refresh_token": "second_refresh_token",
+            },
+        )
+        tado._token_expiry = time.time() - 10
+        await tado._refresh_auth()
+        assert seen == ["test_refresh_token", "second_refresh_token"]
+
+
+async def test_token_update_callback_error_does_not_break_auth(
+    responses: aioresponses,  # noqa: ARG001 # pylint: disable=unused-argument
+) -> None:
+    """A failing callback is logged; authentication still completes."""
+
+    def broken(token: str) -> None:  # noqa: ARG001 # pylint: disable=unused-argument
+        msg = "storage full"
+        raise RuntimeError(msg)
+
+    async with aiohttp.ClientSession() as session:
+        tado = Tado(session=session, token_update_callback=broken)
+        await tado.async_init()
+        await tado.device_activation()
+        assert tado.refresh_token == "test_refresh_token"
+        assert tado.device_activation_status == DeviceActivationStatus.COMPLETED
+
+
 async def test_device_flow_sets_verification_url(responses: aioresponses) -> None:
     """Test device flow sets verification URL."""
     responses.post(