File size: 2,269 Bytes
3f2f9aa c3b49d6 3f2f9aa c3b49d6 3f2f9aa c3b49d6 3f2f9aa | 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 | """Tests for the process-lifetime in-memory AnnData cache (src/cache.py).
These guard the two correctness properties the persistent HTTP server relies on:
1. Each read returns an independent copy, so a caller mutating (or writing
back) the result cannot corrupt the shared, read-mostly cache — this is
what makes one resident server safe to share across concurrent sessions.
2. A rewritten file (newer mtime/size) is transparently re-read instead of
serving a stale parse.
"""
import time
import anndata as ad
import numpy as np
import pytest
from src import cache
@pytest.fixture(autouse=True)
def _clear_cache():
cache._MEM_CACHE.clear()
yield
cache._MEM_CACHE.clear()
def _write(path, fill):
a = ad.AnnData(X=np.full((10, 5), float(fill)))
a.write_h5ad(path)
return a
def test_second_read_hits_cache_single_entry(tmp_path):
p = tmp_path / "x.h5ad"
_write(p, 1)
cache.read_h5ad_cached(str(p))
cache.read_h5ad_cached(str(p))
assert len(cache._MEM_CACHE) == 1 # one entry, reused
def test_returns_independent_copies(tmp_path):
p = tmp_path / "x.h5ad"
_write(p, 1)
a = cache.read_h5ad_cached(str(p))
b = cache.read_h5ad_cached(str(p))
assert a is not b
# Mutating one copy must not affect the cache or other copies.
a.X[:] = 999
c = cache.read_h5ad_cached(str(p))
assert float(c.X.mean()) == 1.0
def test_rewrite_invalidates_stale_entry(tmp_path):
p = tmp_path / "x.h5ad"
_write(p, 1)
assert float(cache.read_h5ad_cached(str(p)).X.mean()) == 1.0
time.sleep(1.1) # ensure mtime advances at 1s resolution
_write(p, 7)
assert float(cache.read_h5ad_cached(str(p)).X.mean()) == 7.0
# Stale entry for the same path is evicted, not accumulated.
assert len(cache._MEM_CACHE) == 1
def test_missing_path_falls_through(tmp_path):
with pytest.raises(Exception): # noqa: B017 -- asserts any failure for a missing path
cache.read_h5ad_cached(str(tmp_path / "nope.h5ad"))
def test_eviction_bounds_entries(tmp_path):
for i in range(cache._MEM_MAX_ENTRIES + 3):
p = tmp_path / f"d{i}.h5ad"
_write(p, i)
cache.read_h5ad_cached(str(p))
assert len(cache._MEM_CACHE) <= cache._MEM_MAX_ENTRIES
|