tests/test_inmemory_adata_cache.py

"""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