diff --git a/docs/getting-started.md b/docs/getting-started.md index 45fa3aff..c81d2aff 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -74,6 +74,77 @@ Use a cloud server when you want to connect through the vendor’s public API. U asyncio.run(main()) ``` +=== "Somfy (multi-account cloud)" + + Use `Server.SOMFY` with `UsernamePasswordCredentials` when a single Somfy + account owns or is invited to **multiple sites (homes)** — the "multi + account sign-in" feature of the TaHoma app. Unlike the region-specific + `Server.SOMFY_EUROPE`/`SOMFY_AMERICA`/`SOMFY_OCEANIA` servers, `Server.SOMFY` + is region-agnostic: it discovers every site on the account and resolves the + correct regional endpoint for the one you select. + + ```python + import asyncio + + from pyoverkiz.auth.credentials import UsernamePasswordCredentials + from pyoverkiz.client import OverkizClient + from pyoverkiz.enums import Server + + + async def main() -> None: + async with OverkizClient( + server=Server.SOMFY, + credentials=UsernamePasswordCredentials("you@example.com", "password"), + ) as client: + await client.login() # auto-selects a sole site + + # Accounts with more than one site must select one explicitly. + gateways = await client.discover_gateways() + if len(gateways) > 1: + client.select_gateway(gateways[0].gateway_id) + + # Client is now scoped to the selected site and ready to use. + setup = await client.get_setup() + print(f"{len(setup.devices)} device(s)") + + asyncio.run(main()) + ``` + + Each `GatewayCandidate` from `discover_gateways()` carries a human-readable + `label` (the site name) and `home_id`, so a multi-site UI can let the user + pick before calling `select_gateway`. + + **Resume without a password.** After selecting a site, call + `client.to_credentials()` to snapshot the session as `SomfyTokenCredentials` + (a refresh token scoped to the selected site). Persist it and pass it back on + the next run to log in without the password grant, token exchange, or + discovery. The refresh token rotates, so supply an `on_token_refresh` + callback to re-persist it. + + ```python + import asyncio + + from pyoverkiz.auth.credentials import SomfyTokenCredentials + from pyoverkiz.client import OverkizClient + from pyoverkiz.enums import Server + + + async def persist(refresh_token: str) -> None: + # Store the rotated refresh token for next time. + ... + + + async def main(stored: SomfyTokenCredentials) -> None: + async with OverkizClient(server=Server.SOMFY, credentials=stored) as client: + await client.login() # no network round trips + setup = await client.get_setup() + print(f"{len(setup.devices)} device(s)") + + + # `stored` is what you persisted earlier via: + # stored = client.to_credentials(on_token_refresh=persist) + ``` + === "Somfy (local)" Local authentication requires a token generated via the official mobile app. For details on obtaining a token, refer to [Somfy TaHoma Developer Mode](https://github.com/Somfy-Developer/Somfy-TaHoma-Developer-Mode). diff --git a/pyoverkiz/auth/base.py b/pyoverkiz/auth/base.py index 3598bc5e..f736a5f7 100644 --- a/pyoverkiz/auth/base.py +++ b/pyoverkiz/auth/base.py @@ -3,9 +3,12 @@ from __future__ import annotations import datetime -from collections.abc import Mapping +from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass, field -from typing import Any, Protocol, runtime_checkable +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable + +if TYPE_CHECKING: + from pyoverkiz.auth.credentials import SomfyTokenCredentials @dataclass(slots=True) @@ -66,6 +69,7 @@ class GatewayCandidate: home_id: str | None = None label: str | None = None external_id: str | None = None + country: str | None = None @runtime_checkable @@ -81,3 +85,14 @@ def select_gateway(self, gateway_id: str) -> None: @property def selected_gateway(self) -> str | None: """Return the currently selected gateway id, or None.""" + + +@runtime_checkable +class SupportsSessionResume(Protocol): + """Optional capability: snapshot the session for later resume without re-login.""" + + def to_credentials( + self, + on_token_refresh: Callable[[str], Awaitable[None]] | None = None, + ) -> SomfyTokenCredentials: + """Return resume credentials for the current session.""" diff --git a/pyoverkiz/auth/bob.py b/pyoverkiz/auth/bob.py new file mode 100644 index 00000000..0d403650 --- /dev/null +++ b/pyoverkiz/auth/bob.py @@ -0,0 +1,90 @@ +"""Models for the Somfy BOB back-office site directory. + +A separate service from the Overkiz enduser API, with its own payload shapes +and casing, so it gets its own small cattrs converter rather than sharing +``pyoverkiz.converter``. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +import cattrs +from cattrs.gen import make_dict_structure_fn, override + +from pyoverkiz.auth.base import GatewayCandidate + + +@dataclass(slots=True) +class BobGateway: + """A gateway entry under a sub-site.""" + + gateway_id: str + + +@dataclass(slots=True) +class BobSubSite: + """A sub-site (setup) grouping one or more gateways.""" + + external_id: str | None = None + gateways: list[BobGateway] = field(default_factory=list) + + +@dataclass(slots=True) +class BobSite: + """A site (home) the account owns or was invited to.""" + + site_oid: str + name: str | None = None + country: str | None = None + sub_sites: list[BobSubSite] = field(default_factory=list) + + +@dataclass(slots=True) +class BobSitesResponse: + """The ``/sites`` listing, flattened on demand to gateway candidates.""" + + results: list[BobSite] = field(default_factory=list) + + def gateway_candidates(self) -> list[GatewayCandidate]: + """Flatten the site -> sub-site -> gateway tree into candidates.""" + return [ + GatewayCandidate( + gateway_id=gateway.gateway_id, + home_id=site.site_oid, + label=site.name, + external_id=sub.external_id, + country=site.country, + ) + for site in self.results + for sub in site.sub_sites + for gateway in sub.gateways + ] + + +def _make_bob_converter() -> cattrs.Converter: + # Converter (not GenConverter) so unknown BOB keys are dropped for forward-compat. + c = cattrs.Converter() + c.register_structure_hook( + BobGateway, + make_dict_structure_fn(BobGateway, c, gateway_id=override(rename="gatewayId")), + ) + c.register_structure_hook( + BobSubSite, + make_dict_structure_fn( + BobSubSite, c, external_id=override(rename="externalOID") + ), + ) + c.register_structure_hook( + BobSite, + make_dict_structure_fn( + BobSite, + c, + site_oid=override(rename="siteOID"), + sub_sites=override(rename="subSites"), + ), + ) + return c + + +bob_converter = _make_bob_converter() diff --git a/pyoverkiz/auth/credentials.py b/pyoverkiz/auth/credentials.py index 93e8b3d4..f67d07ea 100644 --- a/pyoverkiz/auth/credentials.py +++ b/pyoverkiz/auth/credentials.py @@ -31,6 +31,22 @@ class LocalTokenCredentials(TokenCredentials): """Credentials using a local API token.""" +@dataclass(slots=True) +class SomfyTokenCredentials(Credentials): + """Resume credentials for a previously-selected Somfy site (skips login + discovery). + + Persist the ``refresh_token`` plus the site's ``site_oid`` and ``region``. + The refresh token rotates, so supply ``on_token_refresh`` to re-persist it; + otherwise a later reload fails. + """ + + refresh_token: str = field(repr=False) + site_oid: str + region: str + gateway_id: str | None = None + on_token_refresh: Callable[[str], Awaitable[None]] | None = None + + @dataclass(slots=True) class RexelOAuthCodeCredentials(Credentials): """Credentials using Rexel OAuth2 authorization code with PKCE.""" diff --git a/pyoverkiz/auth/factory.py b/pyoverkiz/auth/factory.py index 81912a07..84772685 100644 --- a/pyoverkiz/auth/factory.py +++ b/pyoverkiz/auth/factory.py @@ -11,6 +11,7 @@ LocalTokenCredentials, RexelOAuthCodeCredentials, RexelTokenCredentials, + SomfyTokenCredentials, TokenCredentials, UsernamePasswordCredentials, ) @@ -24,6 +25,7 @@ RexelAuthStrategy, RexelTokenAuthStrategy, SessionLoginStrategy, + SomfyAccountAuthStrategy, SomfyAuthStrategy, ) from pyoverkiz.enums import APIType, Server @@ -62,6 +64,18 @@ def build_auth_strategy( ssl_context, ) + if server == Server.SOMFY: + # Resume from a persisted site-scoped refresh token, or fresh login + # from username/password. + if not isinstance(credentials, SomfyTokenCredentials): + credentials = _ensure_credentials(credentials, UsernamePasswordCredentials) + return SomfyAccountAuthStrategy( + credentials, + session, + server_config, + ssl_context, + ) + if server in { Server.ATLANTIC_COZYTOUCH, Server.THERMOR_COZYTOUCH, diff --git a/pyoverkiz/auth/strategies.py b/pyoverkiz/auth/strategies.py index 972c20fc..e10d7f54 100644 --- a/pyoverkiz/auth/strategies.py +++ b/pyoverkiz/auth/strategies.py @@ -5,9 +5,11 @@ import asyncio import base64 import binascii +import datetime import json +import logging import ssl -from collections.abc import Mapping +from collections.abc import Awaitable, Callable, Mapping from http import HTTPStatus from typing import TYPE_CHECKING, Any, cast @@ -17,10 +19,12 @@ from aiohttp import ClientResponse, ClientSession, FormData from pyoverkiz.auth.base import AuthContext, AuthStrategy, GatewayCandidate +from pyoverkiz.auth.bob import BobSitesResponse, bob_converter from pyoverkiz.auth.credentials import ( LocalTokenCredentials, RexelOAuthCodeCredentials, RexelTokenCredentials, + SomfyTokenCredentials, TokenCredentials, UsernamePasswordCredentials, ) @@ -40,8 +44,17 @@ REXEL_OAUTH_TOKEN_URL, REXEL_REQUIRED_CONSENT, SOMFY_API, + SOMFY_BOB_API_KEY, + SOMFY_BOB_SITE_API, SOMFY_CLIENT_ID, SOMFY_CLIENT_SECRET, + SOMFY_COUNTRY_REGION, + SOMFY_DEFAULT_REGION, + SOMFY_GINAITE_SUBJECT_ISSUER, + SOMFY_GINAITE_SUBJECT_TOKEN_TYPE, + SOMFY_GINAITE_TOKEN_EXCHANGE_GRANT, + SOMFY_GINAITE_TOKEN_URL, + SOMFY_REGION_ENDPOINT, ) from pyoverkiz.exceptions import ( BadCredentialsError, @@ -59,6 +72,8 @@ from pyoverkiz.models import ServerConfig from pyoverkiz.response_handler import check_response +_LOGGER = logging.getLogger(__name__) + MIN_JWT_SEGMENTS = 2 @@ -73,6 +88,40 @@ async def _raise_for_server_error(response: ClientResponse) -> None: await check_response(response) +async def _somfy_password_token( + session: ClientSession, username: str, password: str +) -> dict[str, Any]: + """Perform the Somfy Accounts password grant and return the raw token dict. + + Shared by SomfyAuthStrategy (single-site) and SomfyAccountAuthStrategy + (which feeds the returned access_token into the Keycloak token exchange). + """ + form = FormData( + { + "grant_type": "password", + "client_id": SOMFY_CLIENT_ID, + "client_secret": SOMFY_CLIENT_SECRET, + "username": username, + "password": password, + } + ) + async with session.post( + f"{SOMFY_API}/oauth/oauth/v2/token/jwt", + data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) as response: + await _raise_for_server_error(response) + token = await response.json() + + if token.get("message") == "error.invalid.grant": + raise SomfyBadCredentialsError(token["message"]) + + if not token.get("access_token"): + raise SomfyServiceError("No Somfy access token provided.") + + return cast(dict[str, Any], token) + + class BaseAuthStrategy(AuthStrategy): """Base class for authentication strategies.""" @@ -198,6 +247,15 @@ async def auth_headers(self, path: str | None = None) -> Mapping[str, str]: async def _request_access_token( self, *, grant_type: str, extra_fields: Mapping[str, str] ) -> None: + if grant_type == "password": + token = await _somfy_password_token( + self.session, + self.credentials.username, + self.credentials.password, + ) + self.context.update_from_token(token) + return + form = FormData( { "grant_type": grant_type, @@ -225,6 +283,238 @@ async def _request_access_token( self.context.update_from_token(token) +class SomfyAccountAuthStrategy(BaseAuthStrategy): + """Somfy multi-site auth: password grant -> Keycloak token exchange -> BOB site directory. + + Selecting a site mints a site-scoped token whose Bearer drives the classic + Overkiz enduser API directly (no gateway header needed). + """ + + def __init__( + self, + credentials: UsernamePasswordCredentials | SomfyTokenCredentials, + session: ClientSession, + server: ServerConfig, + ssl_context: ssl.SSLContext | bool, + ) -> None: + """Accept ``UsernamePasswordCredentials`` (fresh login) or ``SomfyTokenCredentials`` (resumed session).""" + super().__init__(session, server, ssl_context) + self.credentials = credentials + self.context = AuthContext() + self._sites: list[GatewayCandidate] = [] + self._selected_site_oid: str | None = None + self._selected_gateway: str | None = None + self._selected_region: str | None = None + self._endpoint: str | None = None + # Refresh-token persistence for resumed sessions (no-op for fresh login). + self._on_token_refresh: Callable[[str], Awaitable[None]] | None = None + self._persisted_refresh_token: str | None = None + + async def login(self) -> None: + """Fresh login (password grant -> exchange -> discover) or resumed session.""" + if isinstance(self.credentials, SomfyTokenCredentials): + self._resume_session(self.credentials) + return + + token = await _somfy_password_token( + self.session, self.credentials.username, self.credentials.password + ) + await self._token_exchange(token["access_token"]) + await self.discover_gateways() + + # Re-select on relogin to re-scope the fresh unscoped token; drop the + # selection if the gateway is gone after rediscovery. + known = {s.gateway_id for s in self._sites} + if self._selected_gateway and self._selected_gateway in known: + self.select_gateway(self._selected_gateway) + elif self._selected_gateway and self._selected_gateway not in known: + self._selected_gateway = None + self._selected_site_oid = None + self._endpoint = None + elif len(self._sites) == 1: + self.select_gateway(self._sites[0].gateway_id) + + def _resume_session(self, credentials: SomfyTokenCredentials) -> None: + """Seed site scope from persisted tokens (no network); first request mints a scoped token.""" + self.context.refresh_token = credentials.refresh_token + self.context.expires_at = datetime.datetime.now(datetime.UTC) + self._selected_site_oid = credentials.site_oid + self._selected_gateway = credentials.gateway_id + self._selected_region = credentials.region + self._endpoint = SOMFY_REGION_ENDPOINT[credentials.region] + self._on_token_refresh = credentials.on_token_refresh + self._persisted_refresh_token = credentials.refresh_token + + async def _token_exchange(self, sso_access_token: str) -> None: + """Exchange a Somfy Accounts SSO token for a Ginaite token (public client).""" + form = FormData( + { + "grant_type": SOMFY_GINAITE_TOKEN_EXCHANGE_GRANT, + "client_id": SOMFY_CLIENT_ID, + "subject_token": sso_access_token, + "subject_issuer": SOMFY_GINAITE_SUBJECT_ISSUER, + "subject_token_type": SOMFY_GINAITE_SUBJECT_TOKEN_TYPE, + } + ) + async with self.session.post(SOMFY_GINAITE_TOKEN_URL, data=form) as response: + await _raise_for_server_error(response) + if response.status != HTTPStatus.OK: + raise SomfyServiceError( + f"Somfy token exchange failed: {response.status}" + ) + self.context.update_from_token(await response.json()) + + async def discover_gateways(self) -> list[GatewayCandidate]: + """List the account's sites from BOB, flattened to gateway candidates.""" + data = await self._bob_get("sites?withGateways=true&limit=20&offset=0") + response = bob_converter.structure(data, BobSitesResponse) + self._sites = response.gateway_candidates() + return self._sites + + def select_gateway(self, gateway_id: str) -> None: + """Scope subsequent requests to the given gateway's site and region.""" + site = next( + (s for s in self._sites if s.gateway_id == gateway_id), + None, + ) + if site is None: + raise SomfyServiceError(f"Unknown gateway id: {gateway_id}") + + region = self._region_for_country(site.country) + + self._selected_gateway = gateway_id + self._selected_site_oid = site.home_id + self._selected_region = region + self._endpoint = SOMFY_REGION_ENDPOINT[region] + # Force the next request to mint a site-scoped token via refresh. + self.context.expires_at = datetime.datetime.now(datetime.UTC) + + @staticmethod + def _region_for_country(country: str | None) -> str: + """Map an ISO country to a region, warning and defaulting to EMEA if unresolvable.""" + region = SOMFY_COUNTRY_REGION.get(country.upper()) if country else None + if region is None: + _LOGGER.warning( + "Unresolvable Somfy site country %r; falling back to %s region", + country, + SOMFY_DEFAULT_REGION, + ) + return SOMFY_DEFAULT_REGION + return region + + @property + def selected_gateway(self) -> str | None: + """Return the currently selected gateway id, or None.""" + return self._selected_gateway + + def to_credentials( + self, + on_token_refresh: Callable[[str], Awaitable[None]] | None = None, + ) -> SomfyTokenCredentials: + """Snapshot the session (refresh token + site scope) as resume credentials. + + Raises if no site is selected or no refresh token is available yet. + """ + if ( + self._selected_site_oid is None + or self._selected_region is None + or self.context.refresh_token is None + ): + raise SomfyServiceError( + "Cannot snapshot resume credentials before a site is " + "selected and a refresh token is available." + ) + return SomfyTokenCredentials( + refresh_token=self.context.refresh_token, + site_oid=self._selected_site_oid, + region=self._selected_region, + gateway_id=self._selected_gateway, + on_token_refresh=on_token_refresh, + ) + + @property + def endpoint(self) -> str: + """Return the resolved per-site endpoint, or the server placeholder.""" + return self._endpoint or self.server.endpoint + + async def refresh_if_needed(self) -> bool: + """Mint/refresh a site-scoped token when expired; raise if a selected site has no refresh token.""" + if not self.context.is_expired(): + return False + if not self.context.refresh_token: + if self._selected_site_oid: + raise SomfyServiceError( + "Cannot mint a site-scoped Somfy token without a refresh token." + ) + return False + await self._refresh() + return True + + async def _refresh(self) -> None: + """Refresh grant scoped to the selected site (?siteOID).""" + url = SOMFY_GINAITE_TOKEN_URL + if self._selected_site_oid: + url = f"{SOMFY_GINAITE_TOKEN_URL}?siteOID={self._selected_site_oid}" + previous_refresh_token = self.context.refresh_token + form = FormData( + { + "grant_type": "refresh_token", + "client_id": SOMFY_CLIENT_ID, + "refresh_token": cast(str, self.context.refresh_token), + } + ) + async with self.session.post(url, data=form) as response: + await _raise_for_server_error(response) + if response.status != HTTPStatus.OK: + # A revoked refresh token (e.g. after a password change) is terminal; + # surface it as bad credentials so callers trigger reauth instead of retrying. + body = await response.json() + if body.get("error") == "invalid_grant": + raise SomfyBadCredentialsError( + body.get("error_description", "invalid_grant") + ) + raise SomfyServiceError( + f"Somfy token refresh failed: {response.status}" + ) + self.context.update_from_token(await response.json()) + + # refresh_token is optional in a refresh response (RFC 6749); reuse the old one if absent. + if self.context.refresh_token is None: + self.context.refresh_token = previous_refresh_token + await self._notify_token_refresh() + + async def _notify_token_refresh(self) -> None: + """Let a resuming caller persist a rotated refresh token (no-op otherwise).""" + if ( + self._on_token_refresh is not None + and self.context.refresh_token is not None + and self.context.refresh_token != self._persisted_refresh_token + ): + self._persisted_refresh_token = self.context.refresh_token + await self._on_token_refresh(self.context.refresh_token) + + async def auth_headers(self, path: str | None = None) -> Mapping[str, str]: + """Return the Bearer header (site-scoped token), or {} before login.""" + if self.context.access_token: + return {"Authorization": f"Bearer {self.context.access_token}"} + return {} + + async def _bob_get(self, path: str) -> dict[str, Any]: + """GET a BOB site-directory resource with Bearer + X-Api-Key.""" + async with self.session.get( + f"{SOMFY_BOB_SITE_API}/{path}", + headers={ + "Authorization": f"Bearer {self.context.access_token}", + "X-Api-Key": SOMFY_BOB_API_KEY, + }, + ssl=self._ssl, + ) as response: + await _raise_for_server_error(response) + if response.status != HTTPStatus.OK: + raise SomfyServiceError(f"BOB request failed: {response.status}") + return cast(dict[str, Any], await response.json()) + + class CozytouchAuthStrategy(SessionLoginStrategy): """Authentication strategy using Cozytouch session-based login.""" diff --git a/pyoverkiz/client.py b/pyoverkiz/client.py index 93b5f973..d091c4f1 100644 --- a/pyoverkiz/client.py +++ b/pyoverkiz/client.py @@ -10,7 +10,7 @@ from http import HTTPStatus from pathlib import Path from types import TracebackType -from typing import Any, Self, cast +from typing import TYPE_CHECKING, Any, Self, cast import backoff from aiohttp import ( @@ -30,6 +30,7 @@ SupportsGatewaySelection, build_auth_strategy, ) +from pyoverkiz.auth.base import SupportsSessionResume from pyoverkiz.const import SUPPORTED_SERVERS, USER_AGENT from pyoverkiz.converter import converter from pyoverkiz.enums import APIType, ExecutionMode, Protocol, Server @@ -71,6 +72,11 @@ from pyoverkiz.response_handler import check_response from pyoverkiz.serializers import prepare_payload +if TYPE_CHECKING: + from collections.abc import Awaitable, Callable + + from pyoverkiz.auth.credentials import SomfyTokenCredentials + _LOGGER = logging.getLogger(__name__) DEFAULT_TIMEOUT = ClientTimeout(total=15, sock_connect=10) @@ -933,6 +939,24 @@ def select_gateway(self, gateway_id: str) -> None: ) self._auth.select_gateway(gateway_id) + def to_credentials( + self, + on_token_refresh: Callable[[str], Awaitable[None]] | None = None, + ) -> SomfyTokenCredentials: + """Snapshot the session as resume credentials, to log in later without a password. + + Call after login and gateway selection. Supply ``on_token_refresh`` to + persist the rotating refresh token. + + Raises: + UnsupportedOperationError: When the server does not support session resume. + """ + if not isinstance(self._auth, SupportsSessionResume): + raise UnsupportedOperationError( + f"{self.server_config.name} does not support session resume." + ) + return self._auth.to_credentials(on_token_refresh) + # ----------------------------------------------------------------------- # Local token management (cloud API) # ----------------------------------------------------------------------- diff --git a/pyoverkiz/const.py b/pyoverkiz/const.py index 3fd8ce27..b3b466ce 100644 --- a/pyoverkiz/const.py +++ b/pyoverkiz/const.py @@ -42,6 +42,111 @@ # OAuth client secrets are public by design (embedded in mobile apps) SOMFY_CLIENT_SECRET = "12k73w1n540g8o4cokg0cw84cog840k84cwggscwg884004kgk" # noqa: S105 +# Somfy multi-site (Keycloak "Ginaite" realm + BOB back-office directory). +# The token exchange reuses SOMFY_CLIENT_ID as a PUBLIC client (no secret). +SOMFY_GINAITE_TOKEN_URL = ( + "https://ginaite-prod.ovkube.net/realms/somfy-tahoma/protocol/openid-connect/token" # noqa: S105 +) +SOMFY_GINAITE_SUBJECT_ISSUER = "somfy-customer" +SOMFY_GINAITE_TOKEN_EXCHANGE_GRANT = "urn:ietf:params:oauth:grant-type:token-exchange" # noqa: S105 +SOMFY_GINAITE_SUBJECT_TOKEN_TYPE = "urn:ietf:params:oauth:token-type:access_token" # noqa: S105 + +SOMFY_BOB_SITE_API = "https://backoffice-service.ovkube.net/site-api/public/v1" +SOMFY_BOB_API_KEY = "184638B3FBE874ACD24C14FBD657B" + +# Site region derived offline from its ISO country, mirroring the TaHoma app's BusinessArea.fromCountry (EMEA fallback). +SOMFY_DEFAULT_REGION = "EMEA" +SOMFY_REGION_ENDPOINT: MappingProxyType[str, str] = MappingProxyType( + { + "EMEA": "https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/", + "APAC": "https://ha201-1.overkiz.com/enduser-mobile-web/enduserAPI/", + "SNABA": "https://ha401-1.overkiz.com/enduser-mobile-web/enduserAPI/", + } +) +SOMFY_COUNTRY_REGION: MappingProxyType[str, str] = MappingProxyType( + { + # Americas — ha401 (SNABA). + "CA": "SNABA", + "US": "SNABA", + "MX": "SNABA", + # Asia-Pacific — ha201 (APAC). + "AU": "APAC", + "HK": "APAC", + "IN": "APAC", + "ID": "APAC", + "JP": "APAC", + "MY": "APAC", + "NZ": "APAC", + "PH": "APAC", + "SG": "APAC", + "TW": "APAC", + "TH": "APAC", + "VN": "APAC", + "KR": "APAC", + "CN": "APAC", + # Europe, Middle East & Africa — ha101 (EMEA). + "AL": "EMEA", + "AD": "EMEA", + "AT": "EMEA", + "BY": "EMEA", + "BE": "EMEA", + "BG": "EMEA", + "HR": "EMEA", + "CY": "EMEA", + "CZ": "EMEA", + "DK": "EMEA", + "EG": "EMEA", + "EE": "EMEA", + "FO": "EMEA", + "FI": "EMEA", + "FR": "EMEA", + "GF": "EMEA", + "PF": "EMEA", + "DE": "EMEA", + "GR": "EMEA", + "GP": "EMEA", + "HU": "EMEA", + "IL": "EMEA", + "IT": "EMEA", + "JE": "EMEA", + "JO": "EMEA", + "KZ": "EMEA", + "KW": "EMEA", + "LV": "EMEA", + "LB": "EMEA", + "LT": "EMEA", + "LU": "EMEA", + "MQ": "EMEA", + "YT": "EMEA", + "MC": "EMEA", + "MA": "EMEA", + "NL": "EMEA", + "NO": "EMEA", + "NC": "EMEA", + "PS": "EMEA", + "PL": "EMEA", + "PT": "EMEA", + "QA": "EMEA", + "IE": "EMEA", + "RE": "EMEA", + "RO": "EMEA", + "RU": "EMEA", + "BL": "EMEA", + "SA": "EMEA", + "RS": "EMEA", + "SK": "EMEA", + "ZA": "EMEA", + "ES": "EMEA", + "SE": "EMEA", + "CH": "EMEA", + "TN": "EMEA", + "TR": "EMEA", + "UA": "EMEA", + "AE": "EMEA", + "GB": "EMEA", + } +) + # Brandt Smart Control middleware (cookie-session Rails API in front of Overkiz) BRANDT_MIDDLEWARE_API = "https://www.smartcontrol-app.com" BRANDT_PARTNER = "brandt-electromenager" @@ -49,6 +154,7 @@ LOCAL_API_PATH = "/enduser-mobile-web/1/enduserAPI/" SERVERS_WITH_LOCAL_API = [ + Server.SOMFY, Server.SOMFY_EUROPE, Server.SOMFY_OCEANIA, Server.SOMFY_AMERICA, @@ -134,6 +240,14 @@ manufacturer="Somfy", api_type=APIType.CLOUD, ), + Server.SOMFY: ServerConfig( + server=Server.SOMFY, + name="Somfy", + # Placeholder; SomfyAccountAuthStrategy sets the real endpoint per selected site. + endpoint="https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Somfy", + api_type=APIType.CLOUD, + ), Server.SOMFY_EUROPE: ServerConfig( # alias of https://tahomalink.com server=Server.SOMFY_EUROPE, name="Somfy (Europe)", diff --git a/pyoverkiz/enums/server.py b/pyoverkiz/enums/server.py index 9970bddb..ff08d9f9 100644 --- a/pyoverkiz/enums/server.py +++ b/pyoverkiz/enums/server.py @@ -26,6 +26,7 @@ class Server(StrEnum): REXEL = "rexel" SAUTER_COZYTOUCH = "sauter_cozytouch" SIMU_LIVEIN2 = "simu_livein2" + SOMFY = "somfy" SOMFY_DEVELOPER_MODE = "somfy_developer_mode" SOMFY_EUROPE = "somfy_europe" SOMFY_AMERICA = "somfy_america" diff --git a/tests/test_auth.py b/tests/test_auth.py index e89384a3..3a530c9d 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -1,7 +1,7 @@ """Tests for authentication module.""" -# ruff: noqa: S105, S106 -# S105/S106: Test credentials use dummy values. +# ruff: noqa: S105, S106, S107 +# S105/S106/S107: Test credentials use dummy values. from __future__ import annotations @@ -9,6 +9,7 @@ import datetime import importlib.util import json +import logging import sys from unittest.mock import AsyncMock, MagicMock, patch @@ -222,6 +223,38 @@ async def test_build_auth_strategy_somfy(self): assert isinstance(strategy, SomfyAuthStrategy) + @pytest.mark.asyncio + async def test_build_auth_strategy_somfy_multisite(self): + """Server.SOMFY + username/password builds SomfyAccountAuthStrategy.""" + from pyoverkiz.auth.strategies import SomfyAccountAuthStrategy + from pyoverkiz.const import SUPPORTED_SERVERS + + strategy = build_auth_strategy( + server_config=SUPPORTED_SERVERS[Server.SOMFY], + credentials=UsernamePasswordCredentials("user", "pass"), + session=AsyncMock(spec=ClientSession), + ssl_context=True, + ) + + assert isinstance(strategy, SomfyAccountAuthStrategy) + + def test_build_auth_strategy_somfy_token_credentials(self): + """Server.SOMFY + SomfyTokenCredentials builds the resume strategy.""" + from pyoverkiz.auth.credentials import SomfyTokenCredentials + from pyoverkiz.auth.strategies import SomfyAccountAuthStrategy + from pyoverkiz.const import SUPPORTED_SERVERS + + strategy = build_auth_strategy( + server_config=SUPPORTED_SERVERS[Server.SOMFY], + credentials=SomfyTokenCredentials( + refresh_token="r", site_oid="s", region="EMEA" + ), + session=AsyncMock(spec=ClientSession), + ssl_context=True, + ) + + assert isinstance(strategy, SomfyAccountAuthStrategy) + @pytest.mark.asyncio async def test_build_auth_strategy_cozytouch(self): """Test building Cozytouch auth strategy.""" @@ -1404,3 +1437,610 @@ def test_rexel_token_strategy_supports_gateway_selection(): creds = RexelTokenCredentials(access_token="static-token") strategy, _ = _build_rexel_token_strategy([], credentials=creds) assert isinstance(strategy, SupportsGatewaySelection) + + +def test_somfy_multisite_constants_and_server(): + """Server.SOMFY and the Ginaite/BOB constants are defined and consistent.""" + from pyoverkiz.const import ( + SOMFY_BOB_API_KEY, + SOMFY_BOB_SITE_API, + SOMFY_COUNTRY_REGION, + SOMFY_GINAITE_SUBJECT_ISSUER, + SOMFY_GINAITE_SUBJECT_TOKEN_TYPE, + SOMFY_GINAITE_TOKEN_EXCHANGE_GRANT, + SOMFY_GINAITE_TOKEN_URL, + SOMFY_REGION_ENDPOINT, + SUPPORTED_SERVERS, + ) + from pyoverkiz.enums import Server + + assert Server.SOMFY == "somfy" + assert SOMFY_GINAITE_TOKEN_URL.endswith("/protocol/openid-connect/token") + assert SOMFY_GINAITE_SUBJECT_ISSUER == "somfy-customer" + assert SOMFY_GINAITE_TOKEN_EXCHANGE_GRANT == ( + "urn:ietf:params:oauth:grant-type:token-exchange" + ) + assert ( + SOMFY_GINAITE_SUBJECT_TOKEN_TYPE + == "urn:ietf:params:oauth:token-type:access_token" + ) + assert SOMFY_BOB_SITE_API.endswith("/site-api/public/v1") + assert SOMFY_BOB_API_KEY == "184638B3FBE874ACD24C14FBD657B" + + # All three regions are enumerated; every mapped region has an endpoint. + assert SOMFY_COUNTRY_REGION["NL"] == "EMEA" + assert SOMFY_COUNTRY_REGION["US"] == "SNABA" + assert SOMFY_COUNTRY_REGION["JP"] == "APAC" + for region in SOMFY_COUNTRY_REGION.values(): + assert region in SOMFY_REGION_ENDPOINT + assert SOMFY_REGION_ENDPOINT["EMEA"] == ( + "https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/" + ) + + config = SUPPORTED_SERVERS[Server.SOMFY] + assert config.server == Server.SOMFY + assert config.name == "Somfy" + + +@pytest.mark.asyncio +async def test_somfy_password_token_returns_token_dict(): + """_somfy_password_token posts the password grant and returns the token dict.""" + from unittest.mock import AsyncMock, MagicMock + + from aiohttp import ClientSession + + from pyoverkiz.auth.strategies import _somfy_password_token + + resp = MagicMock() + resp.status = 200 + resp.json = AsyncMock( + return_value={"access_token": "sso-abc", "refresh_token": "r1"} + ) + resp.__aenter__ = AsyncMock(return_value=resp) + resp.__aexit__ = AsyncMock(return_value=None) + session = MagicMock(spec=ClientSession) + session.post = MagicMock(return_value=resp) + + token = await _somfy_password_token(session, "user", "pass") + + assert token["access_token"] == "sso-abc" + + +@pytest.mark.asyncio +async def test_somfy_password_token_bad_credentials(): + """error.invalid.grant maps to SomfyBadCredentialsError.""" + from unittest.mock import AsyncMock, MagicMock + + from aiohttp import ClientSession + + from pyoverkiz.auth.strategies import _somfy_password_token + from pyoverkiz.exceptions import SomfyBadCredentialsError + + resp = MagicMock() + resp.status = 200 + resp.json = AsyncMock(return_value={"message": "error.invalid.grant"}) + resp.__aenter__ = AsyncMock(return_value=resp) + resp.__aexit__ = AsyncMock(return_value=None) + session = MagicMock(spec=ClientSession) + session.post = MagicMock(return_value=resp) + + with pytest.raises(SomfyBadCredentialsError): + await _somfy_password_token(session, "user", "bad") + + +def _build_somfy_multisite_strategy(): + """Return a SomfyAccountAuthStrategy with a MagicMock session.""" + from unittest.mock import MagicMock + + from aiohttp import ClientSession + + from pyoverkiz.auth.credentials import UsernamePasswordCredentials + from pyoverkiz.auth.strategies import SomfyAccountAuthStrategy + from pyoverkiz.const import SUPPORTED_SERVERS + from pyoverkiz.enums import Server + + session = MagicMock(spec=ClientSession) + strategy = SomfyAccountAuthStrategy( + credentials=UsernamePasswordCredentials("user", "pass"), + session=session, + server=SUPPORTED_SERVERS[Server.SOMFY], + ssl_context=True, + ) + return strategy, session + + +def _json_ctx(body, status=200): + """A MagicMock aiohttp response context manager returning `body` as JSON.""" + from unittest.mock import AsyncMock, MagicMock + + resp = MagicMock() + resp.status = status + resp.json = AsyncMock(return_value=body) + resp.text = AsyncMock(return_value=str(body)) + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=resp) + ctx.__aexit__ = AsyncMock(return_value=None) + return ctx + + +@pytest.mark.asyncio +async def test_somfy_multisite_token_exchange_populates_context(): + """_token_exchange stores the Ginaite access + refresh token.""" + strategy, session = _build_somfy_multisite_strategy() + session.post = MagicMock( + return_value=_json_ctx( + {"access_token": "ginaite-1", "refresh_token": "r-1", "expires_in": 900} + ) + ) + + await strategy._token_exchange("sso-access") + + assert strategy.context.access_token == "ginaite-1" + assert strategy.context.refresh_token == "r-1" + + +@pytest.mark.asyncio +async def test_somfy_multisite_token_exchange_error_raises(): + """A non-200 token exchange raises SomfyServiceError.""" + from pyoverkiz.exceptions import SomfyServiceError + + strategy, session = _build_somfy_multisite_strategy() + session.post = MagicMock(return_value=_json_ctx({"error": "bad"}, status=400)) + + with pytest.raises(SomfyServiceError): + await strategy._token_exchange("sso-access") + + +_BOB_SITES = { + "totalCount": 2, + "results": [ + { + "siteOID": "site-a", + "name": "My Home", + "country": "NL", + "currentUserRoles": [{"roleOID": "owner"}], + "subSites": [ + { + "externalOID": "ext-a", + "type": "SETUP", + "gateways": [{"gatewayId": "2025-0000-0001", "type": 98}], + } + ], + }, + { + "siteOID": "site-b", + "name": "Holiday Home", + "country": "NL", + "currentUserRoles": [{"roleOID": "owner"}], + "subSites": [ + { + "externalOID": "ext-b", + "type": "SETUP", + "gateways": [{"gatewayId": "1225-0000-0002", "type": 29}], + } + ], + }, + ], +} + + +@pytest.mark.asyncio +async def test_somfy_multisite_discover_flattens_sites(): + """discover_gateways returns one GatewayCandidate per gateway across sites.""" + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + session.get = MagicMock(return_value=_json_ctx(_BOB_SITES)) + + candidates = await strategy.discover_gateways() + + assert [c.gateway_id for c in candidates] == ["2025-0000-0001", "1225-0000-0002"] + assert candidates[0].home_id == "site-a" + assert candidates[0].label == "My Home" + assert candidates[0].external_id == "ext-a" + assert candidates[0].country == "NL" + + +@pytest.mark.asyncio +async def test_somfy_multisite_select_resolves_region_endpoint(): + """Selecting a gateway resolves its country to the EMEA endpoint.""" + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + session.get = MagicMock(return_value=_json_ctx(_BOB_SITES)) + await strategy.discover_gateways() + + strategy.select_gateway("2025-0000-0001") + + assert strategy.selected_gateway == "2025-0000-0001" + assert strategy._selected_site_oid == "site-a" + assert strategy.endpoint == ( + "https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/" + ) + + +@pytest.mark.asyncio +async def test_somfy_multisite_select_maps_non_default_region(): + """A country in the map resolves to its non-default region endpoint.""" + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + us_site = { + "totalCount": 1, + "results": [ + { + "siteOID": "site-us", + "name": "Denver", + "country": "US", + "currentUserRoles": [{"roleOID": "owner"}], + "subSites": [ + {"externalOID": "ext-us", "gateways": [{"gatewayId": "gw-us"}]} + ], + } + ], + } + session.get = MagicMock(return_value=_json_ctx(us_site)) + await strategy.discover_gateways() + + strategy.select_gateway("gw-us") + + assert strategy.endpoint == ( + "https://ha401-1.overkiz.com/enduser-mobile-web/enduserAPI/" + ) + + +@pytest.mark.asyncio +async def test_somfy_multisite_select_unknown_country_falls_back_to_emea(caplog): + """An unknown country resolves to EMEA and logs a warning (matches the app).""" + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + unknown = { + "totalCount": 1, + "results": [ + { + "siteOID": "site-x", + "name": "Mars", + "country": "ZZ", + "currentUserRoles": [{"roleOID": "owner"}], + "subSites": [ + {"externalOID": "ext-x", "gateways": [{"gatewayId": "gw-x"}]} + ], + } + ], + } + session.get = MagicMock(return_value=_json_ctx(unknown)) + await strategy.discover_gateways() + + with caplog.at_level(logging.WARNING): + strategy.select_gateway("gw-x") + + assert strategy.endpoint == ( + "https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/" + ) + assert "ZZ" in caplog.text + assert "EMEA" in caplog.text + + +@pytest.mark.asyncio +async def test_somfy_multisite_select_missing_country_falls_back_to_emea(caplog): + """A site without a country resolves to EMEA and logs a warning.""" + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + missing = { + "totalCount": 1, + "results": [ + { + "siteOID": "site-x", + "name": "Mystery", + "currentUserRoles": [{"roleOID": "owner"}], + "subSites": [ + {"externalOID": "ext-x", "gateways": [{"gatewayId": "gw-x"}]} + ], + } + ], + } + session.get = MagicMock(return_value=_json_ctx(missing)) + await strategy.discover_gateways() + + with caplog.at_level(logging.WARNING): + strategy.select_gateway("gw-x") + + assert strategy.endpoint == ( + "https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/" + ) + assert "EMEA" in caplog.text + + +@pytest.mark.asyncio +async def test_somfy_multisite_endpoint_defaults_to_placeholder_before_select(): + """Before selection, endpoint falls back to the server config placeholder.""" + strategy, _ = _build_somfy_multisite_strategy() + assert strategy.endpoint == ( + "https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/" + ) + + +@pytest.mark.asyncio +async def test_somfy_multisite_auth_headers(): + """auth_headers returns the Bearer token, or {} when absent (no gateway header).""" + strategy, _ = _build_somfy_multisite_strategy() + assert await strategy.auth_headers() == {} + strategy.context.access_token = "ginaite-1" + headers = await strategy.auth_headers() + assert headers == {"Authorization": "Bearer ginaite-1"} + assert "gatewayId" not in headers + + +@pytest.mark.asyncio +async def test_somfy_multisite_refresh_scopes_to_selected_site(): + """refresh_if_needed posts a refresh grant to the ?siteOID URL.""" + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + strategy.context.refresh_token = "r-1" + session.get = MagicMock(return_value=_json_ctx(_BOB_SITES)) + await strategy.discover_gateways() + strategy.select_gateway("2025-0000-0001") # forces expiry + + posted = _json_ctx({"access_token": "scoped-1", "refresh_token": "r-2"}) + session.post = MagicMock(return_value=posted) + + refreshed = await strategy.refresh_if_needed() + + assert refreshed is True + assert strategy.context.access_token == "scoped-1" + # The refresh URL must carry ?siteOID=. + called_url = session.post.call_args.args[0] + assert "siteOID=site-a" in called_url + + +@pytest.mark.asyncio +async def test_somfy_multisite_refresh_invalid_grant_raises_bad_credentials(): + """A revoked refresh token (400 invalid_grant) maps to SomfyBadCredentialsError so callers reauth.""" + from pyoverkiz.exceptions import SomfyBadCredentialsError + + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + strategy.context.refresh_token = "r-1" + session.get = MagicMock(return_value=_json_ctx(_BOB_SITES)) + await strategy.discover_gateways() + strategy.select_gateway("2025-0000-0001") # forces expiry + + session.post = MagicMock( + return_value=_json_ctx( + {"error": "invalid_grant", "error_description": "token revoked"}, + status=400, + ) + ) + + with pytest.raises(SomfyBadCredentialsError, match="token revoked"): + await strategy.refresh_if_needed() + + +@pytest.mark.asyncio +async def test_somfy_multisite_refresh_without_refresh_token_raises(): + """No refresh_token after site selection must raise, not silently no-op. + + Without a refresh token, refresh_if_needed() can't mint the site-scoped + token, so it must not return False and let auth_headers() keep serving + the unscoped global token against the site's region endpoint. + """ + from pyoverkiz.exceptions import SomfyServiceError + + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + strategy.context.refresh_token = None + session.get = MagicMock(return_value=_json_ctx(_BOB_SITES)) + await strategy.discover_gateways() + + strategy.select_gateway("2025-0000-0001") # forces expiry, no refresh_token + + with pytest.raises(SomfyServiceError): + await strategy.refresh_if_needed() + + +def _patch_somfy_login_tokens(strategy, *, ginaite_access_token="ginaite-fresh"): + """Patch the password grant + token exchange so login() can run offline. + + Returns a context manager patching the module-level password grant to a + fixed SSO token, and ``strategy._token_exchange`` to install a fresh, + unscoped, non-expired Ginaite token via the real ``update_from_token``. + """ + + def _install_fresh_token(_sso_access_token): + strategy.context.update_from_token( + { + "access_token": ginaite_access_token, + "refresh_token": "r-fresh", + "expires_in": 900, + } + ) + + return ( + patch( + "pyoverkiz.auth.strategies._somfy_password_token", + AsyncMock(return_value={"access_token": "sso-fresh"}), + ), + patch.object( + strategy, + "_token_exchange", + AsyncMock(side_effect=_install_fresh_token), + ), + ) + + +@pytest.mark.asyncio +async def test_somfy_multisite_relogin_rescopes_selected_gateway(): + """Relogin on a multi-site account must re-apply site scoping. + + A relogin mints a fresh, unscoped Ginaite token that is NOT expired + (expires_in=900), so without re-selecting the previously-selected + gateway, refresh_if_needed() would return False and auth_headers() would + serve the unscoped global token against the still-selected region + endpoint. The fix must re-select the gateway so the context is marked + expired again, forcing the next request to mint a site-scoped token. + """ + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + session.get = MagicMock(return_value=_json_ctx(_BOB_SITES)) + await strategy.discover_gateways() + strategy.select_gateway("2025-0000-0001") + emea_endpoint = strategy.endpoint + + patch_password, patch_exchange = _patch_somfy_login_tokens(strategy) + with patch_password, patch_exchange: + await strategy.login() + + assert strategy.selected_gateway == "2025-0000-0001" + assert strategy.endpoint == emea_endpoint + # The fresh token must be treated as stale so the next request re-scopes it. + assert strategy.context.is_expired() + + +@pytest.mark.asyncio +async def test_somfy_multisite_relogin_drops_removed_gateway(): + """If the previously-selected gateway disappears on relogin, drop it. + + Rather than silently keep pointing at a gateway/endpoint that's gone + (e.g. access revoked), the stale selection state must be cleared. + """ + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + session.get = MagicMock(return_value=_json_ctx(_BOB_SITES)) + await strategy.discover_gateways() + strategy.select_gateway("2025-0000-0001") + + reduced_sites = { + "totalCount": 1, + "results": [_BOB_SITES["results"][1]], # only site-b / 1225-0000-0002 + } + session.get = MagicMock(return_value=_json_ctx(reduced_sites)) + + patch_password, patch_exchange = _patch_somfy_login_tokens(strategy) + with patch_password, patch_exchange: + await strategy.login() + + assert strategy.selected_gateway is None + assert strategy._selected_site_oid is None + assert strategy.endpoint == strategy.server.endpoint + + +def _build_somfy_resume_strategy(**credential_overrides): + """Return a SomfyAccountAuthStrategy built from SomfyTokenCredentials.""" + from unittest.mock import MagicMock + + from aiohttp import ClientSession + + from pyoverkiz.auth.credentials import SomfyTokenCredentials + from pyoverkiz.auth.strategies import SomfyAccountAuthStrategy + from pyoverkiz.const import SUPPORTED_SERVERS + from pyoverkiz.enums import Server + + creds_kwargs = { + "refresh_token": "stored-r", + "site_oid": "site-b", + "region": "EMEA", + "gateway_id": "1225-0000-0002", + } + creds_kwargs.update(credential_overrides) + session = MagicMock(spec=ClientSession) + strategy = SomfyAccountAuthStrategy( + credentials=SomfyTokenCredentials(**creds_kwargs), + session=session, + server=SUPPORTED_SERVERS[Server.SOMFY], + ssl_context=True, + ) + return strategy, session + + +@pytest.mark.asyncio +async def test_somfy_resume_login_seeds_scope_without_http(): + """Resuming a session performs no network calls and seeds the site scope.""" + strategy, session = _build_somfy_resume_strategy() + + await strategy.login() + + # No password grant, no token exchange, no discovery. + session.post.assert_not_called() + session.get.assert_not_called() + assert strategy.selected_gateway == "1225-0000-0002" + assert strategy._selected_site_oid == "site-b" + assert strategy.endpoint == ( + "https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/" + ) + # Token is marked expired so the first request mints a site-scoped one. + assert strategy.context.is_expired() + assert strategy.context.refresh_token == "stored-r" + + +@pytest.mark.asyncio +async def test_somfy_resume_first_refresh_scopes_to_site(): + """The first refresh after resuming mints a token via the ?siteOID URL.""" + strategy, session = _build_somfy_resume_strategy() + await strategy.login() + + session.post = MagicMock( + return_value=_json_ctx({"access_token": "scoped-1", "refresh_token": "r-2"}) + ) + refreshed = await strategy.refresh_if_needed() + + assert refreshed is True + assert strategy.context.access_token == "scoped-1" + assert "siteOID=site-b" in session.post.call_args.args[0] + + +@pytest.mark.asyncio +async def test_somfy_resume_credentials_roundtrip(): + """to_credentials() snapshots a fresh session's reusable state.""" + strategy, session = _build_somfy_multisite_strategy() + strategy.context.access_token = "ginaite-1" + strategy.context.refresh_token = "r-1" + session.get = MagicMock(return_value=_json_ctx(_BOB_SITES)) + await strategy.discover_gateways() + strategy.select_gateway("1225-0000-0002") + + snapshot = strategy.to_credentials() + + assert snapshot.refresh_token == "r-1" + assert snapshot.site_oid == "site-b" + assert snapshot.region == "EMEA" + assert snapshot.gateway_id == "1225-0000-0002" + + +@pytest.mark.asyncio +async def test_somfy_resume_credentials_requires_selection(): + """Snapshotting before a site is selected raises rather than half-populating.""" + from pyoverkiz.exceptions import SomfyServiceError + + strategy, _ = _build_somfy_multisite_strategy() + + with pytest.raises(SomfyServiceError): + strategy.to_credentials() + + +@pytest.mark.asyncio +async def test_somfy_resume_rotated_refresh_token_notifies(): + """A rotated refresh token fires on_token_refresh so the caller can persist.""" + persisted = [] + + async def _persist(token): + persisted.append(token) + + strategy, session = _build_somfy_resume_strategy(on_token_refresh=_persist) + await strategy.login() + + session.post = MagicMock( + return_value=_json_ctx({"access_token": "scoped-1", "refresh_token": "r-rot"}) + ) + await strategy.refresh_if_needed() + + assert persisted == ["r-rot"] + + +@pytest.mark.asyncio +async def test_somfy_resume_missing_refresh_token_preserved(): + """A refresh response without a refresh_token keeps the working one.""" + strategy, session = _build_somfy_resume_strategy() + await strategy.login() + + # Ginaite may omit refresh_token on refresh; the old one must survive. + session.post = MagicMock(return_value=_json_ctx({"access_token": "scoped-1"})) + await strategy.refresh_if_needed() + + assert strategy.context.refresh_token == "stored-r" diff --git a/tests/test_client.py b/tests/test_client.py index c556c0b2..5e8ad4a3 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -16,6 +16,7 @@ SupportsGatewaySelection, UsernamePasswordCredentials, ) +from pyoverkiz.auth.base import SupportsSessionResume from pyoverkiz.client import OverkizClient, OverkizClientSettings from pyoverkiz.const import USER_AGENT from pyoverkiz.enums import ( @@ -1397,6 +1398,26 @@ def test_select_gateway_delegates_to_strategy(self, client: OverkizClient) -> No fake_auth.select_gateway.assert_called_once_with("g1") + def test_to_credentials_delegates_to_strategy(self, client: OverkizClient) -> None: + """to_credentials forwards to a resume-capable strategy.""" + fake_auth = MagicMock(spec=SupportsSessionResume) + client._auth = fake_auth + + client.to_credentials() + + fake_auth.to_credentials.assert_called_once() + + def test_to_credentials_raises_for_unsupported_strategy( + self, client: OverkizClient + ) -> None: + """to_credentials raises UnsupportedOperationError when unsupported.""" + # The SOMFY_EUROPE strategy does not implement SupportsSessionResume. + with pytest.raises( + exceptions.UnsupportedOperationError, + match="does not support session resume", + ): + client.to_credentials() + class TestUserAgent: """Tests for the User-Agent header sent by OverkizClient."""