| 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) |
| 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") |
| |
| |
| 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 |
|
|
|
|
| def test_cached_get_retries_on_429_then_succeeds(tmp_path, monkeypatch): |
| monkeypatch.setattr(time, "sleep", lambda s: None) |
| 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 |
|
|
|
|
| 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} |
|
|
