add: chongming-cache 统一缓存工具（diskcache / redis 双后端，dict-like 接口）

tests/test_cache.py

@@ -0,0 +1,205 @@
+"""chongming-cache 双后端测试。
+
+redis 后端优先连接真实服务（REDIS_HOST/REDIS_PORT），不可用时
+在进程内启动 fakeredis TCP 服务兜底，保证 redis 后端逻辑始终被验证。
+"""
+
+import os
+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": "diskcache", "cache_store_path": str(tmp_path / "cache")}
+    )
+    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  # diskcache 读时惰性淘汰
+    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": "diskcache", "cache_store_path": str(tmp_path / "cm")}
+    ) as cache:
+        cache["x"] = 1
+        assert cache["x"] == 1
+
+
+def test_from_config_defaults(tmp_path, monkeypatch):
+    # 缺省 backend 时默认 diskcache；缺省 cache_store_path 时用 ./cache_data
+    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", "cache_store_path": str(tmp_path)})
+
+
+# ── 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