Decouple from Home Assistant with an AbstractAuth interface

pyhilo/__init__.py

@@ -1,6 +1,7 @@
 """Define the hilo package."""
 
 from pyhilo.api import API
+from pyhilo.auth import AbstractAuth
 from pyhilo.const import UNMONITORED_DEVICES
 from pyhilo.device import HiloDevice
 from pyhilo.device.switch import Switch
@@ -12,6 +13,7 @@
 
 __all__ = [
     "API",
+    "AbstractAuth",
     "Devices",
     "HiloDevice",
     "Event",

pyhilo/api.py

@@ -13,8 +13,8 @@
 from aiohttp import ClientSession
 from aiohttp.client_exceptions import ClientResponseError
 import backoff
-from homeassistant.helpers.config_entry_oauth2_flow import OAuth2Session
 
+from pyhilo.auth import AbstractAuth, _LegacyOAuthSessionAuth
 from pyhilo.const import (
     ANDROID_CLIENT_ENDPOINT,
     ANDROID_CLIENT_HEADERS,
@@ -62,7 +62,8 @@ def __init__(
         self,
         *,
         session: ClientSession,
-        oauth_session: OAuth2Session,
+        auth: AbstractAuth | None = None,
+        oauth_session: Any = None,
         request_retries: int = REQUEST_RETRY,
         log_traces: bool = False,
     ) -> None:
@@ -74,7 +75,8 @@ def __init__(
         self.async_request = self._wrap_request_method(self._request_retries)
         self.device_attributes = get_device_attributes()
         self.session: ClientSession = session
-        self._oauth_session = oauth_session
+        self._auth = self._resolve_auth(auth, oauth_session)
+        self._access_token: str | None = None
         self.signalr_devices: SignalRHub
         self.signalr_challenges: SignalRHub
         self.log_traces = log_traces
@@ -84,26 +86,41 @@ def __init__(
         self._device_cache: list[dict[str, Any]] = []
         self._device_cache_event: asyncio.Event = asyncio.Event()
 
+    @staticmethod
+    def _resolve_auth(auth: AbstractAuth | None, oauth_session: Any) -> AbstractAuth:
+        """Pick the auth provider, adapting the deprecated ``oauth_session``."""
+        if auth is not None and oauth_session is not None:
+            raise ValueError("Pass either auth or oauth_session, not both")
+        if auth is not None:
+            return auth
+        if oauth_session is not None:
+            return _LegacyOAuthSessionAuth(oauth_session)
+        raise ValueError("An auth provider is required")
+
     @classmethod
     async def async_create(
         cls,
         *,
         session: ClientSession,
-        oauth_session: OAuth2Session,
+        auth: AbstractAuth | None = None,
+        oauth_session: Any = None,
         request_retries: int = REQUEST_RETRY,
         log_traces: bool = False,
     ) -> API:
         """Get an authenticated API object.
         :param session: The ``aiohttp`` ``ClientSession`` session used for all HTTP requests
         :type session: ``aiohttp.client.ClientSession``
-        :param oauth_session: The session to make requests authenticated with OAuth2.
-        :type oauth_session: ``config_entry_oauth2_flow.OAuth2Session``
+        :param auth: Provides valid access tokens for API requests.
+        :type auth: :class:`pyhilo.auth.AbstractAuth`
+        :param oauth_session: Deprecated, use ``auth`` instead. A Home Assistant
+            ``OAuth2Session``, adapted internally.
         :param request_retries: The default number of request retries to use
         :type request_retries: ``int``
         :rtype: :meth:`pyhilo.api.API`
         """
         api = cls(
             session=session,
+            auth=auth,
             oauth_session=oauth_session,
             request_retries=request_retries,
             log_traces=log_traces,
@@ -128,16 +145,12 @@ def headers(self) -> dict[str, Any]:
 
     async def async_get_access_token(self) -> str:
         """Return a valid access token."""
-        if not self._oauth_session.valid_token:
-            await self._oauth_session.async_ensure_token_valid()
-
-        access_token = str(self._oauth_session.token["access_token"])
-        LOG.debug("SignalR access token is %s", access_token)
+        self._access_token = await self._auth.async_get_access_token()
 
         urn = self.urn
         LOG.debug("Extracted URN: %s", urn)
 
-        return str(self._oauth_session.token["access_token"])
+        return self._access_token
 
     @property
     def urn(self) -> str | None:
@@ -146,10 +159,9 @@ def urn(self) -> str | None:
             return self._urn
 
         try:
-            if not self._oauth_session.valid_token:
+            if not self._access_token:
                 return None
-            token = self._oauth_session.token["access_token"]
-            payload_part = token.split(".")[1]
+            payload_part = self._access_token.split(".")[1]
             # Add padding if necessary
             padding = 4 - len(payload_part) % 4
             if padding != 4:

pyhilo/auth.py

@@ -0,0 +1,42 @@
+"""Authentication interface for the Hilo API."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+from pyhilo.const import LOG
+
+
+class AbstractAuth(ABC):
+    """Supplies valid Hilo access tokens to :class:`pyhilo.api.API`.
+
+    Applications subclass this and implement :meth:`async_get_access_token`.
+    In Home Assistant, the integration wraps its ``OAuth2Session``; standalone
+    scripts can return a static token or run their own refresh logic.
+    """
+
+    @abstractmethod
+    async def async_get_access_token(self) -> str:
+        """Return a valid access token, refreshing it if needed."""
+
+
+class _LegacyOAuthSessionAuth(AbstractAuth):
+    """Adapts a Home Assistant ``OAuth2Session`` passed as ``oauth_session``.
+
+    Deprecated: kept so integration releases that still pass ``oauth_session``
+    keep working. The session is duck-typed so pyhilo never imports
+    Home Assistant.
+    """
+
+    def __init__(self, oauth_session: Any) -> None:
+        LOG.warning(
+            "Passing oauth_session to pyhilo.API is deprecated and will be "
+            "removed in a future release; pass an AbstractAuth via auth instead"
+        )
+        self._oauth_session = oauth_session
+
+    async def async_get_access_token(self) -> str:
+        if not self._oauth_session.valid_token:
+            await self._oauth_session.async_ensure_token_valid()
+        return str(self._oauth_session.token["access_token"])

tests/test_api_auth.py

@@ -0,0 +1,135 @@
+import base64
+import json
+import logging
+import sys
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from pyhilo import API, AbstractAuth
+from pyhilo.const import API_HOSTNAME
+
+URN = "urn:hilo:crm:abc-123"
+
+
+def _jwt(claims: dict[str, Any]) -> str:
+    payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).rstrip(b"=")
+    return f"header.{payload.decode()}.signature"
+
+
+TOKEN = _jwt({"urn:com:hiloenergie:profile:location_hilo_id": [URN]})
+
+
+@pytest.fixture
+def mock_session() -> MagicMock:
+    """Fixture providing a mock aiohttp ClientSession with async context manager response."""
+    session = MagicMock()
+    response = MagicMock()
+    response.headers = {"content-type": "application/json"}
+    response.json = AsyncMock(return_value={"ok": True})
+    response.raise_for_status = MagicMock()
+
+    # Configure async context manager on session.request(...)
+    session.request.return_value.__aenter__ = AsyncMock(return_value=response)
+    session.request.return_value.__aexit__ = AsyncMock(return_value=None)
+    return session
+
+
+def test_import_does_not_require_homeassistant() -> None:
+    assert "homeassistant" not in sys.modules
+
+
+async def test_async_create_fetches_token_through_auth(
+    mock_session: MagicMock,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def _noop(self: API) -> None:
+        pass
+
+    monkeypatch.setattr(API, "_async_post_init", _noop)
+    auth = AsyncMock(spec=AbstractAuth)
+    auth.async_get_access_token.return_value = TOKEN
+
+    api = await API.async_create(session=mock_session, auth=auth)
+
+    auth.async_get_access_token.assert_awaited_once()
+    assert api.urn == URN
+
+
+async def test_each_request_uses_a_fresh_token_from_auth(
+    mock_session: MagicMock,
+) -> None:
+    auth = AsyncMock(spec=AbstractAuth)
+    auth.async_get_access_token.side_effect = [TOKEN, "refreshed-token"]
+    api = API(session=mock_session, auth=auth)
+
+    await api._async_request("get", "/first")
+    await api._async_request("get", "/second")
+
+    assert auth.async_get_access_token.await_count == 2
+    assert mock_session.request.call_count == 2
+
+    first_call_headers = mock_session.request.call_args_list[0].kwargs["headers"]
+    second_call_headers = mock_session.request.call_args_list[1].kwargs["headers"]
+
+    assert first_call_headers["authorization"] == f"Bearer {TOKEN}"
+    assert second_call_headers["authorization"] == "Bearer refreshed-token"
+    assert second_call_headers["Host"] == API_HOSTNAME
+
+
+@pytest.mark.parametrize(
+    ("claims", "expected_urn"),
+    [
+        ({"urn:com:hiloenergie:profile:location_hilo_id": [URN]}, URN),
+        ({"urn:com:hiloenergie:profile:location_hilo_id": []}, None),
+        ({"urn:com:hiloenergie:profile:location_hilo_id": "not-a-list"}, None),
+        ({}, None),
+    ],
+)
+async def test_urn_extraction(
+    mock_session: MagicMock, claims: dict[str, Any], expected_urn: str | None
+) -> None:
+    auth = AsyncMock(spec=AbstractAuth)
+    auth.async_get_access_token.return_value = _jwt(claims)
+    api = API(session=mock_session, auth=auth)
+
+    await api.async_get_access_token()
+
+    assert api.urn == expected_urn
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "match"),
+    [
+        ({}, "An auth provider is required"),
+        (
+            {"auth": MagicMock(spec=AbstractAuth), "oauth_session": MagicMock()},
+            "Pass either auth or oauth_session, not both",
+        ),
+    ],
+)
+def test_invalid_auth_configuration(
+    mock_session: MagicMock, kwargs: dict[str, Any], match: str
+) -> None:
+    with pytest.raises(ValueError, match=match):
+        API(session=mock_session, **kwargs)
+
+
+async def test_deprecated_oauth_session_still_works(
+    mock_session: MagicMock,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    oauth_session = MagicMock(
+        valid_token=False,
+        token={"access_token": TOKEN},
+        async_ensure_token_valid=AsyncMock(),
+    )
+
+    with caplog.at_level(logging.WARNING):
+        api = API(session=mock_session, oauth_session=oauth_session)
+
+    assert "deprecated" in caplog.text
+    assert await api.async_get_access_token() == TOKEN
+    oauth_session.async_ensure_token_valid.assert_awaited_once()
+    assert api.urn == URN