From f1a958a36412b7288b6dd1d7c8e25f581071464c Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Sun, 5 Jul 2026 22:11:30 +0000 Subject: [PATCH 1/7] feat: add Somfy multi-account (multi-site) authentication Add Server.SOMFY with a region-agnostic multi-site auth strategy that lets a single Somfy account authenticate and control each of its sites through pyoverkiz, reusing every existing Overkiz endpoint call. - Password grant + Keycloak (Ginaite) token exchange, no browser/PKCE - Site discovery via the BOB directory; region resolved from a static country->region map mirroring the TaHoma app, with EMEA fallback - Per-site token minting and re-scoping on relogin - Warm-start credentials to skip rediscovery when the site is known --- pyoverkiz/auth/credentials.py | 22 ++ pyoverkiz/auth/factory.py | 14 + pyoverkiz/auth/strategies.py | 349 ++++++++++++++++++- pyoverkiz/const.py | 121 +++++++ pyoverkiz/enums/server.py | 1 + tests/test_auth.py | 620 +++++++++++++++++++++++++++++++++- 6 files changed, 1124 insertions(+), 3 deletions(-) diff --git a/pyoverkiz/auth/credentials.py b/pyoverkiz/auth/credentials.py index 93e8b3d4..7d2372c8 100644 --- a/pyoverkiz/auth/credentials.py +++ b/pyoverkiz/auth/credentials.py @@ -31,6 +31,28 @@ class LocalTokenCredentials(TokenCredentials): """Credentials using a local API token.""" +@dataclass(slots=True) +class SomfyTokenCredentials(Credentials): + """Warm-start credentials for a previously-selected Somfy site. + + Skips the password grant, Keycloak token exchange, and site discovery on + reload: the caller persists the Ginaite ``refresh_token`` plus the selected + site's ``site_oid`` and ``region``, and pyoverkiz mints a site-scoped access + token directly on the first request. ``gateway_id`` is optional bookkeeping + (the id the user selected) and is surfaced via ``selected_gateway``. + + Ginaite rotates the refresh token on refresh, so supply an async + ``on_token_refresh`` callback to re-persist the new refresh token; without + it a rotated token is only kept in memory and a later reload would fail. + """ + + 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..5a085b18 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: + # Warm start from a persisted site-scoped refresh token, or cold start + # 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..0e545b1c 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 @@ -21,6 +23,7 @@ LocalTokenCredentials, RexelOAuthCodeCredentials, RexelTokenCredentials, + SomfyTokenCredentials, TokenCredentials, UsernamePasswordCredentials, ) @@ -40,8 +43,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 +71,8 @@ from pyoverkiz.models import ServerConfig from pyoverkiz.response_handler import check_response +_LOGGER = logging.getLogger(__name__) + MIN_JWT_SEGMENTS = 2 @@ -73,6 +87,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 +246,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 +282,296 @@ async def _request_access_token( self.context.update_from_token(token) +class SomfyAccountAuthStrategy(BaseAuthStrategy): + """Somfy multi-site auth: Keycloak token exchange + BOB site directory. + + Reuses the Somfy Accounts password grant, exchanges the SSO token for a + Ginaite (Keycloak) token, then lists the account's sites from the BOB + 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: + """Create a Somfy multi-site strategy with a fresh auth context. + + Accepts either ``UsernamePasswordCredentials`` (cold start: password + grant + token exchange + discovery) or ``SomfyTokenCredentials`` (warm + start: a persisted refresh token scoped to an already-selected site, + skipping all three network round trips). + """ + super().__init__(session, server, ssl_context) + self.credentials = credentials + self.context = AuthContext() + self._sites: list[GatewayCandidate] = [] + self._site_country: dict[str, str] = {} + self._selected_site_oid: str | None = None + self._selected_gateway: str | None = None + self._selected_region: str | None = None + self._endpoint: str | None = None + # Warm-start refresh-token persistence (no-op for cold start). + self._on_token_refresh: Callable[[str], Awaitable[None]] | None = None + self._persisted_refresh_token: str | None = None + + async def login(self) -> None: + """Cold start (password) or warm start (persisted refresh token). + + With ``SomfyTokenCredentials`` this is a warm start: no password grant, + no token exchange, no discovery. We seed the context from the stored + refresh token and site scope, so the first request mints a site-scoped + access token via the existing refresh path. + + With ``UsernamePasswordCredentials`` this is the cold start: password + grant -> token exchange -> discover, then (re-)select a site. Relogin + (e.g. after ``NotAuthenticatedError``) mints a fresh, unscoped Ginaite + token that is not itself expired, so on a multi-site account the + previously-selected gateway must be re-selected to re-apply site + scoping; otherwise the unscoped global token would keep being served + against the still-selected region endpoint. If that gateway is no + longer present after rediscovery, drop the stale selection instead of + silently pointing at a gateway that's gone. + """ + if isinstance(self.credentials, SomfyTokenCredentials): + self._warm_start(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() + + 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 _warm_start(self, credentials: SomfyTokenCredentials) -> None: + """Seed site scope from persisted tokens; skip password/exchange/discover. + + Sets up exactly the state ``select_gateway`` would have produced, then + marks the (absent) access token expired so the first request mints a + site-scoped token via the existing ``refresh_if_needed`` -> ``_refresh`` + path. No network calls happen here. + """ + 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") + candidates: list[GatewayCandidate] = [] + self._site_country = {} + for site in data.get("results", []): + site_oid = str(site["siteOID"]) + label = site.get("name") + country = site.get("country") + for sub in site.get("subSites", []): + external_id = sub.get("externalOID") + for gateway in sub.get("gateways", []): + gateway_id = str(gateway["gatewayId"]) + if country is not None: + self._site_country[gateway_id] = str(country) + candidates.append( + GatewayCandidate( + gateway_id=gateway_id, + home_id=site_oid, + label=label, + external_id=( + str(external_id) if external_id is not None else None + ), + ) + ) + self._sites = candidates + return candidates + + 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(self._site_country.get(gateway_id)) + + 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 an Overkiz region, defaulting to EMEA. + + Mirrors the TaHoma app's BusinessArea.fromCountry: known countries map + to their region, and anything unresolvable falls back to EMEA. A country + we cannot resolve (missing, or present but unmapped) is logged, since it + likely means the map needs updating for a newly supported region. + """ + 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 warm_start_credentials( + self, + on_token_refresh: Callable[[str], Awaitable[None]] | None = None, + ) -> SomfyTokenCredentials: + """Snapshot the current session as reusable warm-start credentials. + + Call this after login + gateway selection to persist the state a reload + needs (refresh token + site scope + region), skipping password grant, + token exchange, and discovery next time. 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 warm-start 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. + + Raises if a site has been selected but there is no refresh token to + mint the site-scoped token with, rather than silently continuing to + serve the unscoped global token against the site's region endpoint. + """ + 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: + raise SomfyServiceError( + f"Somfy token refresh failed: {response.status}" + ) + self.context.update_from_token(await response.json()) + + # Ginaite may omit refresh_token on a refresh; keep the working one + # rather than dropping to None (which would break the next warm start). + 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 warm-start 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/const.py b/pyoverkiz/const.py index 3fd8ce27..c61fb9d3 100644 --- a/pyoverkiz/const.py +++ b/pyoverkiz/const.py @@ -42,6 +42,117 @@ # 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" + +# The BOB directory carries no region field, so a site's Overkiz region is +# derived from its ISO 3166-1 alpha-2 country. This mirrors the TaHoma app's +# BusinessArea.fromCountry (com.somfy.homeapp v2.5.1): the region is a static, +# offline lookup — never runtime probing. All three regions' countries are +# enumerated so a genuinely unlisted country can be detected (and logged); it +# still falls back to EMEA, matching the app's fromCountry default. +# Verified live: NL -> EMEA. +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" @@ -134,6 +245,16 @@ manufacturer="Somfy", api_type=APIType.CLOUD, ), + Server.SOMFY: ServerConfig( + server=Server.SOMFY, + # Region-agnostic multi-site login. The endpoint here is a + # placeholder; SomfyAccountAuthStrategy overrides it per selected + # site once the region is resolved. + name="Somfy", + 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..736a4029 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 warm-start 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,586 @@ 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": "Mick", + "country": "NL", + "currentUserRoles": [{"roleOID": "owner"}], + "subSites": [ + { + "externalOID": "ext-a", + "type": "SETUP", + "gateways": [{"gatewayId": "2025-0000-0001", "type": 98}], + } + ], + }, + { + "siteOID": "site-b", + "name": "Smientstraat", + "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 == "Mick" + assert candidates[0].external_id == "ext-a" + + +@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_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_warmstart_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_warmstart_login_seeds_scope_without_http(): + """Warm-start login performs no network calls and seeds the site scope.""" + strategy, session = _build_somfy_warmstart_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_warmstart_first_refresh_scopes_to_site(): + """The first refresh after warm start mints a token via the ?siteOID URL.""" + strategy, session = _build_somfy_warmstart_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_warmstart_credentials_roundtrip(): + """warm_start_credentials() snapshots a cold 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.warm_start_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_warmstart_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.warm_start_credentials() + + +@pytest.mark.asyncio +async def test_somfy_warmstart_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_warmstart_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_warmstart_missing_refresh_token_preserved(): + """A refresh response without a refresh_token keeps the working one.""" + strategy, session = _build_somfy_warmstart_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" From 519cc3f609367e21485dcfcc9cf2d730e7c6812a Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Sun, 5 Jul 2026 22:22:19 +0000 Subject: [PATCH 2/7] chore: redact personal site names in tests; condense region-map comment --- pyoverkiz/const.py | 8 +------- tests/test_auth.py | 6 +++--- 2 files changed, 4 insertions(+), 10 deletions(-) diff --git a/pyoverkiz/const.py b/pyoverkiz/const.py index c61fb9d3..55bde647 100644 --- a/pyoverkiz/const.py +++ b/pyoverkiz/const.py @@ -54,13 +54,7 @@ SOMFY_BOB_SITE_API = "https://backoffice-service.ovkube.net/site-api/public/v1" SOMFY_BOB_API_KEY = "184638B3FBE874ACD24C14FBD657B" -# The BOB directory carries no region field, so a site's Overkiz region is -# derived from its ISO 3166-1 alpha-2 country. This mirrors the TaHoma app's -# BusinessArea.fromCountry (com.somfy.homeapp v2.5.1): the region is a static, -# offline lookup — never runtime probing. All three regions' countries are -# enumerated so a genuinely unlisted country can be detected (and logged); it -# still falls back to EMEA, matching the app's fromCountry default. -# Verified live: NL -> EMEA. +# 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( { diff --git a/tests/test_auth.py b/tests/test_auth.py index 736a4029..51817e6e 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -1596,7 +1596,7 @@ async def test_somfy_multisite_token_exchange_error_raises(): "results": [ { "siteOID": "site-a", - "name": "Mick", + "name": "My Home", "country": "NL", "currentUserRoles": [{"roleOID": "owner"}], "subSites": [ @@ -1609,7 +1609,7 @@ async def test_somfy_multisite_token_exchange_error_raises(): }, { "siteOID": "site-b", - "name": "Smientstraat", + "name": "Holiday Home", "country": "NL", "currentUserRoles": [{"roleOID": "owner"}], "subSites": [ @@ -1635,7 +1635,7 @@ async def test_somfy_multisite_discover_flattens_sites(): 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 == "Mick" + assert candidates[0].label == "My Home" assert candidates[0].external_id == "ext-a" From e5970adfa69259baefef8a6c1e58d0f7b413d474 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Sun, 5 Jul 2026 22:33:54 +0000 Subject: [PATCH 3/7] refactor: parse Somfy BOB sites into typed models; rename warm/cold start to resume/fresh Replace the hand-walked triple-nested dict traversal in discover_gateways with typed BobSite/BobSubSite/BobGateway models parsed by a dedicated BOB cattrs converter, flattened via BobSitesResponse.gateway_candidates(). Carry each site's country on GatewayCandidate, dropping the parallel _site_country side-channel that discover_gateways and select_gateway shared. Also rename the warm/cold-start terminology to the industry-standard resume/fresh-login: SomfyTokenCredentials now yields a resumed session, _warm_start -> _resume_session, warm_start_credentials() -> to_credentials(). --- pyoverkiz/auth/base.py | 1 + pyoverkiz/auth/bob.py | 91 +++++++++++++++++++++++++++++++++++ pyoverkiz/auth/credentials.py | 2 +- pyoverkiz/auth/factory.py | 2 +- pyoverkiz/auth/strategies.py | 62 ++++++++---------------- tests/test_auth.py | 35 +++++++------- 6 files changed, 132 insertions(+), 61 deletions(-) create mode 100644 pyoverkiz/auth/bob.py diff --git a/pyoverkiz/auth/base.py b/pyoverkiz/auth/base.py index 3598bc5e..3a0f53d9 100644 --- a/pyoverkiz/auth/base.py +++ b/pyoverkiz/auth/base.py @@ -66,6 +66,7 @@ class GatewayCandidate: home_id: str | None = None label: str | None = None external_id: str | None = None + country: str | None = None @runtime_checkable diff --git a/pyoverkiz/auth/bob.py b/pyoverkiz/auth/bob.py new file mode 100644 index 00000000..12f6dd9d --- /dev/null +++ b/pyoverkiz/auth/bob.py @@ -0,0 +1,91 @@ +"""Models for the Somfy BOB back-office site directory. + +BOB is a separate service from the Overkiz enduser API, with its own payload +shapes (``siteOID``, ``subSites``, ``externalOID``). It therefore gets its own +tiny cattrs converter here rather than sharing ``pyoverkiz.converter``, which is +scoped to the enduser API and its camelCase convention. +""" + +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 7d2372c8..603f3cfb 100644 --- a/pyoverkiz/auth/credentials.py +++ b/pyoverkiz/auth/credentials.py @@ -33,7 +33,7 @@ class LocalTokenCredentials(TokenCredentials): @dataclass(slots=True) class SomfyTokenCredentials(Credentials): - """Warm-start credentials for a previously-selected Somfy site. + """Resume credentials for a previously-selected Somfy site. Skips the password grant, Keycloak token exchange, and site discovery on reload: the caller persists the Ginaite ``refresh_token`` plus the selected diff --git a/pyoverkiz/auth/factory.py b/pyoverkiz/auth/factory.py index 5a085b18..84772685 100644 --- a/pyoverkiz/auth/factory.py +++ b/pyoverkiz/auth/factory.py @@ -65,7 +65,7 @@ def build_auth_strategy( ) if server == Server.SOMFY: - # Warm start from a persisted site-scoped refresh token, or cold start + # Resume from a persisted site-scoped refresh token, or fresh login # from username/password. if not isinstance(credentials, SomfyTokenCredentials): credentials = _ensure_credentials(credentials, UsernamePasswordCredentials) diff --git a/pyoverkiz/auth/strategies.py b/pyoverkiz/auth/strategies.py index 0e545b1c..5b10ef0d 100644 --- a/pyoverkiz/auth/strategies.py +++ b/pyoverkiz/auth/strategies.py @@ -19,6 +19,7 @@ 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, @@ -300,33 +301,32 @@ def __init__( ) -> None: """Create a Somfy multi-site strategy with a fresh auth context. - Accepts either ``UsernamePasswordCredentials`` (cold start: password - grant + token exchange + discovery) or ``SomfyTokenCredentials`` (warm - start: a persisted refresh token scoped to an already-selected site, - skipping all three network round trips). + Accepts either ``UsernamePasswordCredentials`` (fresh login: password + grant + token exchange + discovery) or ``SomfyTokenCredentials`` + (resumed session: a persisted refresh token scoped to an already-selected + site, skipping all three network round trips). """ super().__init__(session, server, ssl_context) self.credentials = credentials self.context = AuthContext() self._sites: list[GatewayCandidate] = [] - self._site_country: dict[str, str] = {} self._selected_site_oid: str | None = None self._selected_gateway: str | None = None self._selected_region: str | None = None self._endpoint: str | None = None - # Warm-start refresh-token persistence (no-op for cold start). + # 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: - """Cold start (password) or warm start (persisted refresh token). + """Fresh login (password) or resumed session (persisted refresh token). - With ``SomfyTokenCredentials`` this is a warm start: no password grant, + With ``SomfyTokenCredentials`` this resumes a session: no password grant, no token exchange, no discovery. We seed the context from the stored refresh token and site scope, so the first request mints a site-scoped access token via the existing refresh path. - With ``UsernamePasswordCredentials`` this is the cold start: password + With ``UsernamePasswordCredentials`` this is a fresh login: password grant -> token exchange -> discover, then (re-)select a site. Relogin (e.g. after ``NotAuthenticatedError``) mints a fresh, unscoped Ginaite token that is not itself expired, so on a multi-site account the @@ -337,7 +337,7 @@ async def login(self) -> None: silently pointing at a gateway that's gone. """ if isinstance(self.credentials, SomfyTokenCredentials): - self._warm_start(self.credentials) + self._resume_session(self.credentials) return token = await _somfy_password_token( @@ -356,7 +356,7 @@ async def login(self) -> None: elif len(self._sites) == 1: self.select_gateway(self._sites[0].gateway_id) - def _warm_start(self, credentials: SomfyTokenCredentials) -> None: + def _resume_session(self, credentials: SomfyTokenCredentials) -> None: """Seed site scope from persisted tokens; skip password/exchange/discover. Sets up exactly the state ``select_gateway`` would have produced, then @@ -395,30 +395,9 @@ async def _token_exchange(self, sso_access_token: str) -> None: 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") - candidates: list[GatewayCandidate] = [] - self._site_country = {} - for site in data.get("results", []): - site_oid = str(site["siteOID"]) - label = site.get("name") - country = site.get("country") - for sub in site.get("subSites", []): - external_id = sub.get("externalOID") - for gateway in sub.get("gateways", []): - gateway_id = str(gateway["gatewayId"]) - if country is not None: - self._site_country[gateway_id] = str(country) - candidates.append( - GatewayCandidate( - gateway_id=gateway_id, - home_id=site_oid, - label=label, - external_id=( - str(external_id) if external_id is not None else None - ), - ) - ) - self._sites = candidates - return candidates + 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.""" @@ -429,7 +408,7 @@ def select_gateway(self, gateway_id: str) -> None: if site is None: raise SomfyServiceError(f"Unknown gateway id: {gateway_id}") - region = self._region_for_country(self._site_country.get(gateway_id)) + region = self._region_for_country(site.country) self._selected_gateway = gateway_id self._selected_site_oid = site.home_id @@ -462,11 +441,11 @@ def selected_gateway(self) -> str | None: """Return the currently selected gateway id, or None.""" return self._selected_gateway - def warm_start_credentials( + def to_credentials( self, on_token_refresh: Callable[[str], Awaitable[None]] | None = None, ) -> SomfyTokenCredentials: - """Snapshot the current session as reusable warm-start credentials. + """Snapshot the current session as reusable resume credentials. Call this after login + gateway selection to persist the state a reload needs (refresh token + site scope + region), skipping password grant, @@ -479,7 +458,7 @@ def warm_start_credentials( or self.context.refresh_token is None ): raise SomfyServiceError( - "Cannot snapshot warm-start credentials before a site is " + "Cannot snapshot resume credentials before a site is " "selected and a refresh token is available." ) return SomfyTokenCredentials( @@ -534,14 +513,13 @@ async def _refresh(self) -> None: ) self.context.update_from_token(await response.json()) - # Ginaite may omit refresh_token on a refresh; keep the working one - # rather than dropping to None (which would break the next warm start). + # 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 warm-start caller persist a rotated refresh token (no-op otherwise).""" + """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 diff --git a/tests/test_auth.py b/tests/test_auth.py index 51817e6e..ba008c7a 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -239,7 +239,7 @@ async def test_build_auth_strategy_somfy_multisite(self): assert isinstance(strategy, SomfyAccountAuthStrategy) def test_build_auth_strategy_somfy_token_credentials(self): - """Server.SOMFY + SomfyTokenCredentials builds the warm-start strategy.""" + """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 @@ -1637,6 +1637,7 @@ async def test_somfy_multisite_discover_flattens_sites(): 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 @@ -1897,7 +1898,7 @@ async def test_somfy_multisite_relogin_drops_removed_gateway(): assert strategy.endpoint == strategy.server.endpoint -def _build_somfy_warmstart_strategy(**credential_overrides): +def _build_somfy_resume_strategy(**credential_overrides): """Return a SomfyAccountAuthStrategy built from SomfyTokenCredentials.""" from unittest.mock import MagicMock @@ -1926,9 +1927,9 @@ def _build_somfy_warmstart_strategy(**credential_overrides): @pytest.mark.asyncio -async def test_somfy_warmstart_login_seeds_scope_without_http(): - """Warm-start login performs no network calls and seeds the site scope.""" - strategy, session = _build_somfy_warmstart_strategy() +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() @@ -1946,9 +1947,9 @@ async def test_somfy_warmstart_login_seeds_scope_without_http(): @pytest.mark.asyncio -async def test_somfy_warmstart_first_refresh_scopes_to_site(): - """The first refresh after warm start mints a token via the ?siteOID URL.""" - strategy, session = _build_somfy_warmstart_strategy() +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( @@ -1962,8 +1963,8 @@ async def test_somfy_warmstart_first_refresh_scopes_to_site(): @pytest.mark.asyncio -async def test_somfy_warmstart_credentials_roundtrip(): - """warm_start_credentials() snapshots a cold session's reusable state.""" +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" @@ -1971,7 +1972,7 @@ async def test_somfy_warmstart_credentials_roundtrip(): await strategy.discover_gateways() strategy.select_gateway("1225-0000-0002") - snapshot = strategy.warm_start_credentials() + snapshot = strategy.to_credentials() assert snapshot.refresh_token == "r-1" assert snapshot.site_oid == "site-b" @@ -1980,25 +1981,25 @@ async def test_somfy_warmstart_credentials_roundtrip(): @pytest.mark.asyncio -async def test_somfy_warmstart_credentials_requires_selection(): +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.warm_start_credentials() + strategy.to_credentials() @pytest.mark.asyncio -async def test_somfy_warmstart_rotated_refresh_token_notifies(): +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_warmstart_strategy(on_token_refresh=_persist) + strategy, session = _build_somfy_resume_strategy(on_token_refresh=_persist) await strategy.login() session.post = MagicMock( @@ -2010,9 +2011,9 @@ async def _persist(token): @pytest.mark.asyncio -async def test_somfy_warmstart_missing_refresh_token_preserved(): +async def test_somfy_resume_missing_refresh_token_preserved(): """A refresh response without a refresh_token keeps the working one.""" - strategy, session = _build_somfy_warmstart_strategy() + strategy, session = _build_somfy_resume_strategy() await strategy.login() # Ginaite may omit refresh_token on refresh; the old one must survive. From dd08d1da821a7561b4716ab17f662d5e0cbe4da1 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Sun, 5 Jul 2026 22:36:11 +0000 Subject: [PATCH 4/7] docs: shorten Server.SOMFY placeholder-endpoint comment --- pyoverkiz/const.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/pyoverkiz/const.py b/pyoverkiz/const.py index 55bde647..141fbd8e 100644 --- a/pyoverkiz/const.py +++ b/pyoverkiz/const.py @@ -241,10 +241,8 @@ ), Server.SOMFY: ServerConfig( server=Server.SOMFY, - # Region-agnostic multi-site login. The endpoint here is a - # placeholder; SomfyAccountAuthStrategy overrides it per selected - # site once the region is resolved. 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, From 55e35b9c4d455f01facb53c53c92951d898b16e9 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Sun, 5 Jul 2026 22:45:45 +0000 Subject: [PATCH 5/7] feat: expose session resume publicly; document Somfy multi-account; trim docstrings Add SupportsSessionResume + OverkizClient.to_credentials() so the Somfy resume flow works through the public API instead of reaching into the private auth strategy. Document multi-account login and session resume in the getting-started guide. Condense the verbose docstrings/comments added earlier in this branch to one-liners. --- docs/getting-started.md | 71 +++++++++++++++++++++++++++++++++++ pyoverkiz/auth/base.py | 18 ++++++++- pyoverkiz/auth/bob.py | 7 ++-- pyoverkiz/auth/credentials.py | 14 ++----- pyoverkiz/auth/strategies.py | 66 ++++++-------------------------- pyoverkiz/client.py | 26 ++++++++++++- tests/test_client.py | 21 +++++++++++ 7 files changed, 152 insertions(+), 71 deletions(-) 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 3a0f53d9..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) @@ -82,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 index 12f6dd9d..0d403650 100644 --- a/pyoverkiz/auth/bob.py +++ b/pyoverkiz/auth/bob.py @@ -1,9 +1,8 @@ """Models for the Somfy BOB back-office site directory. -BOB is a separate service from the Overkiz enduser API, with its own payload -shapes (``siteOID``, ``subSites``, ``externalOID``). It therefore gets its own -tiny cattrs converter here rather than sharing ``pyoverkiz.converter``, which is -scoped to the enduser API and its camelCase convention. +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 diff --git a/pyoverkiz/auth/credentials.py b/pyoverkiz/auth/credentials.py index 603f3cfb..f67d07ea 100644 --- a/pyoverkiz/auth/credentials.py +++ b/pyoverkiz/auth/credentials.py @@ -33,17 +33,11 @@ class LocalTokenCredentials(TokenCredentials): @dataclass(slots=True) class SomfyTokenCredentials(Credentials): - """Resume credentials for a previously-selected Somfy site. + """Resume credentials for a previously-selected Somfy site (skips login + discovery). - Skips the password grant, Keycloak token exchange, and site discovery on - reload: the caller persists the Ginaite ``refresh_token`` plus the selected - site's ``site_oid`` and ``region``, and pyoverkiz mints a site-scoped access - token directly on the first request. ``gateway_id`` is optional bookkeeping - (the id the user selected) and is surfaced via ``selected_gateway``. - - Ginaite rotates the refresh token on refresh, so supply an async - ``on_token_refresh`` callback to re-persist the new refresh token; without - it a rotated token is only kept in memory and a later reload would fail. + 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) diff --git a/pyoverkiz/auth/strategies.py b/pyoverkiz/auth/strategies.py index 5b10ef0d..1f7eccbc 100644 --- a/pyoverkiz/auth/strategies.py +++ b/pyoverkiz/auth/strategies.py @@ -284,12 +284,10 @@ async def _request_access_token( class SomfyAccountAuthStrategy(BaseAuthStrategy): - """Somfy multi-site auth: Keycloak token exchange + BOB site directory. + """Somfy multi-site auth: password grant -> Keycloak token exchange -> BOB site directory. - Reuses the Somfy Accounts password grant, exchanges the SSO token for a - Ginaite (Keycloak) token, then lists the account's sites from the BOB - directory. Selecting a site mints a site-scoped token whose Bearer drives - the classic Overkiz enduser API directly (no gateway header needed). + Selecting a site mints a site-scoped token whose Bearer drives the classic + Overkiz enduser API directly (no gateway header needed). """ def __init__( @@ -299,13 +297,7 @@ def __init__( server: ServerConfig, ssl_context: ssl.SSLContext | bool, ) -> None: - """Create a Somfy multi-site strategy with a fresh auth context. - - Accepts either ``UsernamePasswordCredentials`` (fresh login: password - grant + token exchange + discovery) or ``SomfyTokenCredentials`` - (resumed session: a persisted refresh token scoped to an already-selected - site, skipping all three network round trips). - """ + """Accept ``UsernamePasswordCredentials`` (fresh login) or ``SomfyTokenCredentials`` (resumed session).""" super().__init__(session, server, ssl_context) self.credentials = credentials self.context = AuthContext() @@ -319,23 +311,7 @@ def __init__( self._persisted_refresh_token: str | None = None async def login(self) -> None: - """Fresh login (password) or resumed session (persisted refresh token). - - With ``SomfyTokenCredentials`` this resumes a session: no password grant, - no token exchange, no discovery. We seed the context from the stored - refresh token and site scope, so the first request mints a site-scoped - access token via the existing refresh path. - - With ``UsernamePasswordCredentials`` this is a fresh login: password - grant -> token exchange -> discover, then (re-)select a site. Relogin - (e.g. after ``NotAuthenticatedError``) mints a fresh, unscoped Ginaite - token that is not itself expired, so on a multi-site account the - previously-selected gateway must be re-selected to re-apply site - scoping; otherwise the unscoped global token would keep being served - against the still-selected region endpoint. If that gateway is no - longer present after rediscovery, drop the stale selection instead of - silently pointing at a gateway that's gone. - """ + """Fresh login (password grant -> exchange -> discover) or resumed session.""" if isinstance(self.credentials, SomfyTokenCredentials): self._resume_session(self.credentials) return @@ -346,6 +322,8 @@ async def login(self) -> None: 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) @@ -357,13 +335,7 @@ async def login(self) -> None: self.select_gateway(self._sites[0].gateway_id) def _resume_session(self, credentials: SomfyTokenCredentials) -> None: - """Seed site scope from persisted tokens; skip password/exchange/discover. - - Sets up exactly the state ``select_gateway`` would have produced, then - marks the (absent) access token expired so the first request mints a - site-scoped token via the existing ``refresh_if_needed`` -> ``_refresh`` - path. No network calls happen here. - """ + """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 @@ -419,13 +391,7 @@ def select_gateway(self, gateway_id: str) -> None: @staticmethod def _region_for_country(country: str | None) -> str: - """Map an ISO country to an Overkiz region, defaulting to EMEA. - - Mirrors the TaHoma app's BusinessArea.fromCountry: known countries map - to their region, and anything unresolvable falls back to EMEA. A country - we cannot resolve (missing, or present but unmapped) is logged, since it - likely means the map needs updating for a newly supported region. - """ + """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( @@ -445,12 +411,9 @@ def to_credentials( self, on_token_refresh: Callable[[str], Awaitable[None]] | None = None, ) -> SomfyTokenCredentials: - """Snapshot the current session as reusable resume credentials. + """Snapshot the session (refresh token + site scope) as resume credentials. - Call this after login + gateway selection to persist the state a reload - needs (refresh token + site scope + region), skipping password grant, - token exchange, and discovery next time. Raises if no site is selected - or no refresh token is available yet. + Raises if no site is selected or no refresh token is available yet. """ if ( self._selected_site_oid is None @@ -475,12 +438,7 @@ def endpoint(self) -> str: return self._endpoint or self.server.endpoint async def refresh_if_needed(self) -> bool: - """Mint/refresh a site-scoped token when expired. - - Raises if a site has been selected but there is no refresh token to - mint the site-scoped token with, rather than silently continuing to - serve the unscoped global token against the site's region endpoint. - """ + """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: 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/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.""" From 102a145595f88b1a8195b282a96a8963a515bd7c Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Mon, 6 Jul 2026 13:07:20 +0000 Subject: [PATCH 6/7] feat: offer local API for Server.SOMFY multi-account --- pyoverkiz/const.py | 1 + 1 file changed, 1 insertion(+) diff --git a/pyoverkiz/const.py b/pyoverkiz/const.py index 141fbd8e..b3b466ce 100644 --- a/pyoverkiz/const.py +++ b/pyoverkiz/const.py @@ -154,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, From 806dfc011792d33d5b049f13eed7525602c662d7 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Wed, 8 Jul 2026 15:15:50 +0000 Subject: [PATCH 7/7] fix: map Somfy refresh invalid_grant to SomfyBadCredentialsError A revoked refresh token (e.g. after a password change) returns a 400 invalid_grant on the site-scoped refresh grant. Classify it as bad credentials, mirroring the password grant and CozyTouch strategy, so callers trigger reauth instead of surfacing an unexpected error. --- pyoverkiz/auth/strategies.py | 7 +++++++ tests/test_auth.py | 23 +++++++++++++++++++++++ 2 files changed, 30 insertions(+) diff --git a/pyoverkiz/auth/strategies.py b/pyoverkiz/auth/strategies.py index 1f7eccbc..e10d7f54 100644 --- a/pyoverkiz/auth/strategies.py +++ b/pyoverkiz/auth/strategies.py @@ -466,6 +466,13 @@ async def _refresh(self) -> None: 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}" ) diff --git a/tests/test_auth.py b/tests/test_auth.py index ba008c7a..3a530c9d 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -1790,6 +1790,29 @@ async def test_somfy_multisite_refresh_scopes_to_selected_site(): 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.