File size: 3,614 Bytes
5bfc214 | 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 | import json
import time
from src.data.http_cache import cached_get
class FakeResponse:
def __init__(self, status_code, json_data=None, headers=None):
self.status_code = status_code
self._json_data = json_data
self.headers = headers or {}
def json(self):
return self._json_data
class FakeSession:
def __init__(self, responses):
self.responses = list(responses) # list of FakeResponse, consumed in order
self.calls = 0
def get(self, url, params=None, timeout=None):
self.calls += 1
return self.responses.pop(0)
def test_cached_get_returns_data_on_success(tmp_path):
session = FakeSession([FakeResponse(200, {"hello": "world"})])
result = cached_get(session, "http://x", tmp_path, "key1")
assert result == {"hello": "world"}
assert session.calls == 1
def test_cached_get_uses_disk_cache_on_second_call(tmp_path):
session = FakeSession([FakeResponse(200, {"a": 1})])
r1 = cached_get(session, "http://x", tmp_path, "key1")
# second call: session has no more queued responses, so if it tries to
# hit the network again it would IndexError on an empty list.
r2 = cached_get(session, "http://x", tmp_path, "key1")
assert r1 == r2 == {"a": 1}
assert session.calls == 1
def test_cached_get_returns_none_on_404_without_retry(tmp_path):
session = FakeSession([FakeResponse(404)])
result = cached_get(session, "http://x", tmp_path, "key1")
assert result is None
assert session.calls == 1 # no retries for a real 404
def test_cached_get_retries_on_429_then_succeeds(tmp_path, monkeypatch):
monkeypatch.setattr(time, "sleep", lambda s: None) # don't actually wait in tests
session = FakeSession([
FakeResponse(429, headers={"Retry-After": "1"}),
FakeResponse(200, {"ok": True}),
])
result = cached_get(session, "http://x", tmp_path, "key1", max_retries=3)
assert result == {"ok": True}
assert session.calls == 2
def test_cached_get_respects_github_rate_limit_reset_header(tmp_path, monkeypatch):
slept = []
monkeypatch.setattr(time, "sleep", lambda s: slept.append(s))
monkeypatch.setattr(time, "time", lambda: 1000.0)
session = FakeSession([
FakeResponse(403, headers={"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": "1010"}),
FakeResponse(200, {"ok": True}),
])
result = cached_get(session, "http://x", tmp_path, "key1", max_retries=3)
assert result == {"ok": True}
assert slept and slept[0] == 11 # (1010 - 1000) + 1
def test_cached_get_gives_up_after_max_retries(tmp_path, monkeypatch):
monkeypatch.setattr(time, "sleep", lambda s: None)
session = FakeSession([FakeResponse(500) for _ in range(5)])
result = cached_get(session, "http://x", tmp_path, "key1", max_retries=5)
assert result is None
assert session.calls == 5
def test_cached_get_different_keys_do_not_collide(tmp_path):
session = FakeSession([FakeResponse(200, {"a": 1}), FakeResponse(200, {"b": 2})])
r1 = cached_get(session, "http://x", tmp_path, "key1")
r2 = cached_get(session, "http://x", tmp_path, "key2")
assert r1 == {"a": 1}
assert r2 == {"b": 2}
def test_cached_get_corrupt_cache_file_is_refetched(tmp_path):
session = FakeSession([FakeResponse(200, {"fresh": True})])
from src.data.http_cache import _cache_path
path = _cache_path(tmp_path, "key1")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("not valid json{{{")
result = cached_get(session, "http://x", tmp_path, "key1")
assert result == {"fresh": True}
|