hermes-webui/tests/test_ttl_cache.py

"""
Tests for the TTL cache in api/config.py — get_available_models().

Validates:
  - Cache hit within TTL window
  - TTL expiry triggers re-scan
  - Config mtime change invalidates cache before TTL check
  - copy.deepcopy() isolation (mutating returned dict doesn't pollute cache)
  - invalidate_models_cache() direct invalidation
"""
import time
from unittest.mock import patch

import api.config as config


def _reset_cache():
    """Reset TTL cache globals to a clean state."""
    config._available_models_cache = None
    config._available_models_cache_ts = 0.0


# ── 1. test_cache_hit_within_ttl ──────────────────────────────────────────

def test_cache_hit_within_ttl():
    """Call get_available_models() twice within the TTL window.
    The second call should return cached data without re-scanning providers.
    We verify this by patching reload_config (called when cache is cold)
    and asserting it is only invoked once.
    """
    _reset_cache()
    original_reload = config.reload_config

    call_count = 0

    def _counting_reload():
        nonlocal call_count
        call_count += 1
        return original_reload()

    with patch.object(config, "reload_config", wraps=original_reload, side_effect=_counting_reload):
        saved_mtime = config._cfg_mtime
        try:
            # Force mtime mismatch so the first call triggers reload_config + cache fill
            config._cfg_mtime = 0.0
            result1 = config.get_available_models()
            first_call_count = call_count

            # Sync _cfg_mtime to the actual file so the second call doesn't
            # re-trigger reload_config via mtime mismatch — we want it to hit the TTL cache.
            try:
                config._cfg_mtime = config.Path(config._get_config_path()).stat().st_mtime
            except OSError:
                config._cfg_mtime = 0.0

            result2 = config.get_available_models()

            # Both results should have the same structure
            assert "groups" in result1
            assert "groups" in result2

            # reload_config should not have been called again for the second invocation
            # (the TTL cache served it)
            assert call_count == first_call_count, (
                f"Expected no extra reload_config calls, but got "
                f"{call_count - first_call_count} extra"
            )
        finally:
            config._cfg_mtime = saved_mtime
    _reset_cache()


# ── 2. test_ttl_expiry ───────────────────────────────────────────────────

def test_ttl_expiry():
    """Populate the cache, then advance time.monotonic() past 60s.
    The next call should re-scan (not serve from cache).
    """
    _reset_cache()

    # Ensure _cfg_mtime matches file so mtime check doesn't invalidate
    try:
        config._cfg_mtime = config.Path(config._get_config_path()).stat().st_mtime
    except OSError:
        config._cfg_mtime = 0.0

    # First call populates cache
    result1 = config.get_available_models()
    assert config._available_models_cache is not None, "Cache should be populated"

    # Record the cache timestamp
    cache_ts = config._available_models_cache_ts

    # Advance time.monotonic() by more than the TTL
    original_monotonic = time.monotonic
    offset = config._AVAILABLE_MODELS_CACHE_TTL + 10.0  # 70s past the real monotonic

    with patch.object(time, "monotonic", side_effect=lambda: original_monotonic() + offset):
        result2 = config.get_available_models()

    # The cache should have been refreshed — the timestamp must be newer
    assert config._available_models_cache_ts > cache_ts, (
        "Cache should have been refreshed after TTL expiry"
    )

    _reset_cache()


# ── 3. test_mtime_invalidation ───────────────────────────────────────────

def test_mtime_invalidation():
    """Populate the cache, then change _cfg_mtime to simulate a config file
    change on disk. The next call should invalidate the cache and re-scan.

    This test is fully self-contained: it invalidates both memory and disk
    caches first, then populates the cache via a priming call, then verifies
    that an mtime mismatch triggers a cache rebuild. It does not depend on
    any sibling test's execution order.
    """
    # Fully invalidate (memory + disk) so we start from a known clean state.
    config.invalidate_models_cache()

    # Force the synchronous (unbounded) rebuild path so the priming call is
    # guaranteed to have published the cache when it returns. With the
    # default bounded budget (_LIVE_REBUILD_BUDGET_SECONDS=4.0), a cold
    # rebuild on a slow/loaded box can exceed the budget and return a
    # static fallback WITHOUT populating _available_models_cache — which
    # made this test fail when run in isolation (it previously only passed
    # because a sibling test had already warmed the cache in-process).
    original_budget = config._LIVE_REBUILD_BUDGET_SECONDS
    config._LIVE_REBUILD_BUDGET_SECONDS = 0.0
    real_mtime = 0.0

    try:
        # Ensure _cfg_mtime matches file so first call doesn't re-scan due to mtime
        try:
            real_mtime = config.Path(config._get_config_path()).stat().st_mtime
        except OSError:
            real_mtime = 0.0
        config._cfg_mtime = real_mtime

        # Priming call: populate the cache (both memory and disk)
        result1 = config.get_available_models()
        assert config._available_models_cache is not None, (
            "Priming call should populate the in-memory cache"
        )
        assert config._available_models_cache_ts > 0.0, (
            "Priming call should set a valid cache timestamp"
        )

        # Simulate config.yaml changed on disk by setting _cfg_mtime to 0
        # (which won't match the actual file mtime)
        config._cfg_mtime = 0.0

        # The next call should detect mtime mismatch, reload, and invalidate cache
        result2 = config.get_available_models()

        # Cache must have been refreshed — timestamp advanced since we reset it
        # to 0.0 on invalidation.
        assert config._available_models_cache_ts > 0.0, (
            "Cache timestamp should be updated after invalidation + rebuild"
        )
        # Both calls must return a valid /api/models payload.
        assert "groups" in result1 and "groups" in result2
    finally:
        config._LIVE_REBUILD_BUDGET_SECONDS = original_budget
        config._cfg_mtime = real_mtime
        config.invalidate_models_cache()


# ── 4. test_deepcopy_isolation ────────────────────────────────────────────

def test_deepcopy_isolation():
    """Mutating the returned dict from get_available_models() must not
    affect the cache or subsequent return values.
    """
    _reset_cache()

    # Ensure _cfg_mtime matches file so mtime check doesn't invalidate
    try:
        config._cfg_mtime = config.Path(config._get_config_path()).stat().st_mtime
    except OSError:
        config._cfg_mtime = 0.0

    # First call populates cache
    result1 = config.get_available_models()

    # Mutate the returned dict
    if result1["groups"]:
        result1["groups"][0]["models"].clear()
    result1["groups"].append({"provider": "FAKE", "models": [{"id": "fake-model"}]})
    result1["active_provider"] = "HACKED"

    # Second call should return an unmutated copy
    result2 = config.get_available_models()

    # The mutated keys must not appear in the second result
    assert result2["active_provider"] != "HACKED", "Mutation leaked into cache"
    assert not any(
        g.get("provider") == "FAKE" for g in result2["groups"]
    ), "Fake provider leaked into cache"

    # If there were groups originally, the first group's models should not be empty
    # (unless it genuinely had no models, which is unlikely)
    if result1["groups"] and result2["groups"]:
        # result1["groups"][0]["models"] was cleared, but result2 should be intact
        assert len(result2["groups"][0].get("models", [])) > 0, (
            "Mutation of result1 cleared models in result2 — deepcopy failed"
        )

    _reset_cache()


# ── 5. test_invalidate_models_cache_direct ───────────────────────────────

def test_invalidate_models_cache_direct():
    """Call invalidate_models_cache() after populating the cache.
    _AVAILABLE_MODELS_CACHE should be None and the next call should re-scan.
    """
    _reset_cache()

    # Ensure _cfg_mtime matches file so mtime check doesn't invalidate
    try:
        config._cfg_mtime = config.Path(config._get_config_path()).stat().st_mtime
    except OSError:
        config._cfg_mtime = 0.0

    # First call populates cache
    result1 = config.get_available_models()
    assert config._available_models_cache is not None, "Cache should be populated"
    first_ts = config._available_models_cache_ts

    # Directly invalidate
    config.invalidate_models_cache()

    # Cache must be cleared
    assert config._available_models_cache is None, (
        "invalidate_models_cache() should set _AVAILABLE_MODELS_CACHE to None"
    )

    # Next call should re-scan and produce a fresh cache
    result2 = config.get_available_models()
    assert config._available_models_cache is not None, "Cache should be re-populated"
    assert config._available_models_cache_ts >= first_ts, (
        "Cache timestamp should be updated after re-scan"
    )

    _reset_cache()