diff --git a/README.md b/README.md index 197db771..80919a90 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,7 @@ Full documentation is available at **[imicknl.github.io/python-overkiz-api](http | Nexity Eugénie | ✓ | | | Rexel Energeasy Connect † | ✓ | ✓ | | Sauter Cozytouch | ✓ | | +| Sauter Smart Control | ✓ | | | Simu LiveIn2 | ✓ | | | Somfy | ✓ | ✓ | | Thermor Cozytouch | ✓ | | diff --git a/docs/index.md b/docs/index.md index 3d4c7ad9..9f835595 100644 --- a/docs/index.md +++ b/docs/index.md @@ -32,6 +32,7 @@ pyOverkiz is an async Python library for interacting with Overkiz-based platform | Nexity Eugénie | ✓ | | | Rexel Energeasy Connect † | ✓ | ✓ | | Sauter Cozytouch | ✓ | | +| Sauter Smart Control | ✓ | | | Simu LiveIn2 | ✓ | | | Somfy | ✓ | ✓ | | Thermor Cozytouch | ✓ | | diff --git a/pyoverkiz/auth/factory.py b/pyoverkiz/auth/factory.py index 84772685..3a1efb03 100644 --- a/pyoverkiz/auth/factory.py +++ b/pyoverkiz/auth/factory.py @@ -24,6 +24,7 @@ NexityAuthStrategy, RexelAuthStrategy, RexelTokenAuthStrategy, + SauterAuthStrategy, SessionLoginStrategy, SomfyAccountAuthStrategy, SomfyAuthStrategy, @@ -104,6 +105,14 @@ def build_auth_strategy( ssl_context, ) + if server == Server.SAUTER_SMART_CONTROL: + return SauterAuthStrategy( + _ensure_credentials(credentials, UsernamePasswordCredentials), + session, + server_config, + ssl_context, + ) + if server == Server.REXEL: if isinstance(credentials, RexelTokenCredentials): return RexelTokenAuthStrategy( diff --git a/pyoverkiz/auth/strategies.py b/pyoverkiz/auth/strategies.py index fcd05e6c..db903098 100644 --- a/pyoverkiz/auth/strategies.py +++ b/pyoverkiz/auth/strategies.py @@ -43,6 +43,7 @@ REXEL_OAUTH_SCOPE, REXEL_OAUTH_TOKEN_URL, REXEL_REQUIRED_CONSENT, + SAUTER_PARTNER, SOMFY_API, SOMFY_BOB_API_KEY, SOMFY_BOB_SITE_API, @@ -698,6 +699,8 @@ class BrandtAuthStrategy(SessionLoginStrategy): aiohttp session carries the cookie between the two middleware calls. """ + partner: str = BRANDT_PARTNER + async def login(self) -> None: """Perform the Brandt middleware login, then Overkiz JWT login.""" # 1) Middleware session login (sets the devise session cookie). @@ -707,7 +710,7 @@ async def login(self) -> None: "client": { "email": self.credentials.username, "password": self.credentials.password, - "partner": BRANDT_PARTNER, + "partner": self.partner, } }, ssl=self._ssl, @@ -730,18 +733,24 @@ async def login(self) -> None: ) as response: if response.status >= HTTPStatus.BAD_REQUEST: raise BrandtServiceError( - f"Brandt JWT request failed: {response.status}" + f"{self.server.name} JWT request failed: {response.status}" ) body = await response.json() jwt = body.get("client", {}).get("jwt") if not jwt: - raise BrandtServiceError("No Brandt JWT token provided.") + raise BrandtServiceError(f"No {self.server.name} JWT token provided.") # 3) Overkiz cloud login with the JWT only (no apiKey/applicationId). await self._post_login({"jwt": jwt}) +class SauterAuthStrategy(BrandtAuthStrategy): + """Sauter Smart Control: Brandt middleware flow with the Sauter partner.""" + + partner: str = SAUTER_PARTNER + + class NexityAuthStrategy(SessionLoginStrategy): """Authentication strategy using Nexity session-based login.""" diff --git a/pyoverkiz/const.py b/pyoverkiz/const.py index bbc16575..1a821f98 100644 --- a/pyoverkiz/const.py +++ b/pyoverkiz/const.py @@ -195,6 +195,8 @@ # Brandt Smart Control middleware (cookie-session Rails API in front of Overkiz) BRANDT_MIDDLEWARE_API = "https://www.smartcontrol-app.com" BRANDT_PARTNER = "brandt-electromenager" +# Sauter Smart Control: same middleware as Brandt, different partner/realm. +SAUTER_PARTNER = "sauter-electromenager" LOCAL_API_PATH = "/enduser-mobile-web/1/enduserAPI/" @@ -278,6 +280,13 @@ manufacturer="Sauter", api_type=APIType.CLOUD, ), + Server.SAUTER_SMART_CONTROL: ServerConfig( + server=Server.SAUTER_SMART_CONTROL, + name="Sauter Smart Control", + endpoint="https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Sauter", + api_type=APIType.CLOUD, + ), Server.SIMU_LIVEIN2: ServerConfig( # alias of https://tahomalink.com server=Server.SIMU_LIVEIN2, name="SIMU (LiveIn2)", diff --git a/pyoverkiz/enums/server.py b/pyoverkiz/enums/server.py index 6dd48331..c7c4f3a6 100644 --- a/pyoverkiz/enums/server.py +++ b/pyoverkiz/enums/server.py @@ -25,6 +25,7 @@ class Server(StrEnum): NEXITY = "nexity" REXEL = "rexel" SAUTER_COZYTOUCH = "sauter_cozytouch" + SAUTER_SMART_CONTROL = "sauter_smart_control" SIMU_LIVEIN2 = "simu_livein2" # Experimental: may change within 2.x. SOMFY = "somfy" diff --git a/tests/test_auth.py b/tests/test_auth.py index c68eccd4..c0ce8b45 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -32,11 +32,13 @@ ) from pyoverkiz.auth.strategies import ( BearerTokenAuthStrategy, + BrandtAuthStrategy, CozytouchAuthStrategy, LocalTokenAuthStrategy, NexityAuthStrategy, RexelAuthStrategy, RexelTokenAuthStrategy, + SauterAuthStrategy, SessionLoginStrategy, SomfyAuthStrategy, _decode_jwt_payload, @@ -310,6 +312,30 @@ async def test_build_auth_strategy_brandt(self): assert isinstance(strategy, BrandtAuthStrategy) + @pytest.mark.asyncio + async def test_build_auth_strategy_sauter(self): + """Test building Sauter auth strategy.""" + from pyoverkiz.auth.strategies import SauterAuthStrategy + + server_config = ServerConfig( + server=Server.SAUTER_SMART_CONTROL, + name="Sauter Smart Control", + endpoint="https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Sauter", + api_type=APIType.CLOUD, + ) + credentials = UsernamePasswordCredentials("user", "pass") + session = AsyncMock(spec=ClientSession) + + strategy = build_auth_strategy( + server_config=server_config, + credentials=credentials, + session=session, + ssl_context=True, + ) + + assert isinstance(strategy, SauterAuthStrategy) + @pytest.mark.asyncio async def test_build_auth_strategy_nexity(self): """Test building Nexity auth strategy.""" @@ -678,14 +704,14 @@ async def test_auth_headers_no_token(self): assert headers == {} -class TestBrandtAuthStrategy: - """Test BrandtAuthStrategy.""" +class TestBrandtSauterAuthStrategy: + """Brandt and Sauter share the same middleware flow; only the partner differs.""" @staticmethod - def _server_config(): + def _server_config(name): return ServerConfig( server=Server.BRANDT, - name="Brandt Smart Control", + name=name, endpoint="https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/", manufacturer="Brandt", api_type=APIType.CLOUD, @@ -701,45 +727,43 @@ def _json_response(status, payload): return resp @pytest.mark.asyncio - async def test_login_success_flow(self): + @pytest.mark.parametrize( + ("strategy_cls", "server_name", "partner"), + [ + (BrandtAuthStrategy, "Brandt Smart Control", "brandt-electromenager"), + (SauterAuthStrategy, "Sauter Smart Control", "sauter-electromenager"), + ], + ) + async def test_login_success_flow(self, strategy_cls, server_name, partner): """Full flow: sessions.json -> profile/jwt.json -> Overkiz login.""" - from pyoverkiz.auth.strategies import BrandtAuthStrategy - session = AsyncMock(spec=ClientSession) - # 1) POST sessions.json 2) Overkiz POST login (inherited _post_login) session.post = MagicMock( side_effect=[ self._json_response(200, {"client": {"email": "a@b.c"}}), self._json_response(200, {"success": True}), ] ) - # GET profile/jwt.json session.get = MagicMock( return_value=self._json_response(200, {"client": {"jwt": "the.jwt.token"}}) ) credentials = UsernamePasswordCredentials("a@b.c", "pass") - strategy = BrandtAuthStrategy(credentials, session, self._server_config(), True) + strategy = strategy_cls( + credentials, session, self._server_config(server_name), True + ) await strategy.login() - # First POST is to the middleware with the partner field. first_post = session.post.call_args_list[0] assert ( first_post.args[0] == "https://www.smartcontrol-app.com/api/v1/sessions.json" ) assert first_post.kwargs["json"] == { - "client": { - "email": "a@b.c", - "password": "pass", - "partner": "brandt-electromenager", - } + "client": {"email": "a@b.c", "password": "pass", "partner": partner} } - # JWT fetched from the cookie-authenticated GET. session.get.assert_called_once_with( "https://www.smartcontrol-app.com/api/v1/profile/jwt.json", ssl=True ) - # Final POST is the Overkiz login carrying only the jwt. last_post = session.post.call_args_list[-1] assert last_post.args[0] == ( "https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/login" @@ -747,13 +771,19 @@ async def test_login_success_flow(self): assert last_post.kwargs["data"] == {"jwt": "the.jwt.token"} @pytest.mark.asyncio - async def test_login_bad_credentials(self): + @pytest.mark.parametrize( + ("strategy_cls", "server_name"), + [ + (BrandtAuthStrategy, "Brandt Smart Control"), + (SauterAuthStrategy, "Sauter Smart Control"), + ], + ) + async def test_login_bad_credentials(self, strategy_cls, server_name): """A wrong-password 400 from sessions.json raises BrandtBadCredentialsError. Pins the real middleware contract: status 400 with an ``error`` array; the first element is surfaced as the exception message. """ - from pyoverkiz.auth.strategies import BrandtAuthStrategy from pyoverkiz.exceptions import BrandtBadCredentialsError session = AsyncMock(spec=ClientSession) @@ -764,14 +794,24 @@ async def test_login_bad_credentials(self): ) credentials = UsernamePasswordCredentials("a@b.c", "wrong") - strategy = BrandtAuthStrategy(credentials, session, self._server_config(), True) + strategy = strategy_cls( + credentials, session, self._server_config(server_name), True + ) with pytest.raises(BrandtBadCredentialsError, match="Password wrong password"): await strategy.login() @pytest.mark.asyncio - async def test_login_missing_jwt_raises_service_error(self): + @pytest.mark.parametrize( + ("strategy_cls", "server_name"), + [ + (BrandtAuthStrategy, "Brandt Smart Control"), + (SauterAuthStrategy, "Sauter Smart Control"), + ], + ) + async def test_login_missing_jwt_raises_service_error( + self, strategy_cls, server_name + ): """Login OK but no jwt in the profile response -> BrandtServiceError.""" - from pyoverkiz.auth.strategies import BrandtAuthStrategy from pyoverkiz.exceptions import BrandtServiceError session = AsyncMock(spec=ClientSession) @@ -779,7 +819,9 @@ async def test_login_missing_jwt_raises_service_error(self): session.get = MagicMock(return_value=self._json_response(200, {"client": {}})) credentials = UsernamePasswordCredentials("a@b.c", "pass") - strategy = BrandtAuthStrategy(credentials, session, self._server_config(), True) + strategy = strategy_cls( + credentials, session, self._server_config(server_name), True + ) with pytest.raises(BrandtServiceError): await strategy.login()