# -*- coding: utf-8 -*- """看板缓存与版本失效测试(离线): uv run python tests/test_cache.py""" import os import sys import tempfile import time from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.stdout.reconfigure(encoding="utf-8") from reaction_board.database import Database # noqa: E402 from reaction_board.repository import Repository # noqa: E402 from reaction_board.web.cache import TTLCache, build_cache_key # noqa: E402 def test_ttl_cache_expiry_and_params(): now = [1000.0] cache = TTLCache(clock=lambda: now[0], max_items=32) key = build_cache_key("db", "videos", 0, {"sort": "heat", "page": 1}) assert cache.get(key) is None cache.set(key, 10.0, {"value": 1}) now[0] += 5 assert cache.get(key) == {"value": 1} now[0] += 6 assert cache.get(key) is None same = build_cache_key("db", "videos", 0, {"page": 1, "sort": "heat"}) assert same == key other = build_cache_key("db", "videos", 1, {"page": 1, "sort": "heat"}) assert other != key def test_ttl_cache_prunes_to_max_items(): now = [0.0] cache = TTLCache(clock=lambda: now[0], max_items=16) for index in range(40): cache.set(("k", index), 100.0, index) assert len(cache) <= 16 def test_repository_cache_version(): with tempfile.TemporaryDirectory() as tmp: db = Database(Path(tmp) / "app.db") db.initialize() try: repo = Repository(db) assert repo.cache_version("videos") == 0 assert repo.bump_cache_version("videos") == 1 assert repo.bump_cache_version("videos") == 2 assert repo.cache_version("videos") == 2 assert repo.cache_version("fans") == 0 repo.bump_cache_version("fans") assert repo.cache_version("fans") == 1 finally: db.close() def main_test(): test_ttl_cache_expiry_and_params() test_ttl_cache_prunes_to_max_items() test_repository_cache_version() print("OK: 缓存 TTL/键/版本失效测试全部通过") if __name__ == "__main__": main_test()