Files
MoviePilot/tests/test_cache_system.py

1271 lines
40 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
验证同步与异步缓存的过期、隔离、原子领取和后端资源生命周期。
"""
import asyncio
import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from unittest.mock import AsyncMock, MagicMock
import pytest
from app.adapters.cache import redis as redis_module
from app.adapters.cache.backends import AsyncFileBackend, FileBackend
from app.adapters.cache.redis import (
AsyncRedisBackend,
AsyncRedisHelper,
RedisBackend,
RedisHelper,
serialize,
)
from app.foundation.singleton import Singleton
from app.runtime.cache import (
AsyncFileCache,
AsyncMemoryBackend,
FileCache,
MemoryBackend,
TTLCache,
cached,
)
from app.runtime.config import ConfigModel, settings
def test_file_backend_items_keep_relative_keys_and_bytes(tmp_path):
"""
文件缓存遍历应返回可继续删除的相对 key,并保持二进制内容不变。
"""
cache = FileBackend(base=tmp_path)
cache.set("nested/poster.jpg", b"\xff\xd8image", region="images")
items = list(cache.items(region="images"))
assert items == [("nested/poster.jpg", b"\xff\xd8image")]
assert cache.popitem(region="images") == ("nested/poster.jpg", b"\xff\xd8image")
assert not cache.exists("nested/poster.jpg", region="images")
def test_clear_package_tool_cache_only_removes_pip_and_uv_old_files(tmp_path, monkeypatch):
"""
包安装工具缓存清理只处理 pip/uv 子目录,不接管整个 .cache 或业务缓存。
"""
from app.startup.initializers.modules import clear_package_tool_cache
old_time = time.time() - 40 * 24 * 3600
cache_root = tmp_path / ".cache"
old_pip = cache_root / "pip" / "old.whl"
old_uv = cache_root / "uv" / "old.archive"
unknown = cache_root / "other" / "old.bin"
business = tmp_path / "cache" / "images" / "old.jpg"
for path in (old_pip, old_uv, unknown, business):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("x", encoding="utf-8")
os.utime(path, (old_time, old_time))
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
monkeypatch.setattr(settings, "PACKAGE_CACHE_ROOT", None)
monkeypatch.setattr(settings, "PACKAGE_CACHE_DAYS", 30)
clear_package_tool_cache()
assert not old_pip.exists()
assert not old_uv.exists()
assert unknown.exists()
assert business.exists()
def test_clear_package_tool_cache_disabled_when_days_non_positive(tmp_path, monkeypatch):
"""
PACKAGE_CACHE_DAYS 小于等于 0 时不清理包安装缓存。
"""
from app.startup.initializers.modules import clear_package_tool_cache
old_time = time.time() - 40 * 24 * 3600
old_pip = tmp_path / ".cache" / "pip" / "old.whl"
old_pip.parent.mkdir(parents=True, exist_ok=True)
old_pip.write_text("x", encoding="utf-8")
os.utime(old_pip, (old_time, old_time))
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
monkeypatch.setattr(settings, "PACKAGE_CACHE_ROOT", None)
monkeypatch.setattr(settings, "PACKAGE_CACHE_DAYS", 0)
clear_package_tool_cache()
assert old_pip.exists()
def test_clear_package_tool_cache_isolates_subdir_errors(tmp_path, monkeypatch):
"""
单个工具缓存目录清理失败,不影响另一个工具缓存目录。
"""
from app.startup.initializers.modules import clear_package_tool_cache
calls = []
def fake_clear(path, days):
calls.append((path.name, days))
if path.name == "pip":
raise OSError("pip cache locked")
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
monkeypatch.setattr(settings, "PACKAGE_CACHE_ROOT", str(tmp_path / "custom-package-cache"))
monkeypatch.setattr(settings, "PACKAGE_CACHE_DAYS", 30)
monkeypatch.setattr("app.startup.initializers.modules.SystemUtils.clear", fake_clear)
clear_package_tool_cache()
assert calls == [("pip", 30), ("uv", 30)]
def test_clear_package_tool_cache_uses_package_cache_root(tmp_path, monkeypatch):
"""
PACKAGE_CACHE_ROOT 用作 pip/uv 清理根目录,不扩大到配置目录下其他缓存。
"""
from app.startup.initializers.modules import clear_package_tool_cache
old_time = time.time() - 40 * 24 * 3600
package_cache_root = tmp_path / "custom-package-cache"
old_pip = package_cache_root / "pip" / "old.whl"
default_pip = tmp_path / ".cache" / "pip" / "old.whl"
for path in (old_pip, default_pip):
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("x", encoding="utf-8")
os.utime(path, (old_time, old_time))
monkeypatch.setattr(settings, "CONFIG_DIR", str(tmp_path))
monkeypatch.setattr(settings, "PACKAGE_CACHE_ROOT", str(package_cache_root))
monkeypatch.setattr(settings, "PACKAGE_CACHE_DAYS", 30)
clear_package_tool_cache()
assert not old_pip.exists()
assert default_pip.exists()
def test_init_modules_does_not_clear_package_tool_cache(monkeypatch):
"""
包安装缓存清理由通用临时清理入口触发,模块启动路径不直接执行清理。
"""
from app.startup.initializers import modules as modules_initializer
called = False
def fail_if_called():
nonlocal called
called = True
raise AssertionError("init_modules must not clear package tool cache directly")
monkeypatch.setattr(modules_initializer, "clear_package_tool_cache", fail_if_called)
monkeypatch.setattr(modules_initializer, "init_managed_resources", lambda: None)
monkeypatch.setattr(modules_initializer, "configure_doh_composition", lambda: None)
monkeypatch.setattr(modules_initializer, "SitesHelper", lambda: None)
monkeypatch.setattr(modules_initializer, "update_resources", lambda: None)
monkeypatch.setattr(modules_initializer, "user_auth", lambda: None)
monkeypatch.setattr(modules_initializer, "ModuleManager", lambda: None)
monkeypatch.setattr(modules_initializer.EventManager, "start", lambda self: None)
monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "async_init_plugin_report", AsyncMock())
monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "async_init_subscribe_report", AsyncMock())
monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "get_user_uuid", lambda: None)
monkeypatch.setattr(modules_initializer.MoviePilotServerHelper, "get_github_user", lambda: None)
init_agent = AsyncMock()
monkeypatch.setattr(modules_initializer, "init_agent", init_agent)
monkeypatch.setattr(modules_initializer, "start_frontend", lambda: None)
monkeypatch.setattr(modules_initializer, "check_auth", lambda: None)
async def initialize_modules() -> None:
try:
await modules_initializer.init_modules()
finally:
await modules_initializer.stop_database_runtime()
asyncio.run(initialize_modules())
assert called is False
init_agent.assert_awaited_once_with()
def test_file_backend_delete_missing_key_is_noop(tmp_path):
"""
删除不存在的文件缓存 key 应保持幂等,不向调用方抛出文件系统异常。
"""
cache = FileBackend(base=tmp_path)
cache.delete("missing", region="default")
assert not cache.exists("missing", region="default")
def test_memory_backend_delete_missing_key_is_noop():
"""
内存缓存后端 delete 与其他后端保持一致,不存在时直接返回。
"""
cache = MemoryBackend()
cache.delete("missing", region="missing_delete")
assert not cache.exists("missing", region="missing_delete")
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_cached_zero_ttl_does_not_cache_sync_result():
"""
同步 cached(ttl=0) 应立即过期,不能退化为 LRU 永久缓存。
"""
calls = 0
@cached(region="sync_zero_ttl", ttl=0)
def load_value():
nonlocal calls
calls += 1
return calls
assert load_value() == 1
assert load_value() == 2
def test_cached_zero_ttl_does_not_cache_async_result():
"""
异步 cached(ttl=0) 应与同步路径保持一致。
"""
calls = 0
@cached(region="async_zero_ttl", ttl=0)
async def load_value():
nonlocal calls
calls += 1
return calls
async def run_test():
return await load_value(), await load_value()
assert asyncio.run(run_test()) == (1, 2)
def test_cached_ttl_provider_resolves_current_value_for_each_write():
"""
动态 TTL 工厂应在每次写入时读取新快照,而不是在导入期固化。
"""
state = {"ttl": 10}
calls = 0
@cached(region="sync_dynamic_ttl", ttl_provider=lambda: state["ttl"])
def load_value():
nonlocal calls
calls += 1
return calls
assert load_value() == 1
region_cache = MemoryBackend._region_caches[
MemoryBackend.get_region("sync_dynamic_ttl")
]
started_at = region_cache.timer()
region_cache.expire(time=started_at + 11)
state["ttl"] = 30
assert load_value() == 2
region_cache.expire(time=started_at + 20)
assert load_value() == 2
def test_cached_rejects_fixed_and_dynamic_ttl_together():
"""
固定 TTL 与动态 TTL 同时存在时应在装饰阶段明确拒绝。
"""
with pytest.raises(ValueError, match="不能同时设置"):
@cached(ttl=10, ttl_provider=lambda: 20)
def load_value():
"""
返回固定值,触发装饰阶段的互斥过期配置校验。
"""
return 1
def test_cached_empty_ttl_expires_empty_result_sooner_sync():
"""
同步 cached 的空结果应按 empty_ttl 独立过期,不受默认 ttl 影响。
"""
calls = 0
@cached(region="sync_empty_ttl", ttl=600, empty_ttl=10,
empty_if=lambda value: not value.get("results"))
def load_value():
nonlocal calls
calls += 1
return {"results": []}
assert load_value() == {"results": []}
assert load_value() == {"results": []}
assert calls == 1
region_cache = MemoryBackend._region_caches[MemoryBackend.get_region("sync_empty_ttl")]
started_at = region_cache.timer()
region_cache.expire(time=started_at + 11)
assert load_value() == {"results": []}
assert calls == 2
def test_cached_empty_ttl_keeps_default_ttl_for_non_empty_result():
"""
非空结果应继续使用默认 ttl,不被 empty_ttl 缩短。
"""
calls = 0
@cached(region="sync_empty_ttl_nonempty", ttl=600, empty_ttl=10,
empty_if=lambda value: not value.get("results"))
def load_value():
nonlocal calls
calls += 1
return {"results": [1]}
assert load_value() == {"results": [1]}
region_cache = MemoryBackend._region_caches[MemoryBackend.get_region("sync_empty_ttl_nonempty")]
started_at = region_cache.timer()
region_cache.expire(time=started_at + 11)
assert load_value() == {"results": [1]}
assert calls == 1
def test_cached_empty_ttl_without_empty_if_uses_falsy_check():
"""
未提供 empty_if 时按假值判断空结果,空列表按 empty_ttl 独立过期。
"""
calls = 0
@cached(region="sync_empty_ttl_falsy", ttl=600, empty_ttl=10)
def load_value():
nonlocal calls
calls += 1
return []
assert load_value() == []
assert load_value() == []
assert calls == 1
region_cache = MemoryBackend._region_caches[MemoryBackend.get_region("sync_empty_ttl_falsy")]
started_at = region_cache.timer()
region_cache.expire(time=started_at + 11)
assert load_value() == []
assert calls == 2
def test_cached_empty_ttl_expires_empty_result_sooner_async():
"""
异步 cached 的空结果应与同步路径一致,按 empty_ttl 独立过期。
"""
calls = 0
@cached(region="async_empty_ttl", ttl=600, empty_ttl=10,
empty_if=lambda value: not value.get("results"))
async def load_value():
nonlocal calls
calls += 1
return {"results": []}
async def run_first_round():
assert await load_value() == {"results": []}
assert await load_value() == {"results": []}
asyncio.run(run_first_round())
assert calls == 1
region_cache = MemoryBackend._region_caches[MemoryBackend.get_region("async_empty_ttl")]
started_at = region_cache.timer()
region_cache.expire(time=started_at + 11)
assert asyncio.run(load_value()) == {"results": []}
assert calls == 2
def test_memory_backend_global_clear_is_safe_during_region_creation():
"""
全局清理与新 region 创建应由同一把锁串行化,不能并发修改注册表。
"""
cache = MemoryBackend()
cache.set("existing", 1, region="clear_existing")
started = threading.Event()
release = threading.Event()
region_cache = MemoryBackend._region_caches[cache.get_region("clear_existing")]
original_clear = region_cache.clear
def blocking_clear():
started.set()
release.wait(timeout=5)
original_clear()
region_cache.clear = blocking_clear
clear_thread = threading.Thread(target=cache.clear, args=(None,))
clear_thread.start()
assert started.wait(timeout=5)
set_thread = threading.Thread(
target=cache.set,
args=("new", 2),
kwargs={"region": "clear_new"},
)
set_thread.start()
set_thread.join(timeout=0.1)
assert set_thread.is_alive()
release.set()
clear_thread.join(timeout=5)
set_thread.join(timeout=5)
assert not clear_thread.is_alive()
assert not set_thread.is_alive()
assert cache.get("existing", region="clear_existing") is None
assert cache.get("new", region="clear_new") == 2
def test_memory_backend_rejects_region_cache_type_conflicts():
"""
同名 region 不得同时作为 TTL 和 LRU 缓存使用。
"""
region = "cache_type_conflict"
ttl_cache = MemoryBackend(cache_type="ttl")
lru_cache = MemoryBackend(cache_type="lru")
ttl_cache.set("ttl", 1, ttl=10, region=region)
try:
lru_cache.set("lru", 2, region=region)
except ValueError as err:
assert "different cache type" in str(err)
else:
raise AssertionError("cache type conflict must be rejected")
assert ttl_cache.get("ttl", region=region) == 1
assert ttl_cache.get("lru", region=region) is None
def test_memory_backend_reuses_existing_region_cache():
"""
同一 region 的后续写入应复用首次创建的底层缓存对象。
"""
region = "reuse_region_cache"
cache = MemoryBackend()
cache.set("first", 1, ttl=10, region=region)
first_region_cache = MemoryBackend._region_caches[cache.get_region(region)]
cache.set("second", 2, ttl=20, region=region)
assert MemoryBackend._region_caches[cache.get_region(region)] is first_region_cache
def test_memory_backend_uses_default_maxsize_for_zero_override():
"""
动态 maxsize=0 应与构造参数一致,回退到 backend 默认容量。
"""
region = "zero_maxsize_override"
cache = MemoryBackend(maxsize=8)
cache.set("key", "value", maxsize=0, region=region)
region_cache = MemoryBackend._region_caches[cache.get_region(region)]
assert region_cache.maxsize == 8
assert cache.get("key", region=region) == "value"
def test_memory_backend_preserves_zero_ttl():
"""
显式 ttl=0 不应回退到默认 TTL,并应删除已有同名值。
"""
cache = MemoryBackend(ttl=30)
cache.set("key", "old", region="zero_ttl")
cache.set("key", "new", ttl=0, region="zero_ttl")
assert cache.get("key", region="zero_ttl") is None
def test_memory_backend_preserves_negative_ttl():
"""
显式负 TTL 应保持立即过期语义,并删除已有同名值。
"""
cache = MemoryBackend(ttl=30)
cache.set("key", "old", region="negative_ttl")
cache.set("key", "new", ttl=-1, region="negative_ttl")
assert cache.get("key", region="negative_ttl") is None
def test_memory_backend_uses_zero_default_ttl():
"""
backend 的默认 ttl=0 应保持立即过期语义。
"""
cache = MemoryBackend(ttl=0)
cache.set("key", "value", region="zero_default_ttl")
assert cache.get("key", region="zero_default_ttl") is None
def test_redis_backend_close_keeps_shared_pool(monkeypatch):
"""
关闭单个 Redis 缓存不得关闭全部缓存共享的连接池,连接池由关闭流程统一收口。
"""
helper = MagicMock()
async_helper = MagicMock()
async_helper.close = AsyncMock()
monkeypatch.setattr("app.adapters.cache.redis.RedisHelper", lambda: helper)
monkeypatch.setattr("app.adapters.cache.redis.AsyncRedisHelper", lambda: async_helper)
RedisBackend().close()
asyncio.run(AsyncRedisBackend().close())
helper.close.assert_not_called()
async_helper.close.assert_not_awaited()
def test_redis_backend_treats_zero_ttl_as_expired():
"""
Redis backend 应删除 ttl=0 的同名 key,避免向 Redis 发送无效 EX 0。
"""
class RedisHelperStub:
"""
记录同步 Redis 写入和删除调用,不访问真实缓存。
"""
deleted = None
set_called = False
def set(self, key, value, ttl, region, **kwargs):
"""
记录是否尝试写入缓存,供零过期时间的删除语义断言使用。
"""
self.set_called = True
def delete(self, key, region):
"""
记录被删除的缓存键和区域。
"""
self.deleted = (key, region)
cache = object.__new__(RedisBackend)
cache.ttl = 30
cache.redis_helper = RedisHelperStub()
cache.set("key", "value", ttl=0, region="zero_ttl")
assert cache.redis_helper.deleted == ("key", "zero_ttl")
assert not cache.redis_helper.set_called
def test_async_redis_backend_treats_zero_ttl_as_expired():
"""
异步 Redis backend 应删除 ttl=0 的同名 key,不发送无效 EX 0。
"""
class AsyncRedisHelperStub:
"""
记录异步 Redis 写入和删除调用,不访问真实缓存。
"""
deleted = None
set_called = False
async def set(self, key, value, ttl, region, **kwargs):
"""
记录是否尝试写入缓存,供零过期时间的删除语义断言使用。
"""
self.set_called = True
async def delete(self, key, region):
"""
记录被删除的缓存键和区域。
"""
self.deleted = (key, region)
async def run_test():
cache = object.__new__(AsyncRedisBackend)
cache.ttl = 30
cache.redis_helper = AsyncRedisHelperStub()
await cache.set("key", "value", ttl=0, region="zero_ttl")
return cache.redis_helper
helper = asyncio.run(run_test())
assert helper.deleted == ("key", "zero_ttl")
assert not helper.set_called
@pytest.fixture
def redis_composed(compose_cache_backend):
"""
以 Redis 缓存装配平台缓存。
"""
compose_cache_backend("redis")
def test_file_cache_preserves_zero_ttl_in_redis_mode(redis_composed):
"""
FileCache 在 Redis 模式下不应把显式 ttl=0 替换为临时文件默认 TTL。
"""
assert FileCache(ttl=0).ttl == 0
def test_async_file_cache_preserves_zero_ttl_in_redis_mode(redis_composed):
"""
AsyncFileCache 在 Redis 模式下应与同步工厂保持相同 TTL 语义。
"""
assert AsyncFileCache(ttl=0).ttl == 0
def test_file_cache_uses_default_ttl_when_omitted(monkeypatch, redis_composed):
"""
未传 TTL 时仍使用 TEMP_FILE_DAYS 配置的默认值。
"""
monkeypatch.setattr(settings, "TEMP_FILE_DAYS", 7)
assert FileCache().ttl == 7 * 24 * 3600
assert AsyncFileCache().ttl == 7 * 24 * 3600
def test_redis_original_key_decodes_quoted_key():
"""
Redis items 返回的 key 应还原为原始缓存 key,确保带特殊字符的 key 可继续删除。
"""
redis_key = b"region:DEFAULT:key:nested/poster%20one.jpg"
assert RedisHelper._RedisHelper__get_original_key(redis_key) == "nested/poster one.jpg"
def test_redis_helper_uses_blocking_pool_settings(monkeypatch):
"""
Redis 同步客户端应使用阻塞连接池,避免并发峰值直接耗尽 Redis 连接数。
"""
calls = {}
class FakeClient:
"""
模拟同步 Redis 客户端。
"""
def __init__(self, connection_pool):
self.connection_pool = connection_pool
self.config_calls = []
self.closed = False
def ping(self):
"""
模拟 Redis ping。
"""
calls["ping"] = True
def config_set(self, key, value):
"""
记录 Redis 配置写入。
"""
self.config_calls.append((key, value))
def close(self):
"""
标记客户端已关闭。
"""
self.closed = True
def fake_from_url(url, **kwargs):
"""
记录连接池构造参数。
"""
calls["pool"] = {"url": url, **kwargs}
return "pool"
monkeypatch.setattr(settings, "CACHE_BACKEND_URL", "redis://cache:6379/2")
monkeypatch.setattr(settings, "CACHE_REDIS_MAX_CONNECTIONS", 7)
monkeypatch.setattr(settings, "CACHE_REDIS_POOL_TIMEOUT", 3)
monkeypatch.setattr("app.adapters.cache.redis.redis.BlockingConnectionPool.from_url", fake_from_url)
monkeypatch.setattr("app.adapters.cache.redis.redis.Redis", FakeClient)
helper = RedisHelper()
helper.close()
helper._connect()
assert calls["pool"]["url"] == "redis://cache:6379/2"
assert calls["pool"]["max_connections"] == 7
assert calls["pool"]["timeout"] == 3
assert calls["pool"]["decode_responses"] is False
assert calls["ping"] is True
assert ("maxmemory-policy", "allkeys-lru") in helper.client.config_calls
helper.close()
def test_redis_helper_pop_uses_atomic_getdel():
"""
Redis 缓存领取必须通过单条 GETDEL 命令完成。
"""
calls = []
class FakeClient:
"""
记录原子领取命令并返回序列化的挑战数据。
"""
def getdel(self, key):
"""
记录领取的缓存键,返回预设的序列化挑战内容。
"""
calls.append(key)
return serialize({"challenge": "value"})
helper = RedisHelper()
helper.client = FakeClient()
try:
value = helper.pop("token", region="passkey_challenge")
finally:
helper.client = None
assert value == {"challenge": "value"}
assert calls == ["region:passkey_challenge:key:token"]
def test_redis_helper_strict_consume_propagates_backend_failure():
"""
严格领取必须区分 Redis 故障与键不存在,旧 pop 仍保持兼容返回值。
"""
class FailingClient:
"""
模拟 GETDEL 连接故障。
"""
def getdel(self, _key):
"""
报告 Redis 命令失败。
"""
raise ConnectionError("redis unavailable")
helper = RedisHelper()
helper.client = FailingClient()
try:
with pytest.raises(ConnectionError, match="redis unavailable"):
helper.consume("token", region="passkey_challenge")
assert helper.pop("token", region="passkey_challenge") is None
finally:
helper.client = None
def test_redis_helper_strict_store_requires_backend_acknowledgement():
"""
严格写入不得把 Redis 未确认写入当成成功签发。
"""
class RejectingClient:
"""
模拟 Redis 拒绝确认 SET。
"""
def set(self, *_args, **_kwargs):
"""
返回未写入状态。
"""
return False
helper = RedisHelper()
helper.client = RejectingClient()
try:
with pytest.raises(RuntimeError, match="not acknowledged"):
helper.store("token", "challenge", region="passkey_challenge")
helper.set("token", "challenge", region="passkey_challenge")
finally:
helper.client = None
def test_memory_atomic_cache_consume_has_single_winner_and_honors_ttl():
"""
内存原子缓存与 Redis 一样只允许一次领取并服从 TTL。
"""
cache = MemoryBackend(ttl=60)
region = "atomic_memory_contract"
cache.clear(region=region)
cache.store("token", "challenge", region=region)
with ThreadPoolExecutor(max_workers=8) as executor:
results = list(executor.map(
lambda _: cache.consume("token", region=region),
range(8),
))
assert results.count("challenge") == 1
assert results.count(None) == 7
cache.store("expired", "challenge", ttl=0, region=region)
assert cache.consume("expired", region=region) is None
def test_ttl_cache_legacy_pop_uses_atomic_consume_contract():
"""
插件既有 TTLCache.pop 入口保持可用并获得原子领取语义。
"""
cache = TTLCache(region="legacy_atomic_pop", maxsize=8, ttl=60)
cache.clear()
cache.set("token", "challenge")
def pop_token(_index):
"""
兼容入口不存在时返回空值,便于汇总并发结果。
"""
try:
return cache.pop("token")
except KeyError:
return None
with ThreadPoolExecutor(max_workers=8) as executor:
results = list(executor.map(pop_token, range(8)))
assert results.count("challenge") == 1
assert results.count(None) == 7
def test_redis_backend_strict_store_and_consume_use_atomic_helper():
"""
Redis 适配器将严格缓存契约完整委托给底层原子实现。
"""
calls = []
class FakeHelper:
"""
记录严格 Redis 缓存调用。
"""
def store(self, key, value, ttl=None, region=None, **kwargs):
"""
记录严格写入。
"""
calls.append(("store", key, value, ttl, region, kwargs))
def consume(self, key, region=None):
"""
记录原子领取。
"""
calls.append(("consume", key, region))
return "challenge"
backend = RedisBackend(ttl=60)
backend.redis_helper = FakeHelper()
backend.store("token", "challenge", region="passkey_challenge")
assert backend.consume("token", region="passkey_challenge") == "challenge"
assert calls == [
("store", "token", "challenge", 60, "passkey_challenge", {}),
("consume", "token", "passkey_challenge"),
]
def test_async_redis_helper_uses_blocking_pool_settings(monkeypatch):
"""
Redis 异步客户端应使用阻塞连接池,避免高并发缓存读取立刻抛出连接耗尽错误。
"""
calls = {}
class FakeAsyncClient:
"""
模拟异步 Redis 客户端。
"""
def __init__(self, connection_pool):
self.connection_pool = connection_pool
self.config_calls = []
self.closed = False
async def ping(self):
"""
模拟 Redis ping。
"""
calls["ping"] = True
async def config_set(self, key, value):
"""
记录 Redis 配置写入。
"""
self.config_calls.append((key, value))
async def close(self):
"""
标记客户端已关闭。
"""
self.closed = True
def fake_from_url(url, **kwargs):
"""
记录连接池构造参数。
"""
calls["pool"] = {"url": url, **kwargs}
return "async_pool"
async def run_connect():
helper = AsyncRedisHelper()
await helper.close()
await helper._connect()
config_calls = list(helper.client.config_calls)
await helper.close()
return config_calls
monkeypatch.setattr(settings, "CACHE_BACKEND_URL", "redis://cache:6379/3")
monkeypatch.setattr(settings, "CACHE_REDIS_MAX_CONNECTIONS", 9)
monkeypatch.setattr(settings, "CACHE_REDIS_POOL_TIMEOUT", 4)
monkeypatch.setattr("app.adapters.cache.redis.AsyncBlockingConnectionPool.from_url", fake_from_url)
monkeypatch.setattr("app.adapters.cache.redis.Redis", FakeAsyncClient)
config_calls = asyncio.run(run_connect())
assert calls["pool"]["url"] == "redis://cache:6379/3"
assert calls["pool"]["max_connections"] == 9
assert calls["pool"]["timeout"] == 4
assert calls["pool"]["decode_responses"] is False
assert calls["ping"] is True
assert ("maxmemory-policy", "allkeys-lru") in config_calls
def test_async_redis_helper_does_not_close_client_from_foreign_loop(monkeypatch):
"""
异步 Redis 客户端按事件循环隔离,切换循环时不得跨循环关闭旧连接池。
"""
singleton_key = (AsyncRedisHelper, (), frozenset())
monkeypatch.delitem(Singleton._instances, singleton_key, raising=False)
clients = []
class FakePool:
"""
记录连接池所属事件循环。
"""
def __init__(self):
self.loop = asyncio.get_running_loop()
self.wrong_loop_close = False
async def aclose(self):
"""
模拟连接池关闭并校验事件循环归属。
"""
if asyncio.get_running_loop() is not self.loop:
self.wrong_loop_close = True
raise RuntimeError("pool belongs to a different event loop")
class FakeClient:
"""
模拟带事件循环归属的异步 Redis 客户端。
"""
def __init__(self, connection_pool):
self.connection_pool = connection_pool
self.loop = asyncio.get_running_loop()
self.wrong_loop_close = False
clients.append(self)
async def ping(self):
"""
模拟 Redis ping。
"""
async def config_set(self, _key, _value):
"""
模拟 Redis 配置写入。
"""
async def close(self):
"""
模拟客户端关闭并校验事件循环归属。
"""
if asyncio.get_running_loop() is not self.loop:
self.wrong_loop_close = True
raise RuntimeError("client belongs to a different event loop")
def fake_from_url(_url, **_kwargs):
"""
为当前事件循环创建独立连接池。
"""
return FakePool()
monkeypatch.setattr(
"app.adapters.cache.redis.AsyncBlockingConnectionPool.from_url",
fake_from_url,
)
monkeypatch.setattr("app.adapters.cache.redis.Redis", FakeClient)
async def connect():
"""
在当前事件循环建立异步 Redis 客户端。
"""
return await AsyncRedisHelper()._connect()
first = asyncio.run(connect())
second = asyncio.run(connect())
assert first is not second
assert len(clients) == 2
assert not first.wrong_loop_close
assert not first.connection_pool.wrong_loop_close
assert second.loop is not first.loop
def test_redis_helpers_watch_pool_settings():
"""
Redis 连接池配置变化应触发客户端重建。
"""
assert "CACHE_REDIS_MAX_CONNECTIONS" in RedisHelper.CONFIG_WATCH
assert "CACHE_REDIS_POOL_TIMEOUT" in RedisHelper.CONFIG_WATCH
assert "CACHE_REDIS_MAX_CONNECTIONS" in AsyncRedisHelper.CONFIG_WATCH
assert "CACHE_REDIS_POOL_TIMEOUT" in AsyncRedisHelper.CONFIG_WATCH
assert "BIG_MEMORY_MODE" in RedisHelper.CONFIG_WATCH
assert "BIG_MEMORY_MODE" in AsyncRedisHelper.CONFIG_WATCH
def test_redis_pool_defaults_cover_startup_concurrency():
"""
Redis 默认连接池应为启动期并发读写留出容量和等待时间。
"""
config = ConfigModel()
assert config.CACHE_REDIS_MAX_CONNECTIONS == 512
assert config.CACHE_REDIS_POOL_TIMEOUT == 10
def test_async_file_backend_missing_region_has_no_items(tmp_path):
"""
异步文件缓存缺失区域时应返回空迭代,而不是伪造空 key。
"""
async def collect_items():
cache = AsyncFileBackend(base=tmp_path)
return [item async for item in cache.items(region="missing")]
assert asyncio.run(collect_items()) == []
def test_async_file_backend_items_keep_relative_keys_and_bytes(tmp_path):
"""
异步文件缓存遍历应与同步文件缓存保持相同 key 和二进制语义。
"""
async def collect_items():
cache = AsyncFileBackend(base=tmp_path)
await cache.set("nested/poster.jpg", b"\xff\xd8image", region="images")
items = [item async for item in cache.items(region="images")]
popped = await cache.popitem(region="images")
exists = await cache.exists("nested/poster.jpg", region="images")
return items, popped, exists
items, popped, exists = asyncio.run(collect_items())
assert items == [("nested/poster.jpg", b"\xff\xd8image")]
assert popped == ("nested/poster.jpg", b"\xff\xd8image")
assert not exists
def test_file_backend_items_skip_directories(tmp_path):
"""
文件缓存遍历应递归读取有效缓存文件,不把目录当成缓存项。
"""
cache = FileBackend(base=tmp_path)
cache.set("nested/value", b"value", region="region")
(tmp_path / "region" / "empty_dir").mkdir()
assert list(cache.items(region="region")) == [("nested/value", b"value")]
# 第三方请求封装把带凭据的完整请求地址当作缓存键,这里用假值代表真实键形态
CREDENTIAL_CACHE_KEY = (
"request_('GET', 'https://api.example.com/3/person/25673"
"?api_key=SUPERSECRET&append_to_response=videos&language=zh', None, None)"
)
@pytest.fixture
def redis_failure_logs(monkeypatch):
"""
收集 Redis 后端故障日志,避免用例把错误打到真实日志。
:return: 收集到的错误与告警日志文本列表
"""
logs: list[str] = []
monkeypatch.setattr(redis_module.logger, "error", logs.append)
monkeypatch.setattr(redis_module.logger, "warn", logs.append)
return logs
def _failing_sync_client():
"""
构造各命令都抛出连接错误的同步 Redis 客户端替身。
:return: set/exists/get/delete 均失败的 MagicMock 客户端
"""
client = MagicMock()
error = ConnectionError("(104, 'Connection reset by peer')")
for command in ("set", "exists", "get", "delete"):
getattr(client, command).side_effect = error
return client
class _FailingAsyncClient:
"""
所有命令都抛出连接错误的异步 Redis 客户端替身。
"""
async def _fail(self, *_args, **_kwargs):
"""
模拟 Redis 命令在读写阶段失败。
"""
raise ConnectionError("(104, 'Connection reset by peer')")
set = _fail
exists = _fail
get = _fail
delete = _fail
def test_redis_helper_failure_logs_mask_cache_key_credentials(monkeypatch, redis_failure_logs):
"""
缓存键携带请求凭据时,Redis 故障日志不得输出凭据原文,并保留可定位的键结构。
"""
helper = RedisHelper()
monkeypatch.setattr(helper, "client", None)
monkeypatch.setattr(helper, "_connect", lambda: _failing_sync_client())
helper.set(CREDENTIAL_CACHE_KEY, {"result": 1}, region="TMDb:request")
assert helper.get(CREDENTIAL_CACHE_KEY, region="TMDb:request") is None
assert helper.exists(CREDENTIAL_CACHE_KEY, region="TMDb:request") is False
helper.delete(CREDENTIAL_CACHE_KEY, region="TMDb:request")
assert len(redis_failure_logs) == 4
for message in redis_failure_logs:
assert "SUPERSECRET" not in message
assert "api_key=<REDACTED>" in message
assert "3/person/25673" in message
assert "language=zh" in message
def test_async_redis_helper_failure_logs_mask_cache_key_credentials(monkeypatch, redis_failure_logs):
"""
异步后端与同步后端必须采用同样的日志脱敏语义。
"""
helper = AsyncRedisHelper()
async def _fake_connect():
"""
返回命令必然失败的异步客户端替身。
"""
return _FailingAsyncClient()
monkeypatch.setattr(helper, "client", None)
monkeypatch.setattr(helper, "_connect", _fake_connect)
async def exercise():
"""
依次触发异步后端的写入、读取、存在性判断与删除故障。
"""
await helper.set(CREDENTIAL_CACHE_KEY, {"result": 1}, region="TMDb:request")
assert await helper.get(CREDENTIAL_CACHE_KEY, region="TMDb:request") is None
assert await helper.exists(CREDENTIAL_CACHE_KEY, region="TMDb:request") is False
await helper.delete(CREDENTIAL_CACHE_KEY, region="TMDb:request")
asyncio.run(exercise())
assert len(redis_failure_logs) == 4
for message in redis_failure_logs:
assert "SUPERSECRET" not in message
assert "api_key=<REDACTED>" in message
assert "language=zh" in message
def test_mask_cache_key_masks_encoded_redis_keys_and_bytes():
"""
日志脱敏对 URL 编码后的 Redis 键与字节键同样生效。
"""
redis_key = f"region:TMDb:request:key:{redis_module.quote(CREDENTIAL_CACHE_KEY)}"
masked = redis_module._mask_cache_key(redis_key, encoded=True)
assert "SUPERSECRET" not in masked
assert "api_key=<REDACTED>" in masked
assert redis_module._mask_cache_key(b"region:poster:key:nested/value") == (
"region:poster:key:nested/value"
)