232 lines
6.6 KiB
Python
232 lines
6.6 KiB
Python
"""chongming-cache 双后端测试。
|
||
|
||
redis 后端优先连接真实服务（REDIS_HOST/REDIS_PORT），不可用时
|
||
在进程内启动 fakeredis TCP 服务兜底，保证 redis 后端逻辑始终被验证。
|
||
"""
|
||
|
||
import os
|
||
import sqlite3
|
||
import threading
|
||
import time
|
||
|
||
import pytest
|
||
|
||
from chongming_cache import Cache
|
||
|
||
REDIS_HOST = os.getenv("REDIS_HOST", "127.0.0.1")
|
||
REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
|
||
REDIS_DB = int(os.getenv("REDIS_DB", 15)) # 测试专用库号，避免污染默认库
|
||
|
||
_server = None # fakeredis 兜底服务（进程内）
|
||
|
||
|
||
def _get_redis_endpoint() -> tuple[str, int]:
|
||
"""真实 redis 可用则用之；否则启动 fakeredis TCP 服务兜底。"""
|
||
global _server
|
||
import redis
|
||
|
||
try:
|
||
client = redis.Redis(
|
||
host=REDIS_HOST, port=REDIS_PORT, socket_connect_timeout=1
|
||
)
|
||
if client.ping():
|
||
client.close()
|
||
return REDIS_HOST, REDIS_PORT
|
||
except Exception:
|
||
pass
|
||
|
||
import fakeredis
|
||
|
||
_server = fakeredis.TcpFakeServer(("127.0.0.1", 0), server_type="redis")
|
||
port = _server.server_address[1]
|
||
threading.Thread(target=_server.serve_forever, daemon=True).start()
|
||
return "127.0.0.1", port
|
||
|
||
|
||
_redis_host, _redis_port = _get_redis_endpoint()
|
||
|
||
|
||
@pytest.fixture(scope="session", autouse=True)
|
||
def _shutdown_fake_redis():
|
||
"""会话结束时显式关闭 fakeredis 服务，避免其非 daemon 线程卡住解释器退出。"""
|
||
yield
|
||
if _server is not None:
|
||
_server.shutdown()
|
||
_server.server_close()
|
||
|
||
|
||
@pytest.fixture
|
||
def disk_cache(tmp_path):
|
||
cache = Cache.from_config(
|
||
{"backend": "sqlite", "path": str(tmp_path / "cache.db")}
|
||
)
|
||
yield cache
|
||
cache.close()
|
||
|
||
|
||
@pytest.fixture
|
||
def redis_cache():
|
||
"""redis 后端实例（唯一前缀命名空间，测试结束只清自己的 key）。"""
|
||
cache = Cache.from_config(
|
||
{
|
||
"backend": "redis",
|
||
"redis_host": _redis_host,
|
||
"redis_port": _redis_port,
|
||
"redis_db": REDIS_DB,
|
||
"redis_prefix": f"test:cache:{os.getpid()}:",
|
||
}
|
||
)
|
||
yield cache
|
||
cache.clear()
|
||
cache.close()
|
||
|
||
|
||
def _mapping_checks(cache: Cache) -> None:
|
||
cache["k1"] = {"a": 1}
|
||
cache.set("k2", [1, 2, 3])
|
||
cache.set("k3", "bytes-able", expire=60)
|
||
|
||
assert "k1" in cache
|
||
assert cache["k1"] == {"a": 1}
|
||
assert cache["k2"] == [1, 2, 3]
|
||
assert cache.get("k3") == "bytes-able"
|
||
assert cache.get("missing") is None
|
||
assert cache.get("missing", "d") == "d"
|
||
with pytest.raises(KeyError):
|
||
cache["missing"]
|
||
assert len(cache) == 3
|
||
assert set(cache) == {"k1", "k2", "k3"}
|
||
|
||
del cache["k1"]
|
||
assert "k1" not in cache
|
||
with pytest.raises(KeyError):
|
||
del cache["k1"]
|
||
assert cache.delete("k2") is True
|
||
assert cache.delete("k2") is False
|
||
assert len(cache) == 1
|
||
|
||
cache.clear()
|
||
assert len(cache) == 0
|
||
|
||
|
||
def test_mapping_disk(disk_cache):
|
||
_mapping_checks(disk_cache)
|
||
|
||
|
||
def test_expire_disk(disk_cache):
|
||
disk_cache.set("e", 1, expire=1)
|
||
assert "e" in disk_cache
|
||
time.sleep(1.2)
|
||
assert "e" not in disk_cache # sqlite 读时惰性淘汰
|
||
assert disk_cache.get("e", "miss") == "miss"
|
||
|
||
|
||
def test_touch_disk(disk_cache):
|
||
disk_cache.set("e", 1, expire=1)
|
||
assert disk_cache.touch("e", expire=10) is True
|
||
time.sleep(1.2)
|
||
assert "e" in disk_cache
|
||
assert disk_cache.touch("missing", expire=10) is False
|
||
|
||
|
||
def test_context_manager(tmp_path):
|
||
with Cache.from_config(
|
||
{"backend": "sqlite", "path": str(tmp_path / "cm.db")}
|
||
) as cache:
|
||
cache["x"] = 1
|
||
assert cache["x"] == 1
|
||
|
||
|
||
def test_from_config_defaults(tmp_path, monkeypatch):
|
||
# 缺省 backend 时默认 sqlite；缺省 path 时用 ./cache_data/cache.db
|
||
monkeypatch.chdir(tmp_path)
|
||
cache = Cache.from_config({})
|
||
try:
|
||
cache.set("k", 1)
|
||
assert cache["k"] == 1
|
||
finally:
|
||
cache.close()
|
||
|
||
|
||
def test_invalid_backend(tmp_path):
|
||
with pytest.raises(ValueError):
|
||
Cache.from_config({"backend": "memcached", "path": str(tmp_path)})
|
||
|
||
|
||
def test_sqlite_json_interop_and_multi_instance(tmp_path):
|
||
"""SQLite 落盘是纯 JSON 文本（Rust 可读），且两个后端实例共享同一文件。"""
|
||
db = str(tmp_path / "shared.db")
|
||
a = Cache.from_config({"backend": "sqlite", "path": db})
|
||
b = Cache.from_config({"backend": "sqlite", "path": db})
|
||
try:
|
||
# 待共享值：纯 JSON 类型，应原样以 JSON 文本落盘
|
||
a.set("app:latest", {"msg": "hello", "n": 42})
|
||
# 共享文件可跨实例读取（多进程安全语义）
|
||
assert b.get("app:latest") == {"msg": "hello", "n": 42}
|
||
# 直接查库：v 列为 JSON 文本而非 pickle 字节
|
||
row = sqlite3.connect(db).execute(
|
||
"SELECT v FROM cache WHERE k='app:latest'"
|
||
).fetchone()
|
||
assert row is not None
|
||
assert row[0] == '{"msg":"hello","n":42}'
|
||
# TTL 过期后跨实例同样读不到
|
||
a.set("exp", 1, expire=1)
|
||
time.sleep(1.1)
|
||
assert b.get("exp") is None
|
||
finally:
|
||
a.close()
|
||
b.close()
|
||
|
||
|
||
# ── redis 后端 ──────────────────────────────────────────────────
|
||
|
||
|
||
def test_mapping_redis(redis_cache):
|
||
_mapping_checks(redis_cache)
|
||
|
||
|
||
def test_expire_redis(redis_cache):
|
||
redis_cache.set("e", 1, expire=1)
|
||
assert "e" in redis_cache
|
||
time.sleep(1.2)
|
||
assert "e" not in redis_cache # TTL 到期后 EXISTS 为假
|
||
|
||
|
||
def test_touch_redis(redis_cache):
|
||
redis_cache.set("e", 1, expire=1)
|
||
assert redis_cache.touch("e", expire=10) is True
|
||
time.sleep(1.2)
|
||
assert "e" in redis_cache
|
||
assert redis_cache.touch("missing", expire=10) is False
|
||
|
||
|
||
def test_prefix_isolation():
|
||
"""不同前缀实例共享同一 Redis，clear 只清自己的 key。"""
|
||
base = {
|
||
"backend": "redis",
|
||
"redis_host": _redis_host,
|
||
"redis_port": _redis_port,
|
||
"redis_db": REDIS_DB,
|
||
}
|
||
a = Cache.from_config({**base, "redis_prefix": f"test:iso:a:{os.getpid()}:"})
|
||
b = Cache.from_config({**base, "redis_prefix": f"test:iso:b:{os.getpid()}:"})
|
||
try:
|
||
a["x"] = 1
|
||
b["y"] = 2
|
||
a.clear()
|
||
assert "x" not in a and "y" in b
|
||
assert len(a) == 0 and len(b) == 1
|
||
assert set(b) == {"y"}
|
||
finally:
|
||
a.clear()
|
||
b.clear()
|
||
a.close()
|
||
b.close()
|
||
|
||
|
||
def test_redis_binary_value_roundtrip(redis_cache):
|
||
"""非 pickle 可序列化值应正常往返（dict/bytes 等）。"""
|
||
value = {"nested": {"list": [1, 2, 3]}, "bytes": b"\x00\x01"}
|
||
redis_cache["k"] = value
|
||
assert redis_cache["k"] == value
|