# ------------------------------------------------------------------- # This source file is available under the terms of the # Pimcore Open Core License (POCL) # Full copyright and license information is available in # LICENSE.md which is distributed with this source code. # ------------------------------------------------------------------- """The cache has to forget. Callers name the model, so the set of models a running instance is asked for is open-ended; without a bound, every distinct id ever requested stays resident and the container eventually dies. Least-recently-used is the right eviction order here because the working set in practice is a couple of models used constantly plus a long tail used once. """ import logging import pytest from src.model_cache import LruCache def _cache(max_entries: int = 2) -> LruCache: return LruCache(max_entries=max_entries, name="test", logger=logging.getLogger("test")) def test_stores_and_returns_values(): cache = _cache() cache["a"] = 1 assert "a" in cache assert cache["a"] == 1 def test_reports_absent_keys(): cache = _cache() assert "missing" not in cache def test_evicts_the_least_recently_used_entry_when_full(): cache = _cache(max_entries=2) cache["a"] = 1 cache["b"] = 2 cache["c"] = 3 assert "a" not in cache, "the oldest entry should have been evicted" assert "b" in cache assert "c" in cache assert len(cache) == 2 def test_reading_an_entry_makes_it_recently_used(): """Otherwise the model being hammered by every request would be the one evicted.""" cache = _cache(max_entries=2) cache["a"] = 1 cache["b"] = 2 _ = cache["a"] # refreshes "a", making "b" the coldest cache["c"] = 3 assert "a" in cache assert "b" not in cache assert "c" in cache def test_overwriting_an_existing_key_does_not_evict(): cache = _cache(max_entries=2) cache["a"] = 1 cache["b"] = 2 cache["a"] = 99 assert len(cache) == 2 assert cache["a"] == 99 assert "b" in cache def test_capacity_is_at_least_one(): """A zero or negative limit would evict the entry a request just loaded, before it could be used, and turn every single request into a fresh load.""" cache = _cache(max_entries=0) cache["a"] = 1 assert cache["a"] == 1 assert len(cache) == 1 def test_eviction_is_logged(caplog): """Thrashing is invisible otherwise — the symptom is only slowness.""" cache = _cache(max_entries=1) with caplog.at_level(logging.INFO): cache["a"] = 1 cache["b"] = 2 assert any("a" in record.message for record in caplog.records) def test_evicted_entries_are_gone_not_merely_hidden(): """The point is to release the reference so the weights can be garbage collected.""" cache = _cache(max_entries=1) cache["a"] = object() cache["b"] = object() with pytest.raises(KeyError): _ = cache["a"]