153 lines
4.5 KiB
Python
153 lines
4.5 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from gaiia import (
|
|
GaiiaAPIError,
|
|
GaiiaClient,
|
|
GaiiaMutationError,
|
|
GaiiaRateLimitedError,
|
|
)
|
|
from gaiia.client import API_KEY_HEADER, TIMEZONE_HEADER
|
|
|
|
|
|
def _client(handler):
|
|
transport = httpx.MockTransport(handler)
|
|
return GaiiaClient(
|
|
api_key="gaiia_sk_sbox_test",
|
|
timezone="America/Chicago",
|
|
transport=transport,
|
|
)
|
|
|
|
|
|
def test_query_sends_expected_headers_and_body():
|
|
captured = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured["headers"] = dict(request.headers)
|
|
captured["body"] = request.read()
|
|
return httpx.Response(200, json={"data": {"x": 1}})
|
|
|
|
with _client(handler) as g:
|
|
out = g.query("{ x }", {"v": 1})
|
|
assert out == {"x": 1}
|
|
assert captured["headers"][API_KEY_HEADER.lower()] == "gaiia_sk_sbox_test"
|
|
assert captured["headers"][TIMEZONE_HEADER] == "America/Chicago"
|
|
import json
|
|
body = json.loads(captured["body"])
|
|
assert body["query"] == "{ x }"
|
|
assert body["variables"] == {"v": 1}
|
|
|
|
|
|
def test_top_level_errors_raise_api_error():
|
|
def handler(request):
|
|
return httpx.Response(
|
|
200,
|
|
json={"errors": [{"message": "bad", "extensions": {"code": "BAD_USER_INPUT"}}]},
|
|
)
|
|
|
|
with _client(handler) as g, pytest.raises(GaiiaAPIError) as exc:
|
|
g.query("{}")
|
|
assert exc.value.errors[0].code == "BAD_USER_INPUT"
|
|
|
|
|
|
def test_rate_limited_classifies_separately():
|
|
def handler(request):
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"errors": [
|
|
{
|
|
"message": "too many",
|
|
"extensions": {
|
|
"code": "RATE_LIMITED",
|
|
"limit": 500,
|
|
"used": 500,
|
|
"remaining": 0,
|
|
"retryAt": "2026-06-27T19:01:02.000Z",
|
|
},
|
|
}
|
|
]
|
|
},
|
|
headers={
|
|
"X-Rate-Limit-Allowed": "false",
|
|
"X-Rate-Limit-Limit": "500",
|
|
"X-Rate-Limit-Used": "500",
|
|
"X-Rate-Limit-Remaining": "0",
|
|
},
|
|
)
|
|
|
|
with _client(handler) as g, pytest.raises(GaiiaRateLimitedError) as exc:
|
|
g.query("{}")
|
|
assert exc.value.limit == 500
|
|
assert g.last_rate_limit is not None
|
|
assert g.last_rate_limit.remaining == 0
|
|
|
|
|
|
def test_mutate_raises_on_payload_errors():
|
|
def handler(request):
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"data": {
|
|
"createTicket": {
|
|
"ticket": None,
|
|
"errors": [{"code": "TITLE_TOO_LONG", "message": "long"}],
|
|
}
|
|
}
|
|
},
|
|
)
|
|
|
|
with _client(handler) as g, pytest.raises(GaiiaMutationError) as exc:
|
|
g.mutate("mutation { createTicket { ticket { id } errors { code message } } }", "createTicket")
|
|
assert exc.value.errors[0]["code"] == "TITLE_TOO_LONG"
|
|
|
|
|
|
def test_iter_nodes_paginates_via_cursor():
|
|
calls = []
|
|
|
|
def handler(request):
|
|
body = request.read().decode()
|
|
calls.append(body)
|
|
if '"after":null' in body.replace(" ", ""):
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"data": {
|
|
"things": {
|
|
"nodes": [{"id": 1}, {"id": 2}],
|
|
"pageInfo": {"hasNextPage": True, "endCursor": "c1"},
|
|
}
|
|
}
|
|
},
|
|
)
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"data": {
|
|
"things": {
|
|
"nodes": [{"id": 3}],
|
|
"pageInfo": {"hasNextPage": False, "endCursor": None},
|
|
}
|
|
}
|
|
},
|
|
)
|
|
|
|
with _client(handler) as g:
|
|
ids = [n["id"] for n in g.iter_nodes(
|
|
"query Q($after: String) { things(first: 2, after: $after) { nodes { id } pageInfo { hasNextPage endCursor } } }",
|
|
"things",
|
|
)]
|
|
assert ids == [1, 2, 3]
|
|
assert len(calls) == 2
|
|
|
|
|
|
def test_missing_api_key_raises():
|
|
import os
|
|
|
|
saved = os.environ.pop("GAIIA_KEY", None)
|
|
try:
|
|
with pytest.raises(ValueError):
|
|
GaiiaClient()
|
|
finally:
|
|
if saved is not None:
|
|
os.environ["GAIIA_KEY"] = saved
|