chongming-utils-cache/tests/test_cache.py

"""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()


# ── set_nx / 分布式锁租约语义（sqlite）────────────────────────────
# 回归：持锁进程崩溃后残留的「已过期」锁行必须让位，否则后续 set_nx 永远失败，
# inline 持有者选举 / MQTT Bridge 等分布式锁死锁（生产单网关 sqlite 后端场景）。


def test_set_nx_treats_expired_row_as_vacant(tmp_path):
    """已过期条目（如崩溃残留的锁行）不得阻塞 set_nx（对齐 redis SET NX 语义）。"""
    db = str(tmp_path / "nx.db")
    cache = Cache.from_config({"backend": "sqlite", "path": db})
    try:
        raw = sqlite3.connect(db)
        raw.execute(
            "INSERT INTO cache(k, v, expire) VALUES('lk', ?, ?)",
            ('"old-token"', time.time() - 5),
        )
        raw.commit()
        raw.close()
        assert cache.backend.set_nx("lk", "new-token", expire=30) is True
        assert cache.get("lk") == "new-token"
    finally:
        cache.close()


def test_set_nx_live_row_blocks(tmp_path):
    """未过期的存活条目仍保持互斥：第二次 set_nx 必须失败。"""
    db = str(tmp_path / "nx2.db")
    cache = Cache.from_config({"backend": "sqlite", "path": db})
    try:
        assert cache.backend.set_nx("lk", "a", expire=60) is True
        assert cache.backend.set_nx("lk", "b", expire=60) is False
        assert cache.get("lk") == "a"
    finally:
        cache.close()


def test_distributed_lock_reacquire_after_crash_expiry(tmp_path):
    """锁级回归：持有者崩溃（不释放、租约过期）后，新实例必须能重新获取锁。"""
    from chongming_cache.lock import DistributedLock

    db = str(tmp_path / "lock.db")
    cache = Cache.from_config({"backend": "sqlite", "path": db})
    try:
        # 正常释放后立即可重新获取
        l1 = DistributedLock(cache, "inline_owner:test")
        assert l1.acquire() is True
        l1.release()
        l2 = DistributedLock(cache, "inline_owner:test")
        assert l2.acquire() is True
        l2.release()

        # 崩溃场景：获取后不 release，直接把锁行置为已过期（模拟租约到期未续约）
        crashed = DistributedLock(cache, "inline_owner:test")
        assert crashed.acquire() is True
        raw = sqlite3.connect(db)
        raw.execute(
            "UPDATE cache SET expire = ? WHERE k = 'inline_owner:test'",
            (time.time() - 1,),
        )
        raw.commit()
        raw.close()
        # 新进程（新 token）应重新获得锁；修复前 INSERT OR IGNORE 撞残留行恒失败
        assert DistributedLock(cache, "inline_owner:test").acquire() is True
    finally:
        cache.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


def test_set_nx_after_expiry_redis(redis_cache):
    """redis 对照：过期 key 自动让位，set_nx 可再次成功；存活 key 保持互斥。"""
    assert redis_cache.backend.set_nx("lk", "a", expire=1) is True
    assert redis_cache.backend.set_nx("lk", "b", expire=10) is False
    time.sleep(1.2)
    assert redis_cache.backend.set_nx("lk", "c", expire=10) is True
    assert redis_cache.get("lk") == "c"