From 229d126ea2713b8444f5d2659b023cc13b3bcafd Mon Sep 17 00:00:00 2001 From: Moustachauve <2206577+Moustachauve@users.noreply.github.com> Date: Sun, 4 Oct 2026 23:06:11 -0400 Subject: [PATCH 1/2] Replace Home Assistant OAuth2Session with an AbstractAuth interface pyhilo.api imported homeassistant.helpers.config_entry_oauth2_flow at module level, but homeassistant is not (and should not be) a dependency. Importing pyhilo outside Home Assistant failed with ModuleNotFoundError, and the test suite only passed because conftest.py injected fake homeassistant modules into sys.modules. API now takes an auth argument implementing pyhilo.auth.AbstractAuth, following Home Assistant's recommended pattern for API libraries: the library defines the interface, and the integration wraps its own OAuth2Session. The URN is decoded from the last token returned by the auth provider instead of reading the session directly. The oauth_session argument is still accepted, adapted internally and logged as deprecated, because already-released integration versions declare python-hilo>= and would otherwise break on a fresh install. It should be removed after a couple of releases. The raw access token is no longer logged at DEBUG level, since it ended up in Home Assistant logs. --- pyhilo/__init__.py | 2 + pyhilo/api.py | 42 ++++++++----- pyhilo/auth.py | 42 +++++++++++++ tests/conftest.py | 18 ------ tests/test_api_auth.py | 139 +++++++++++++++++++++++++++++++++++++++++ 5 files changed, 210 insertions(+), 33 deletions(-) create mode 100644 pyhilo/auth.py delete mode 100644 tests/conftest.py create mode 100644 tests/test_api_auth.py 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..bfe2c4e --- /dev/null +++ b/tests/test_api_auth.py @@ -0,0 +1,139 @@ +import base64 +import json +import logging +from typing import Any + +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]}) + + +class FakeAuth(AbstractAuth): + def __init__(self, tokens: list[str]) -> None: + self._tokens = tokens + self.calls = 0 + + async def async_get_access_token(self) -> str: + token = self._tokens[min(self.calls, len(self._tokens) - 1)] + self.calls += 1 + return token + + +class FakeResponse: + headers = {"content-type": "application/json"} + + async def json(self, content_type: Any = None) -> dict[str, Any]: + return {"ok": True} + + def raise_for_status(self) -> None: + pass + + async def __aenter__(self) -> "FakeResponse": + return self + + async def __aexit__(self, *args: object) -> None: + pass + + +class FakeSession: + def __init__(self) -> None: + self.requests: list[dict[str, Any]] = [] + + def request(self, method: str, url: str, **kwargs: Any) -> FakeResponse: + self.requests.append({"method": method, "url": url, **kwargs}) + return FakeResponse() + + +class FakeOAuth2Session: + """Duck-types Home Assistant's OAuth2Session.""" + + def __init__(self, token: str, valid: bool = True) -> None: + self.token = {"access_token": token} + self.valid_token = valid + self.ensure_calls = 0 + + async def async_ensure_token_valid(self) -> None: + self.ensure_calls += 1 + self.valid_token = True + + +def test_import_does_not_require_homeassistant() -> None: + import sys + + assert "homeassistant" not in sys.modules + + +async def test_async_create_fetches_token_through_auth( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def _noop(self: API) -> None: + pass + + monkeypatch.setattr(API, "_async_post_init", _noop) + auth = FakeAuth([TOKEN]) + + api = await API.async_create(session=FakeSession(), auth=auth) # type: ignore[arg-type] + + assert auth.calls == 1 + assert api.urn == URN + + +async def test_each_request_uses_a_fresh_token_from_auth() -> None: + session = FakeSession() + auth = FakeAuth([TOKEN, "refreshed-token"]) + api = API(session=session, auth=auth) # type: ignore[arg-type] + + await api._async_request("get", "/first") + await api._async_request("get", "/second") + + assert auth.calls == 2 + assert session.requests[0]["headers"]["authorization"] == f"Bearer {TOKEN}" + assert session.requests[1]["headers"]["authorization"] == "Bearer refreshed-token" + assert session.requests[1]["headers"]["Host"] == API_HOSTNAME + + +async def test_urn_is_none_for_token_without_claim() -> None: + api = API(session=FakeSession(), auth=FakeAuth([_jwt({})])) # type: ignore[arg-type] + + await api.async_get_access_token() + + assert api.urn is None + + +def test_auth_is_required() -> None: + with pytest.raises(ValueError): + API(session=FakeSession()) # type: ignore[arg-type] + + +def test_auth_and_oauth_session_are_mutually_exclusive() -> None: + with pytest.raises(ValueError): + API( + session=FakeSession(), # type: ignore[arg-type] + auth=FakeAuth([TOKEN]), + oauth_session=FakeOAuth2Session(TOKEN), + ) + + +async def test_deprecated_oauth_session_still_works( + caplog: pytest.LogCaptureFixture, +) -> None: + oauth_session = FakeOAuth2Session(TOKEN, valid=False) + + with caplog.at_level(logging.WARNING): + api = API(session=FakeSession(), oauth_session=oauth_session) # type: ignore[arg-type] + + assert "deprecated" in caplog.text + assert await api.async_get_access_token() == TOKEN + assert oauth_session.ensure_calls == 1 + assert api.urn == URN From ca014db3fdab9122028c93a8ef8421f50286dfbd Mon Sep 17 00:00:00 2001 From: Moustachauve <2206577+Moustachauve@users.noreply.github.com> Date: Sun, 4 Oct 2026 23:39:52 -0400 Subject: [PATCH 2/2] Improve test_api_auth with unittest.mock and parametrization Custom test stub classes have been replaced with standard unittest.mock fixtures and AsyncMock/MagicMock assertions to eliminate boilerplate. Edge cases for URN claim extraction and invalid auth initialization parameters are now parametrized with matching error assertions. --- tests/test_api_auth.py | 152 ++++++++++++++++++++--------------------- 1 file changed, 74 insertions(+), 78 deletions(-) diff --git a/tests/test_api_auth.py b/tests/test_api_auth.py index bfe2c4e..4d0c029 100644 --- a/tests/test_api_auth.py +++ b/tests/test_api_auth.py @@ -1,7 +1,9 @@ import base64 import json import logging +import sys from typing import Any +from unittest.mock import AsyncMock, MagicMock import pytest @@ -19,121 +21,115 @@ def _jwt(claims: dict[str, Any]) -> str: TOKEN = _jwt({"urn:com:hiloenergie:profile:location_hilo_id": [URN]}) -class FakeAuth(AbstractAuth): - def __init__(self, tokens: list[str]) -> None: - self._tokens = tokens - self.calls = 0 +@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() - async def async_get_access_token(self) -> str: - token = self._tokens[min(self.calls, len(self._tokens) - 1)] - self.calls += 1 - return token - - -class FakeResponse: - headers = {"content-type": "application/json"} - - async def json(self, content_type: Any = None) -> dict[str, Any]: - return {"ok": True} - - def raise_for_status(self) -> None: - pass - - async def __aenter__(self) -> "FakeResponse": - return self - - async def __aexit__(self, *args: object) -> None: - pass - - -class FakeSession: - def __init__(self) -> None: - self.requests: list[dict[str, Any]] = [] - - def request(self, method: str, url: str, **kwargs: Any) -> FakeResponse: - self.requests.append({"method": method, "url": url, **kwargs}) - return FakeResponse() - - -class FakeOAuth2Session: - """Duck-types Home Assistant's OAuth2Session.""" - - def __init__(self, token: str, valid: bool = True) -> None: - self.token = {"access_token": token} - self.valid_token = valid - self.ensure_calls = 0 - - async def async_ensure_token_valid(self) -> None: - self.ensure_calls += 1 - self.valid_token = True + # 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: - import sys - 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 = FakeAuth([TOKEN]) + auth = AsyncMock(spec=AbstractAuth) + auth.async_get_access_token.return_value = TOKEN - api = await API.async_create(session=FakeSession(), auth=auth) # type: ignore[arg-type] + api = await API.async_create(session=mock_session, auth=auth) - assert auth.calls == 1 + auth.async_get_access_token.assert_awaited_once() assert api.urn == URN -async def test_each_request_uses_a_fresh_token_from_auth() -> None: - session = FakeSession() - auth = FakeAuth([TOKEN, "refreshed-token"]) - api = API(session=session, auth=auth) # type: ignore[arg-type] +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.calls == 2 - assert session.requests[0]["headers"]["authorization"] == f"Bearer {TOKEN}" - assert session.requests[1]["headers"]["authorization"] == "Bearer refreshed-token" - assert session.requests[1]["headers"]["Host"] == API_HOSTNAME + 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"] -async def test_urn_is_none_for_token_without_claim() -> None: - api = API(session=FakeSession(), auth=FakeAuth([_jwt({})])) # type: ignore[arg-type] + assert first_call_headers["authorization"] == f"Bearer {TOKEN}" + assert second_call_headers["authorization"] == "Bearer refreshed-token" + assert second_call_headers["Host"] == API_HOSTNAME - await api.async_get_access_token() - assert api.urn is None - - -def test_auth_is_required() -> None: - with pytest.raises(ValueError): - API(session=FakeSession()) # type: ignore[arg-type] +@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() -def test_auth_and_oauth_session_are_mutually_exclusive() -> None: - with pytest.raises(ValueError): - API( - session=FakeSession(), # type: ignore[arg-type] - auth=FakeAuth([TOKEN]), - oauth_session=FakeOAuth2Session(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 = FakeOAuth2Session(TOKEN, valid=False) + oauth_session = MagicMock( + valid_token=False, + token={"access_token": TOKEN}, + async_ensure_token_valid=AsyncMock(), + ) with caplog.at_level(logging.WARNING): - api = API(session=FakeSession(), oauth_session=oauth_session) # type: ignore[arg-type] + api = API(session=mock_session, oauth_session=oauth_session) assert "deprecated" in caplog.text assert await api.async_get_access_token() == TOKEN - assert oauth_session.ensure_calls == 1 + oauth_session.async_ensure_token_valid.assert_awaited_once() assert api.urn == URN