SDK: More resilient HTTPStatus errors

sdk/src/spectrumx/gateway.py

@@ -6,7 +6,6 @@
 from collections.abc import Collection
 from collections.abc import Iterator
 from enum import StrEnum
-from http import HTTPStatus
 from pathlib import Path
 from pathlib import PurePosixPath
 from typing import Annotated
@@ -21,6 +20,7 @@
 from urllib3.util import Retry
 
 from spectrumx.models.captures import CaptureType
+from spectrumx.ops.network import _safe_http_status
 
 from .config import DEFAULT_HTTP_TIMEOUT
 from .errors import AuthError
@@ -258,7 +258,7 @@ def authenticate(self, *, verbose: bool = False) -> None:
         if code is None:
             msg = "No response code received from authentication request."
             raise AuthError(msg)
-        status = HTTPStatus(code)
+        status = _safe_http_status(code)
         log.bind(cat=LogCategory.NETWORK).debug(f"Authentication response: {status}")
         if status.is_success:
             return

sdk/src/spectrumx/ops/network.py

@@ -16,12 +16,59 @@
     HTTPStatus.is_server_error = property(lambda s: 500 <= s <= 599)  # noqa: PLR2004
 
 
+from dataclasses import dataclass
+
 import requests
 from loguru import logger as log
 
 from spectrumx import errors
 
 
+@dataclass(frozen=True, slots=True)
+class _SyntheticHTTPStatus:
+    """Fallback for non-standard HTTP codes that HTTPStatus rejects."""
+
+    code: int
+
+    @property
+    def is_informational(self) -> bool:
+        return 100 <= self.code <= 199  # noqa: PLR2004
+
+    @property
+    def is_success(self) -> bool:
+        return 200 <= self.code <= 299  # noqa: PLR2004
+
+    @property
+    def is_redirection(self) -> bool:
+        return 300 <= self.code <= 399  # noqa: PLR2004
+
+    @property
+    def is_client_error(self) -> bool:
+        return 400 <= self.code <= 499  # noqa: PLR2004
+
+    @property
+    def is_server_error(self) -> bool:
+        return 500 <= self.code <= 599  # noqa: PLR2004
+
+    def __eq__(self, other: object) -> bool:
+        if isinstance(other, _SyntheticHTTPStatus):
+            return self.code == other.code
+        if isinstance(other, HTTPStatus):
+            return self.code == other.value
+        return NotImplemented
+
+    def __hash__(self) -> int:
+        return hash(self.code)
+
+
+def _safe_http_status(code: int) -> HTTPStatus | _SyntheticHTTPStatus:
+    """Return an HTTPStatus-like object, handling non-standard codes gracefully."""
+    try:
+        return HTTPStatus(code)
+    except ValueError:
+        return _SyntheticHTTPStatus(code=code)
+
+
 def success_or_raise(
     response: requests.Response,
     ContextException: type[errors.SDSError] = errors.SDSError,  # noqa: N803
@@ -30,7 +77,7 @@ def success_or_raise(
     code = response.status_code
     if code is None:
         raise errors.SDSError(message="No status code in response.")
-    status = HTTPStatus(code)
+    status = _safe_http_status(code)
     if status.is_success:
         return
 

sdk/tests/ops/test_network.py

@@ -126,6 +126,24 @@ def test_success_or_raise_catchall_fallback() -> None:
         success_or_raise(response)
 
 
+def test_success_or_raise_non_standard_code_499() -> None:
+    """Non-standard 499 should be treated as client error, not crash."""
+    response = requests.Response()
+    response.status_code = 499
+    response._content = b'{"detail": "Client closed request"}'
+    with pytest.raises(errors.SDSError, match="Client closed request"):
+        success_or_raise(response)
+
+
+def test_success_or_raise_non_standard_code_599() -> None:
+    """Non-standard 599 should be treated as server error, not crash."""
+    response = requests.Response()
+    response.status_code = 599
+    response._content = b'{"detail": "Network connect timeout"}'
+    with pytest.raises(errors.ServiceError, match="Network connect timeout"):
+        success_or_raise(response, ContextException=errors.SDSError)
+
+
 def test_extract_error_details_from_html_no_matching_element() -> None:
     """When HTML lacks #summary/#pastebinTraceback, falls back to reason."""
     response = requests.Response()