Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 | ✓ | |
Expand Down
1 change: 1 addition & 0 deletions docs/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 | ✓ | |
Expand Down
9 changes: 9 additions & 0 deletions pyoverkiz/auth/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
NexityAuthStrategy,
RexelAuthStrategy,
RexelTokenAuthStrategy,
SauterAuthStrategy,
SessionLoginStrategy,
SomfyAccountAuthStrategy,
SomfyAuthStrategy,
Expand Down Expand Up @@ -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(
Expand Down
15 changes: 12 additions & 3 deletions pyoverkiz/auth/strategies.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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).
Expand All @@ -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,
Expand All @@ -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):
Comment thread
Mincka marked this conversation as resolved.
Resolved
"""Sauter Smart Control: Brandt middleware flow with the Sauter partner."""

partner: str = SAUTER_PARTNER


class NexityAuthStrategy(SessionLoginStrategy):
"""Authentication strategy using Nexity session-based login."""

Expand Down
9 changes: 9 additions & 0 deletions pyoverkiz/const.py
Original file line number Diff line number Diff line change
Expand Up @@ -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/"

Expand Down Expand Up @@ -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)",
Expand Down
1 change: 1 addition & 0 deletions pyoverkiz/enums/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
90 changes: 66 additions & 24 deletions tests/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,13 @@
)
from pyoverkiz.auth.strategies import (
BearerTokenAuthStrategy,
BrandtAuthStrategy,
CozytouchAuthStrategy,
LocalTokenAuthStrategy,
NexityAuthStrategy,
RexelAuthStrategy,
RexelTokenAuthStrategy,
SauterAuthStrategy,
SessionLoginStrategy,
SomfyAuthStrategy,
_decode_jwt_payload,
Expand Down Expand Up @@ -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."""
Expand Down Expand Up @@ -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,
Expand All @@ -701,59 +727,63 @@ 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"
)
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)
Expand All @@ -764,22 +794,34 @@ 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)
session.post = MagicMock(return_value=self._json_response(200, {"client": {}}))
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()

Expand Down
Loading