avoigt1121
Claude Opus 5.5
feat: governed ToolUniverse core for the knowledge specialist (s1)
377417d | """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 | |