tests/test_cache.py

"""Tests for backend/cache.py — TTLCache and module-level singletons."""
import time
import threading
import pytest

from backend.cache import TTLCache, plans_cache, stats_cache


# ── Basic get/set ─────────────────────────────────────────────────────────────

def test_set_and_get_basic():
    cache = TTLCache(ttl=60)
    cache.set("key1", "value1")
    assert cache.get("key1") == "value1"


def test_get_missing_key_returns_none():
    cache = TTLCache(ttl=60)
    assert cache.get("nonexistent") is None


def test_set_various_value_types():
    cache = TTLCache(ttl=60)
    cache.set("int_key", 42)
    cache.set("dict_key", {"a": 1, "b": [1, 2, 3]})
    cache.set("list_key", [1, 2, 3])
    assert cache.get("int_key") == 42
    assert cache.get("dict_key") == {"a": 1, "b": [1, 2, 3]}
    assert cache.get("list_key") == [1, 2, 3]


# ── TTL expiry ────────────────────────────────────────────────────────────────

def test_expired_entry_returns_none():
    cache = TTLCache(ttl=1)
    cache.set("expiring", "will_expire", ttl=1)
    time.sleep(1.1)
    assert cache.get("expiring") is None


def test_ttl_not_yet_expired():
    cache = TTLCache(ttl=60)
    cache.set("fresh", "still_alive", ttl=60)
    assert cache.get("fresh") == "still_alive"


def test_per_entry_ttl_overrides_default():
    cache = TTLCache(ttl=60)
    # Use a very short TTL for this specific entry
    cache.set("short_lived", "gone_soon", ttl=1)
    assert cache.get("short_lived") == "gone_soon"
    time.sleep(1.1)
    assert cache.get("short_lived") is None


def test_default_ttl_is_used_when_not_specified():
    """Setting an entry without explicit TTL uses the cache's default TTL."""
    cache = TTLCache(ttl=300)
    cache.set("default_ttl_key", "hello")
    # Should be present because 300s haven't passed
    assert cache.get("default_ttl_key") == "hello"


# ── delete() ──────────────────────────────────────────────────────────────────

def test_delete_removes_entry():
    cache = TTLCache(ttl=60)
    cache.set("to_delete", "present")
    assert cache.get("to_delete") == "present"
    cache.delete("to_delete")
    assert cache.get("to_delete") is None


def test_delete_nonexistent_key_does_not_raise():
    cache = TTLCache(ttl=60)
    cache.delete("ghost_key")  # should not raise


# ── clear() ───────────────────────────────────────────────────────────────────

def test_clear_empties_cache():
    cache = TTLCache(ttl=60)
    cache.set("a", 1)
    cache.set("b", 2)
    cache.set("c", 3)
    cache.clear()
    assert cache.get("a") is None
    assert cache.get("b") is None
    assert cache.get("c") is None


def test_clear_on_empty_cache_does_not_raise():
    cache = TTLCache(ttl=60)
    cache.clear()  # should not raise


# ── size() ────────────────────────────────────────────────────────────────────

def test_size_returns_correct_count():
    cache = TTLCache(ttl=60)
    assert cache.size() == 0
    cache.set("x", 1)
    assert cache.size() == 1
    cache.set("y", 2)
    assert cache.size() == 2
    cache.delete("x")
    assert cache.size() == 1
    cache.clear()
    assert cache.size() == 0


def test_size_does_not_count_expired_entries():
    """Expired entries that have been accessed are removed, shrinking size."""
    cache = TTLCache(ttl=1)
    cache.set("expires", "soon", ttl=1)
    assert cache.size() == 1
    time.sleep(1.1)
    # Access triggers eviction
    cache.get("expires")
    assert cache.size() == 0


# ── Thread safety ─────────────────────────────────────────────────────────────

def test_concurrent_reads_writes_do_not_crash():
    """Multiple threads reading and writing should not raise or corrupt data."""
    cache = TTLCache(ttl=60)
    errors = []

    def writer(n):
        try:
            for i in range(50):
                cache.set(f"key_{n}_{i}", f"val_{n}_{i}")
        except Exception as e:
            errors.append(e)

    def reader(n):
        try:
            for i in range(50):
                cache.get(f"key_{n}_{i}")
        except Exception as e:
            errors.append(e)

    threads = []
    for t in range(5):
        threads.append(threading.Thread(target=writer, args=(t,)))
        threads.append(threading.Thread(target=reader, args=(t,)))

    for th in threads:
        th.start()
    for th in threads:
        th.join()

    assert errors == [], f"Thread errors: {errors}"


# ── Module-level singletons ───────────────────────────────────────────────────

def test_plans_cache_is_ttlcache_instance():
    assert isinstance(plans_cache, TTLCache)


def test_stats_cache_is_ttlcache_instance():
    assert isinstance(stats_cache, TTLCache)


def test_plans_cache_ttl_is_600():
    assert plans_cache.ttl == 600


def test_stats_cache_ttl_is_120():
    assert stats_cache.ttl == 120


def test_plans_cache_and_stats_cache_are_different_objects():
    assert plans_cache is not stats_cache


def test_singletons_are_same_object_on_reimport():
    from backend.cache import plans_cache as pc2, stats_cache as sc2
    assert plans_cache is pc2
    assert stats_cache is sc2