Repository navigation
Expand file tree
/
Copy pathtest_errors.py
More file actions
137 lines (113 loc) · 4.35 KB
/
Copy pathtest_errors.py
File metadata and controls
137 lines (113 loc) · 4.35 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
"""Error-status mapping and bonus fields on rate-limit / insufficient-credits."""
from __future__ import annotations
import httpx
import pytest
import respx
from stemsplit_python import (
APIStatusError,
AuthenticationError,
BadRequestError,
InsufficientCreditsError,
InternalServerError,
NotFoundError,
PermissionDeniedError,
RateLimitError,
StemSplit,
)
from stemsplit_python._transport import Transport
def _make_client() -> StemSplit:
return StemSplit(api_key="sk_live_test", base_url="https://stemsplit.io/api/v1", max_retries=0)
@respx.mock
def test_400_bad_request_maps_to_BadRequestError() -> None:
respx.get("https://stemsplit.io/api/v1/balance").respond(
400, json={"error": {"code": "FILE_TOO_LARGE", "message": "File exceeds 50MB"}}
)
client = _make_client()
with pytest.raises(BadRequestError) as exc:
client.account.balance()
assert exc.value.status_code == 400
assert exc.value.code == "FILE_TOO_LARGE"
assert "File exceeds 50MB" in str(exc.value)
@respx.mock
def test_401_maps_to_AuthenticationError() -> None:
respx.get("https://stemsplit.io/api/v1/balance").respond(
401, json={"error": {"code": "MISSING_API_KEY", "message": "Missing API key."}}
)
with pytest.raises(AuthenticationError):
_make_client().account.balance()
@respx.mock
def test_402_carries_insufficient_credits_fields() -> None:
respx.post("https://stemsplit.io/api/v1/jobs").respond(
402,
json={
"error": {
"code": "INSUFFICIENT_CREDITS",
"message": "Not enough credits",
"requiredSeconds": 180,
"purchaseUrl": "https://stemsplit.io/pricing",
}
},
)
client = _make_client()
with pytest.raises(InsufficientCreditsError) as exc:
client.jobs.create(source_url="https://example.com/song.mp3")
assert exc.value.required_seconds == 180
assert exc.value.purchase_url == "https://stemsplit.io/pricing"
@respx.mock
def test_403_maps_to_PermissionDeniedError() -> None:
respx.get("https://stemsplit.io/api/v1/balance").respond(
403, json={"error": {"code": "API_KEY_REVOKED", "message": "Key was revoked"}}
)
with pytest.raises(PermissionDeniedError) as exc:
_make_client().account.balance()
assert exc.value.code == "API_KEY_REVOKED"
@respx.mock
def test_404_maps_to_NotFoundError() -> None:
respx.get("https://stemsplit.io/api/v1/jobs/missing").respond(
404, json={"error": {"code": "JOB_NOT_FOUND", "message": "Job not found"}}
)
with pytest.raises(NotFoundError):
_make_client().jobs.get("missing")
@respx.mock
def test_429_carries_rate_limit_headers() -> None:
respx.get("https://stemsplit.io/api/v1/balance").mock(
return_value=httpx.Response(
429,
headers={
"retry-after": "30",
"x-ratelimit-limit": "60",
"x-ratelimit-remaining": "0",
"x-ratelimit-reset": "1700000000",
},
json={"error": {"code": "RATE_LIMIT_EXCEEDED", "message": "Slow down"}},
)
)
with pytest.raises(RateLimitError) as exc:
_make_client().account.balance()
assert exc.value.retry_after == 30
assert exc.value.limit == 60
assert exc.value.remaining == 0
@respx.mock
def test_500_maps_to_InternalServerError() -> None:
respx.get("https://stemsplit.io/api/v1/balance").respond(
500, json={"error": {"code": "INTERNAL", "message": "boom"}}
)
with pytest.raises(InternalServerError):
_make_client().account.balance()
@respx.mock
def test_unknown_status_uses_APIStatusError() -> None:
respx.get("https://stemsplit.io/api/v1/balance").respond(
418, json={"error": {"code": "TEAPOT", "message": "I am a teapot"}}
)
with pytest.raises(APIStatusError) as exc:
_make_client().account.balance()
assert exc.value.status_code == 418
@respx.mock
def test_retries_exhausted_then_raises() -> None:
"""With max_retries=2 we should see 3 attempts total before giving up."""
route = respx.get("https://stemsplit.io/api/v1/balance").respond(503)
transport = Transport(api_key="sk_live_test", max_retries=2)
with pytest.raises(InternalServerError):
transport.request("GET", "/balance")
assert route.call_count == 3
transport.close()