From 93587a698b06c0c630ca1998a64fb59f601104f1 Mon Sep 17 00:00:00 2001 From: Mick Vleeshouwer Date: Sat, 24 Oct 2026 15:02:52 +0200 Subject: [PATCH] fix: Handle server errors in Brandt login Route 5xx from the Brandt middleware sessions.json and profile/jwt.json calls through _raise_for_server_error so outages raise ServiceUnavailableError instead of BrandtBadCredentialsError or BrandtServiceError. Parse the sessions.json error body with _json_body so a non-JSON 4xx no longer leaks aiohttp.ContentTypeError. --- pyoverkiz/auth/strategies.py | 13 +++--- tests/test_auth.py | 85 +++++++++++++++++++++++++++++++++++- 2 files changed, 89 insertions(+), 9 deletions(-) diff --git a/pyoverkiz/auth/strategies.py b/pyoverkiz/auth/strategies.py index fcd05e6c..fdc705e6 100644 --- a/pyoverkiz/auth/strategies.py +++ b/pyoverkiz/auth/strategies.py @@ -712,15 +712,13 @@ async def login(self) -> None: }, ssl=self._ssl, ) as response: + await _raise_for_server_error(response) if response.status >= HTTPStatus.BAD_REQUEST: message = "Login failed: bad credentials" - try: - body = await response.json() - errors = body.get("error") - if errors: - message = errors[0] - except ValueError: - pass + body = await _json_body(response) + errors = body.get("error") + if errors: + message = errors[0] raise BrandtBadCredentialsError(message) # 2) Fetch the JWT, authenticated purely by the session cookie. @@ -728,6 +726,7 @@ async def login(self) -> None: f"{BRANDT_MIDDLEWARE_API}/api/v1/profile/jwt.json", ssl=self._ssl, ) as response: + await _raise_for_server_error(response) if response.status >= HTTPStatus.BAD_REQUEST: raise BrandtServiceError( f"Brandt JWT request failed: {response.status}" diff --git a/tests/test_auth.py b/tests/test_auth.py index c68eccd4..0b59a0c7 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -692,10 +692,34 @@ def _server_config(): ) @staticmethod - def _json_response(status, payload): + def _json_response(status, payload, *, html: str | None = None): + """Return a mock middleware response. + + With `html`, the body is non-JSON and `json()` fails the way aiohttp + does: ContentTypeError by default, JSONDecodeError with + `content_type=None` (as `check_response` calls it). + """ + from json import JSONDecodeError + + from aiohttp import ContentTypeError + resp = MagicMock() resp.status = status - resp.json = AsyncMock(return_value=payload) + if html is None: + resp.json = AsyncMock(return_value=payload) + else: + + async def _json(content_type: str | None = "application/json"): + if content_type is None: + raise JSONDecodeError("Expecting value", html, 0) + raise ContentTypeError( + request_info=MagicMock(), + history=(), + message="Attempt to decode JSON with unexpected mimetype: text/html", + ) + + resp.json = AsyncMock(side_effect=_json) + resp.text = AsyncMock(return_value=html) resp.__aenter__ = AsyncMock(return_value=resp) resp.__aexit__ = AsyncMock(return_value=None) return resp @@ -768,6 +792,63 @@ async def test_login_bad_credentials(self): with pytest.raises(BrandtBadCredentialsError, match="Password wrong password"): await strategy.login() + @pytest.mark.asyncio + async def test_login_sessions_503_raises_service_unavailable(self): + """A middleware outage must not be reported as bad credentials.""" + from pyoverkiz.auth.strategies import BrandtAuthStrategy + from pyoverkiz.exceptions import ServiceUnavailableError + + session = AsyncMock(spec=ClientSession) + session.post = MagicMock( + return_value=self._json_response( + 503, None, html="503 Service Unavailable" + ) + ) + + credentials = UsernamePasswordCredentials("a@b.c", "pass") + strategy = BrandtAuthStrategy(credentials, session, self._server_config(), True) + with pytest.raises(ServiceUnavailableError): + await strategy.login() + + @pytest.mark.asyncio + async def test_login_non_json_401_raises_bad_credentials(self): + """A non-JSON 401 (e.g. proxy HTML page) still maps to bad credentials.""" + from pyoverkiz.auth.strategies import BrandtAuthStrategy + from pyoverkiz.exceptions import BrandtBadCredentialsError + + session = AsyncMock(spec=ClientSession) + session.post = MagicMock( + return_value=self._json_response( + 401, None, html="401 Unauthorized" + ) + ) + + credentials = UsernamePasswordCredentials("a@b.c", "wrong") + strategy = BrandtAuthStrategy(credentials, session, self._server_config(), True) + with pytest.raises(BrandtBadCredentialsError, match="bad credentials"): + await strategy.login() + + @pytest.mark.asyncio + async def test_login_jwt_503_raises_service_unavailable(self): + """A 503 from profile/jwt.json raises ServiceUnavailableError.""" + from pyoverkiz.auth.strategies import BrandtAuthStrategy + from pyoverkiz.exceptions import ServiceUnavailableError + + session = AsyncMock(spec=ClientSession) + session.post = MagicMock( + return_value=self._json_response(200, {"client": {"email": "a@b.c"}}) + ) + session.get = MagicMock( + return_value=self._json_response( + 503, None, html="503 Service Unavailable" + ) + ) + + credentials = UsernamePasswordCredentials("a@b.c", "pass") + strategy = BrandtAuthStrategy(credentials, session, self._server_config(), True) + with pytest.raises(ServiceUnavailableError): + await strategy.login() + @pytest.mark.asyncio async def test_login_missing_jwt_raises_service_error(self): """Login OK but no jwt in the profile response -> BrandtServiceError."""