92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
"""Unit tests for api_client.py's own plumbing (cache eviction, single-flight)
|
|
-- as opposed to test_content.py, which exercises it indirectly through the
|
|
rendered pages via the `client` fixture's monkeypatched api_client functions.
|
|
"""
|
|
import threading
|
|
import time
|
|
import types
|
|
|
|
import api_client
|
|
|
|
|
|
def setup_function(_):
|
|
api_client._cache.clear()
|
|
api_client._inflight.clear()
|
|
|
|
|
|
def test_cache_eviction_is_lru(monkeypatch):
|
|
monkeypatch.setattr(api_client, "_CACHE_MAX_ENTRIES", 3)
|
|
for i in range(3):
|
|
api_client._cache_put(f"k{i}", i)
|
|
# Touch k0 so it's most-recently-used; k1 is now the least-recently-used
|
|
# entry and should be the one evicted when the cap is exceeded.
|
|
api_client._cache_get("k0")
|
|
api_client._cache_put("k3", 3)
|
|
|
|
assert api_client._cache_get("k1") is None
|
|
assert api_client._cache_get("k0") == 0
|
|
assert api_client._cache_get("k2") == 2
|
|
assert api_client._cache_get("k3") == 3
|
|
assert len(api_client._cache) == 3
|
|
|
|
|
|
def test_cache_entries_expire_after_ttl(monkeypatch):
|
|
monkeypatch.setattr(api_client, "_TTL", 0.01)
|
|
api_client._cache_put("k", "v")
|
|
assert api_client._cache_get("k") == "v"
|
|
time.sleep(0.02)
|
|
assert api_client._cache_get("k") is None
|
|
|
|
|
|
def test_single_flight_dedupes_concurrent_misses(monkeypatch):
|
|
"""N concurrent misses on the same key must reach the backend once, not
|
|
N times -- every caller gets the same result either way."""
|
|
calls = []
|
|
release = threading.Event()
|
|
|
|
class _FakeResp:
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
def json(self):
|
|
return {"ok": True}
|
|
|
|
def fake_get(path, headers=None):
|
|
calls.append(path)
|
|
release.wait(timeout=2)
|
|
return _FakeResp()
|
|
|
|
monkeypatch.setattr(api_client, "_client", types.SimpleNamespace(get=fake_get))
|
|
|
|
results = []
|
|
results_lock = threading.Lock()
|
|
|
|
def worker():
|
|
r = api_client._get("/thing")
|
|
with results_lock:
|
|
results.append(r)
|
|
|
|
threads = [threading.Thread(target=worker) for _ in range(5)]
|
|
for t in threads:
|
|
t.start()
|
|
time.sleep(0.1) # let every thread reach (and block behind) the first fetch
|
|
release.set()
|
|
for t in threads:
|
|
t.join(timeout=2)
|
|
|
|
assert len(calls) == 1 # exactly one backend call for 5 concurrent misses
|
|
assert results == [{"ok": True}] * 5
|
|
assert api_client._cache_get("/thing") == {"ok": True}
|
|
|
|
|
|
def test_single_flight_clears_inflight_entry_after_completion(monkeypatch):
|
|
class _FakeResp:
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
def json(self):
|
|
return {"n": 1}
|
|
|
|
monkeypatch.setattr(api_client, "_client", types.SimpleNamespace(get=lambda path, headers=None: _FakeResp()))
|
|
api_client._get("/thing")
|
|
assert "/thing" not in api_client._inflight
|