diff --git a/pyhilo/__init__.py b/pyhilo/__init__.py index b998e17..8234f89 100644 --- a/pyhilo/__init__.py +++ b/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", diff --git a/pyhilo/api.py b/pyhilo/api.py index 942b769..50914e8 100755 --- a/pyhilo/api.py +++ b/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: diff --git a/pyhilo/auth.py b/pyhilo/auth.py new file mode 100644 index 0000000..e481f40 --- /dev/null +++ b/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"]) diff --git a/tests/conftest.py b/tests/conftest.py deleted file mode 100644 index d71a9ed..0000000 --- a/tests/conftest.py +++ /dev/null @@ -1,18 +0,0 @@ -import sys -import types - - -def pytest_configure(config): - ha = types.ModuleType("homeassistant") - ha_helpers = types.ModuleType("homeassistant.helpers") - oauth2 = types.ModuleType("homeassistant.helpers.config_entry_oauth2_flow") - oauth2.OAuth2Session = type("OAuth2Session", (), {}) - sys.modules["homeassistant"] = ha - sys.modules["homeassistant.helpers"] = ha_helpers - sys.modules["homeassistant.helpers.config_entry_oauth2_flow"] = oauth2 - - -def pytest_unconfigure(config): - sys.modules.pop("homeassistant", None) - sys.modules.pop("homeassistant.helpers", None) - sys.modules.pop("homeassistant.helpers.config_entry_oauth2_flow", None) diff --git a/tests/test_api_auth.py b/tests/test_api_auth.py new file mode 100644 index 0000000..4d0c029 --- /dev/null +++ b/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