refactor: Add UserLockedOutError exception and handle lockout in Mellophone client

Author: GingerTe

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "mellophone-valve"
-version = "3.2.0"
+version = "3.2.1"
 description = "Python-клиент для Mellophone"
 readme = "README.md"
 requires-python = ">=3.7"

src/mellophone/__init__.py

@@ -13,6 +13,7 @@
     ServerError,
     TransportError,
     UnauthorizedError,
+    UserLockedOutError,
 )
 
 __all__ = [
@@ -22,6 +23,7 @@
     "BadRequestError",
     "UnauthorizedError",
     "ForbiddenError",
+    "UserLockedOutError",
     "NotFoundError",
     "ServerError",
     "AsyncClientUnavailableError",

src/mellophone/client.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import re
 import warnings
 import xml.etree.ElementTree as ET
 from http import HTTPStatus
@@ -20,24 +21,30 @@
     ServerError,
     TransportError,
     UnauthorizedError,
+    UserLockedOutError,
 )
 from .structures import RequestArgs, RequestParams
 from .utils import user_to_xml, xml_to_json
 
 try:
     import httpx
 except ImportError:  # pragma: no cover - depends on installed extra
-    httpx = None  # type: ignore[assignment]
+    httpx = None  # type: ignore[assignment]  # ty:ignore[invalid-assignment]
 
 try:
     import requests
 except ImportError:  # pragma: no cover - depends on installed extra
-    requests = None  # type: ignore[assignment]
+    requests = None  # type: ignore[assignment]  # ty:ignore[invalid-assignment]
 
 
 class Mellophone:
     """Единый клиент Mellophone с синхронными и асинхронными методами."""
 
+    _LOCKOUT_RE = re.compile(
+        r"locked out for too many unsuccessful login attempts\.\s*Time to unlock:\s*(\d+)\s*s\.",
+        re.IGNORECASE,
+    )
+
     base_url: str
     token_set_settings: Optional[str]
     token_user_manage: Optional[str]
@@ -128,6 +135,13 @@ def _build_url(self, path: str, params: Optional[Dict[str, Any]] = None) -> str:
         clean = {k: v for k, v in params.items() if v is not None}
         return f"{self.base_url}{path}?{urlencode(clean)}"
 
+    @staticmethod
+    def _match_lockout_error(response_text: str) -> Optional[int]:
+        match = Mellophone._LOCKOUT_RE.search(response_text)
+        if match is None:
+            return None
+        return int(match.group(1))
+
     @staticmethod
     def _raise_for_status(response: Any) -> None:
         """Выбрасывает доменное исключение для неуспешных HTTP-статусов."""
@@ -139,6 +153,9 @@ def _raise_for_status(response: Any) -> None:
         if status_code == HTTPStatus.UNAUTHORIZED:
             raise UnauthorizedError(status_code, response_text)
         if status_code == HTTPStatus.FORBIDDEN:
+            unlock_in_seconds = Mellophone._match_lockout_error(response_text)
+            if unlock_in_seconds is not None:
+                raise UserLockedOutError(status_code, response_text, unlock_in_seconds)
             raise ForbiddenError(status_code, response_text)
         if status_code == HTTPStatus.NOT_FOUND:
             raise NotFoundError(status_code, response_text)

src/mellophone/exceptions.py

@@ -22,6 +22,14 @@ class ForbiddenError(HttpError):
     """Raised for HTTP 403."""
 
 
+class UserLockedOutError(ForbiddenError):
+    """Raised when API reports user lockout after too many failed login attempts."""
+
+    def __init__(self, status_code: int, response_text: str, unlock_in_seconds: int) -> None:
+        self.unlock_in_seconds = unlock_in_seconds
+        super().__init__(status_code, response_text)
+
+
 class NotFoundError(HttpError):
     """Raised for HTTP 404."""
 
@@ -55,6 +63,7 @@ class MissingTokenError(ValueError):
     "BadRequestError",
     "UnauthorizedError",
     "ForbiddenError",
+    "UserLockedOutError",
     "NotFoundError",
     "ServerError",
     "AsyncClientUnavailableError",

tests/unit/test_client_common_methods.py

@@ -77,6 +77,23 @@ def test_bad_request_maps_to_domain_error(mode_and_mock, response_factory):
     assert exc.value.status_code == 400
 
 
+def test_user_lockout_maps_to_specific_domain_error(mode_and_mock, response_factory):
+    mode, install = mode_and_mock
+    install(
+        response_factory(
+            403,
+            "User expertise@curs.ru is locked out for too many unsuccessful login attempts. Time to unlock: 305 s.",
+        )
+    )
+
+    client = mellophone.Mellophone("http://example.com")
+    with pytest.raises(mellophone.UserLockedOutError) as exc:
+        _call(client, mode, "login", "john", "secret", ses_id="ses-1")
+
+    assert exc.value.status_code == 403
+    assert exc.value.unlock_in_seconds == 305
+
+
 def test_server_error_maps_to_domain_error(mode_and_mock, response_factory):
     mode, install = mode_and_mock
     install(response_factory(500, "server error"))