81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""Unit tests for the in-memory per-session classification cache (session_cache.py).
|
|
|
|
Offline, no network. Drive the pure get/put/TTL logic directly. time.time is
|
|
monkeypatched so expiry tests are deterministic (no wall-clock dependence).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
import session_cache
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clean_cache():
|
|
session_cache.clear()
|
|
yield
|
|
session_cache.clear()
|
|
|
|
|
|
def _freeze(monkeypatch, t: float):
|
|
"""Freeze session_cache.time.time at ``t`` seconds."""
|
|
monkeypatch.setattr(session_cache.time, "time", lambda: t)
|
|
|
|
|
|
def test_get_miss_returns_none():
|
|
assert session_cache.get("sess-none", staleness_seconds=60) is None
|
|
|
|
|
|
def test_get_hit_returns_fresh_entry(monkeypatch):
|
|
_freeze(monkeypatch, 1000.0)
|
|
session_cache.put("sess-1", "coding_general", 2)
|
|
got = session_cache.get("sess-1", staleness_seconds=60)
|
|
assert got is not None
|
|
assert got.task_category == "coding_general"
|
|
assert got.task_tier == 2
|
|
assert got.cached_at == 1000.0
|
|
|
|
|
|
def test_get_expired_returns_none(monkeypatch):
|
|
_freeze(monkeypatch, 1000.0)
|
|
session_cache.put("sess-2", "debugging", 3)
|
|
# Advance past the 60s staleness window.
|
|
_freeze(monkeypatch, 1000.0 + 60.0 + 0.001)
|
|
assert session_cache.get("sess-2", staleness_seconds=60) is None
|
|
|
|
|
|
def test_get_on_staleness_boundary_is_fresh(monkeypatch):
|
|
_freeze(monkeypatch, 1000.0)
|
|
session_cache.put("sess-3", "coding_refactor", 1)
|
|
# Exactly 60s later: not strictly greater than staleness_seconds, so fresh.
|
|
_freeze(monkeypatch, 1000.0 + 60.0)
|
|
got = session_cache.get("sess-3", staleness_seconds=60)
|
|
assert got is not None
|
|
assert got.task_category == "coding_refactor"
|
|
|
|
|
|
def test_put_overwrites_existing_entry(monkeypatch):
|
|
_freeze(monkeypatch, 1000.0)
|
|
session_cache.put("sess-4", "coding_general", 1)
|
|
_freeze(monkeypatch, 1100.0)
|
|
session_cache.put("sess-4", "summarization", 2)
|
|
got = session_cache.get("sess-4", staleness_seconds=60)
|
|
assert got is not None
|
|
assert got.task_category == "summarization"
|
|
assert got.task_tier == 2
|
|
assert got.cached_at == 1100.0
|
|
|
|
|
|
def test_get_staleness_is_per_session(monkeypatch):
|
|
_freeze(monkeypatch, 1000.0)
|
|
session_cache.put("sess-a", "coding_general", 2)
|
|
session_cache.put("sess-b", "debugging", 3)
|
|
_freeze(monkeypatch, 1000.0 + 61.0)
|
|
# sess-a is now stale, sess-b was written at the same time so also stale;
|
|
# write sess-b fresh to isolate per-session expiry.
|
|
_freeze(monkeypatch, 1000.0 + 62.0)
|
|
session_cache.put("sess-b", "debugging", 3)
|
|
assert session_cache.get("sess-a", staleness_seconds=60) is None
|
|
assert session_cache.get("sess-b", staleness_seconds=60) is not None
|