group_xinghuo_jinrong/tests/test_locks_redis.py

"""T-201 · Redis 分布式锁双层（Redis 为主、进程内为备）单元测试。

不依赖本机 Redis：所有用例通过 monkeypatch `redis_gateway._gateway` 注入
可控假网关，覆盖「抢到 / 占用超时 / 不可用降级 / 释放只删自己锁 / 三处 key 前缀」。
降级路径任意异常（ConnectionError、Fake 缺方法 AttributeError）一律归类为
unavailable，绝不向上抛——保证 Redis 故障时安全退回进程内锁、业务照常完成。
"""

from __future__ import annotations

import pytest

from app.service.risk import locks
from app.service.risk import redis_gateway


class FakeLockGateway:
    """可控假网关：内存字典模拟 SET NX；raise_on_acquire 模拟 Redis 故障。"""

    def __init__(self, acquire_result: bool = True, raise_on_acquire: Exception | None = None) -> None:
        self._store: dict[str, str] = {}
        self.acquire_result = acquire_result
        self.raise_on_acquire = raise_on_acquire
        self.captured: list[tuple[str, str, int]] = []

    def acquire_lock(self, key: str, token: str, ttl_seconds: int) -> bool:
        self.captured.append((key, token, ttl_seconds))
        if self.raise_on_acquire is not None:
            raise self.raise_on_acquire
        if not self.acquire_result:
            return False
        if key in self._store:
            return False
        self._store[key] = token
        return True

    def release_lock(self, key: str, token: str) -> bool:
        if self._store.get(key) == token:
            del self._store[key]
            return True
        return False


class NoAcquireGateway:
    """测试 Fake 缺方法场景：只有 publish 等旧方法，无 acquire_lock/release_lock。"""

    def publish(self, channel, payload):
        pass


@pytest.fixture()
def gateway(monkeypatch):
    g = FakeLockGateway()
    monkeypatch.setattr(redis_gateway, "_gateway", g)
    return g


def test_redis_lock_acquired_runs_locked_true(gateway):
    """Redis 抢到锁 → fn(locked=True)，且 key 带 TTL 与 lock: 前缀。"""
    seen = {}

    def _fn(locked: bool) -> dict:
        seen["locked"] = locked
        return {"ok": True}

    result = locks.run_locked("agg:event:CUST-1:2026-09-09", _fn)
    assert seen["locked"] is True
    assert result == {"ok": True}
    key, _token, ttl = gateway.captured[0]
    assert key == "lock:agg:event:CUST-1:2026-09-09"
    assert ttl == locks.LOCK_TTL_SECONDS


def test_redis_lock_occupied_times_out(gateway, monkeypatch):
    """锁被占用（抢不到）→ 等待超时降级 fn(locked=False)，不抛异常。"""
    gateway.acquire_result = False
    monkeypatch.setattr(locks, "LOCK_TIMEOUT_SECONDS", 0.2)
    monkeypatch.setattr(locks, "_RETRY_INTERVAL", 0.02)
    seen = {}

    def _fn(locked: bool) -> str:
        seen["locked"] = locked
        return "done"

    assert locks.run_locked("agg:event:CUST-2:2026-09-09", _fn) == "done"
    assert seen["locked"] is False


def test_redis_connection_error_falls_back_process_lock(gateway):
    """Redis 抛 ConnectionError → 归类 unavailable → 退回进程内锁，业务完成。"""
    gateway.raise_on_acquire = ConnectionError("redis down")
    seen = {}

    def _fn(locked: bool) -> str:
        seen["locked"] = locked
        return "business-done"

    assert locks.run_locked("agg:suitability:CUST-3:P-9:2026-09-09", _fn) == "business-done"
    assert seen["locked"] is True  # 进程内锁拿到 → locked=True


def test_redis_missing_method_attribute_error_falls_back(gateway, monkeypatch):
    """Fake 缺 acquire_lock（AttributeError）→ 归类 unavailable，测试不红。"""
    monkeypatch.setattr(redis_gateway, "_gateway", NoAcquireGateway())
    seen = {}

    def _fn(locked: bool) -> str:
        seen["locked"] = locked
        return "ok"

    assert locks.run_locked("l3:CUST-4", _fn) == "ok"
    assert seen["locked"] is True


def test_release_only_owns_lock():
    """释放只删自己的锁：错误 token 返回 False 且锁仍在，正确 token 才删。"""
    g = FakeLockGateway()
    assert g.acquire_lock("lock:k", "tokA", 30) is True
    assert g.release_lock("lock:k", "tokB") is False
    assert "lock:k" in g._store  # 锁仍在
    assert g.release_lock("lock:k", "tokA") is True
    assert "lock:k" not in g._store


def test_three_call_site_keys_have_lock_prefix(gateway):
    """三处调用点 key 形态全覆盖：agg:event: / agg:suitability: / l3: 均带 lock: 前缀。"""
    keys = [
        "agg:event:CUST-X:2026-09-09",
        "agg:suitability:CUST-X:P-Y:2026-09-09",
        "l3:CUST-X",
    ]
    for k in keys:
        locks.run_locked(k, lambda locked: True)
    redis_keys = [c[0] for c in gateway.captured]
    assert all(rk.startswith("lock:") for rk in redis_keys)
    assert "lock:agg:event:CUST-X:2026-09-09" in redis_keys
    assert "lock:agg:suitability:CUST-X:P-Y:2026-09-09" in redis_keys
    assert "lock:l3:CUST-X" in redis_keys