LatteandClaude Opus 4.8
0f7a861fc4
EsiClient now caches GET responses honoring the Expires header and backs off when the rolling ESI error window is nearly exhausted (X-Esi-Error-Limit-*). README gains Setup / Connect-to-Claude / enabling-writes sections. Closes #16. SDE bulk-load split out to its own follow-up (#51). 135 tests, coverage 90%. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
105 lines
2.8 KiB
Python
105 lines
2.8 KiB
Python
"""ESI client: error handling, empty bodies, datasource param."""
|
|
|
|
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from familiar import esi
|
|
|
|
|
|
_FUTURE = "Wed, 21 Oct 2099 07:28:00 GMT"
|
|
|
|
|
|
def test_get_caches_when_expires_present(mock_esi):
|
|
calls = {"n": 0}
|
|
|
|
def handler(req):
|
|
calls["n"] += 1
|
|
return httpx.Response(200, json={"v": 1}, headers={"Expires": _FUTURE})
|
|
|
|
mock_esi(handler)
|
|
first = asyncio.run(esi.client().get("/x/"))
|
|
second = asyncio.run(esi.client().get("/x/"))
|
|
assert first == second == {"v": 1}
|
|
assert calls["n"] == 1 # second served from cache
|
|
|
|
|
|
def test_no_cache_without_expires(mock_esi):
|
|
calls = {"n": 0}
|
|
|
|
def handler(req):
|
|
calls["n"] += 1
|
|
return httpx.Response(200, json={"v": 1})
|
|
|
|
mock_esi(handler)
|
|
asyncio.run(esi.client().get("/y/"))
|
|
asyncio.run(esi.client().get("/y/"))
|
|
assert calls["n"] == 2
|
|
|
|
|
|
def test_error_limit_triggers_backoff(mock_esi, monkeypatch):
|
|
slept = []
|
|
|
|
async def fake_sleep(s):
|
|
slept.append(s)
|
|
|
|
monkeypatch.setattr(esi, "_sleep", fake_sleep)
|
|
mock_esi(
|
|
lambda req: httpx.Response(
|
|
200, json={}, headers={"X-Esi-Error-Limit-Remain": "1", "X-Esi-Error-Limit-Reset": "3"}
|
|
)
|
|
)
|
|
c = esi.client()
|
|
asyncio.run(c.get("/a/")) # records remain=1
|
|
asyncio.run(c.get("/b/")) # next request backs off before firing
|
|
assert slept and slept[0] > 0
|
|
|
|
|
|
def test_get_raises_esi_error_on_http_error(mock_esi):
|
|
mock_esi(lambda req: httpx.Response(404, json={"error": "not found"}))
|
|
with pytest.raises(esi.EsiError) as exc:
|
|
asyncio.run(esi.client().get("/universe/types/999999/"))
|
|
assert exc.value.status == 404
|
|
assert "404" in str(exc.value)
|
|
|
|
|
|
def test_get_raises_on_network_error(mock_esi):
|
|
def boom(req):
|
|
raise httpx.ConnectError("boom")
|
|
|
|
mock_esi(boom)
|
|
with pytest.raises(esi.EsiError):
|
|
asyncio.run(esi.client().get("/anything/"))
|
|
|
|
|
|
def test_empty_body_returns_none(mock_esi):
|
|
mock_esi(lambda req: httpx.Response(200, content=b""))
|
|
assert asyncio.run(esi.client().get("/x/")) is None
|
|
|
|
|
|
def test_datasource_param_is_attached(mock_esi):
|
|
seen = {}
|
|
|
|
def handler(req):
|
|
seen["datasource"] = req.url.params.get("datasource")
|
|
return httpx.Response(200, json={"ok": True})
|
|
|
|
mock_esi(handler)
|
|
asyncio.run(esi.client().get("/x/"))
|
|
assert seen["datasource"] == "tranquility"
|
|
|
|
|
|
def test_error_detail_falls_back_to_text(mock_esi):
|
|
mock_esi(lambda req: httpx.Response(500, content=b"upstream exploded"))
|
|
with pytest.raises(esi.EsiError) as exc:
|
|
asyncio.run(esi.client().get("/x/"))
|
|
assert "upstream exploded" in str(exc.value)
|
|
|
|
|
|
def test_default_client_is_singleton():
|
|
esi.set_client(None)
|
|
first = esi.client()
|
|
assert esi.client() is first
|
|
esi.set_client(None)
|