test: 删除重复的逐键 TTL 缓存测试 (#6161) (#6168)

This commit is contained in:
Ekko
2026-07-21 21:51:09 +08:00
committed by GitHub
parent 3b709b7f2e
commit de97cb3c0a

View File

@@ -490,96 +490,6 @@ def test_file_cache_uses_default_ttl_when_omitted(monkeypatch):
assert AsyncFileCache().ttl == 7 * 24 * 3600
def test_memory_backend_supports_per_key_ttl():
"""
同一 region 的 key 应按各自 TTL 过期,不受首个 key 的 TTL 影响。
"""
region = "per_key_ttl"
cache = MemoryBackend()
cache.set("short", "short-value", ttl=10, region=region)
cache.set("long", "long-value", ttl=20, region=region)
region_cache = MemoryBackend._region_caches[cache.get_region(region)]
started_at = region_cache.timer()
region_cache.expire(time=started_at + 11)
assert cache.get("short", region=region) is None
assert cache.get("long", region=region) == "long-value"
assert list(cache.items(region=region)) == [("long", "long-value")]
def test_memory_backend_resets_ttl_when_key_is_rewritten():
"""
重写已有 key 时应从重写时刻按新 TTL 重新计算过期时间。
"""
region = "rewrite_per_key_ttl"
cache = MemoryBackend()
cache.set("key", "old", ttl=10, region=region)
region_cache = MemoryBackend._region_caches[cache.get_region(region)]
started_at = region_cache.timer()
cache.set("key", "new", ttl=20, region=region)
region_cache.expire(time=started_at + 11)
assert cache.get("key", region=region) == "new"
region_cache.expire(time=started_at + 21)
assert cache.get("key", region=region) is None
def test_memory_backend_instances_share_region_with_per_key_ttl():
"""
多个 backend 仍共享 region 数据,但每次写入的显式 TTL 应独立生效。
"""
region = "shared_per_key_ttl"
first = MemoryBackend()
second = MemoryBackend()
first.set("first", "first-value", ttl=10, region=region)
second.set("second", "second-value", ttl=20, region=region)
region_cache = MemoryBackend._region_caches[first.get_region(region)]
started_at = region_cache.timer()
region_cache.expire(time=started_at + 11)
assert second.get("first", region=region) is None
assert first.get("second", region=region) == "second-value"
def test_async_memory_backend_supports_per_key_ttl():
"""
异步代理路径应与同步 backend 共享 region并保留每个 key 的 TTL。
"""
async def run_test():
region = "async_per_key_ttl"
sync_cache = MemoryBackend()
async_cache = AsyncMemoryBackend()
await async_cache.set("short", "short-value", ttl=10, region=region)
sync_cache.set("long", "long-value", ttl=20, region=region)
region_cache = MemoryBackend._region_caches[sync_cache.get_region(region)]
started_at = region_cache.timer()
region_cache.expire(time=started_at + 11)
assert await async_cache.get("short", region=region) is None
assert await async_cache.get("long", region=region) == "long-value"
asyncio.run(run_test())
def test_memory_lru_backend_keeps_capacity_eviction_behavior():
"""
per-key TTL 改造不应影响 LRU region 的容量淘汰行为。
"""
region = "memory_lru"
cache = MemoryBackend(cache_type="lru", maxsize=2)
cache.set("first", 1, region=region)
cache.set("second", 2, region=region)
cache.set("third", 3, region=region)
assert cache.get("first", region=region) is None
assert list(cache.items(region=region)) == [("second", 2), ("third", 3)]
def test_redis_original_key_decodes_quoted_key():
"""
Redis items 返回的 key 应还原为原始缓存 key确保带特殊字符的 key 可继续删除。