File size: 2,914 Bytes
377417d | 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 | """C5 — the cache keeps retrieved_at honest and expires per source (ADR-0001 §8)."""
from __future__ import annotations
import pytest
from src import config
from src.cache import cache_key
from src.tu_client import SourceError, ToolClient
from tests.conftest import FakeRunner
OT_DRUGS = "OpenTargets_get_associated_drugs_by_target_ensemblID"
ARGS = {"ensemblId": "ENSG00000120889", "size": 100}
def _client(runner, memory_cache, clock):
return ToolClient(runner=runner, cache=memory_cache, clock=clock, tooluniverse_version="1.5.6")
def test_a_repeat_call_is_a_hit_with_the_original_retrieved_at(
memory_cache, clock, tnfrsf10b_response
):
runner = FakeRunner({OT_DRUGS: tnfrsf10b_response})
client = _client(runner, memory_cache, clock)
first = client.call(OT_DRUGS, ARGS)
clock.advance(3600)
second = client.call(OT_DRUGS, ARGS)
assert (first.cache, second.cache) == ("miss", "hit")
assert second.retrieved_at == first.retrieved_at
assert second.data == first.data
assert len(runner.calls) == 1
def test_an_entry_expires_at_the_sources_ttl(memory_cache, clock, tnfrsf10b_response):
runner = FakeRunner({OT_DRUGS: tnfrsf10b_response})
client = _client(runner, memory_cache, clock)
client.call(OT_DRUGS, ARGS)
clock.advance(config.source("opentargets")["ttl_seconds"] - 1)
assert client.call(OT_DRUGS, ARGS).cache == "hit"
clock.advance(2)
again = client.call(OT_DRUGS, ARGS)
assert again.cache == "miss"
assert again.retrieved_at == clock.now
assert len(runner.calls) == 2
def test_ttls_are_seven_days_and_one_day_for_trials():
assert config.source("clinicaltrials")["ttl_seconds"] == 86_400
for sid in ("opentargets", "chembl", "uniprot", "reactome", "hgnc"):
assert config.source(sid)["ttl_seconds"] == 7 * 86_400
def test_store_and_read_directly(memory_cache, clock):
memory_cache.put("chembl", "ChEMBL_get_drug_mechanisms", {"x": 1}, {"ok": True}, clock.now - 10)
data, retrieved_at = memory_cache.get("chembl", "ChEMBL_get_drug_mechanisms", {"x": 1})
assert data == {"ok": True}
assert retrieved_at == clock.now - 10
assert memory_cache.get("chembl", "ChEMBL_get_drug_mechanisms", {"x": 2}) is None
def test_the_key_ignores_argument_order():
assert cache_key("T", {"a": 1, "b": 2}) == cache_key("T", {"b": 2, "a": 1})
assert cache_key("T", {"a": 1}) != cache_key("U", {"a": 1})
def test_a_drifted_response_is_never_cached(memory_cache, clock):
drifted = {"status": "success", "data": {"target": {"id": "ENSG00000120889"}}}
runner = FakeRunner({OT_DRUGS: drifted})
client = _client(runner, memory_cache, clock)
for _ in range(2):
with pytest.raises(SourceError, match="upstream_schema_drift"):
client.call(OT_DRUGS, ARGS)
assert memory_cache.get("opentargets", OT_DRUGS, ARGS) is None
assert len(runner.calls) == 2
|