68 lines
2.6 KiB
Python
68 lines
2.6 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from app.platforms.base import PlatformAPIError, TikHubClient
|
|
|
|
|
|
class SequenceTransport:
|
|
def __init__(self, responses):
|
|
self.responses = list(responses)
|
|
self.requests = []
|
|
|
|
def __call__(self, request: httpx.Request) -> httpx.Response:
|
|
self.requests.append(request)
|
|
response = self.responses.pop(0)
|
|
response.request = request
|
|
return response
|
|
|
|
|
|
def test_tikhub_client_retries_429_with_exponential_backoff(monkeypatch):
|
|
sleeps = []
|
|
monkeypatch.setattr("app.platforms.base.time.sleep", sleeps.append)
|
|
transport = SequenceTransport(
|
|
[
|
|
httpx.Response(429, json={"message": "Too Many Requests"}),
|
|
httpx.Response(429, json={"message": "Too Many Requests"}),
|
|
httpx.Response(200, json={"ok": True}),
|
|
]
|
|
)
|
|
http_client = httpx.Client(transport=httpx.MockTransport(transport))
|
|
client = TikHubClient(base_url="https://api.test", api_key="secret-token", http_client=http_client)
|
|
|
|
result = client.get("/demo")
|
|
|
|
assert result == {"ok": True}
|
|
assert sleeps == [1, 2]
|
|
assert len(transport.requests) == 3
|
|
assert transport.requests[0].headers["Authorization"] == "Bearer secret-token"
|
|
|
|
|
|
def test_tikhub_client_raises_structured_error_after_retries(monkeypatch):
|
|
sleeps = []
|
|
monkeypatch.setattr("app.platforms.base.time.sleep", sleeps.append)
|
|
transport = SequenceTransport([httpx.Response(429, json={"message": "Too Many Requests"}) for _ in range(4)])
|
|
http_client = httpx.Client(transport=httpx.MockTransport(transport))
|
|
client = TikHubClient(base_url="https://api.test", api_key="secret-token", http_client=http_client)
|
|
|
|
with pytest.raises(PlatformAPIError) as exc_info:
|
|
client.get("/demo")
|
|
|
|
assert exc_info.value.error_type == "rate_limited"
|
|
assert exc_info.value.status_code == 429
|
|
assert "secret-token" not in str(exc_info.value)
|
|
assert sleeps == [1, 2, 4]
|
|
|
|
|
|
def test_tikhub_client_reports_401_as_auth_error_without_leaking_token():
|
|
transport = SequenceTransport([httpx.Response(401, json={"message": "Unauthorized"})])
|
|
http_client = httpx.Client(transport=httpx.MockTransport(transport))
|
|
client = TikHubClient(base_url="https://api.test", api_key="secret-token", http_client=http_client)
|
|
|
|
with pytest.raises(PlatformAPIError) as exc_info:
|
|
client.get("/demo")
|
|
|
|
assert exc_info.value.error_type == "auth_error"
|
|
assert exc_info.value.status_code == 401
|
|
assert str(exc_info.value) == "TikHub 鉴权失败，请检查 TIKHUB_API_KEY 是否有效"
|
|
assert "secret-token" not in str(exc_info.value)
|