PolyGateway/tests/integration/test_redis_cache.py

"""真实 Redis 缓存集成测试(实验室远程实例;CLAUDE.md: Redis 相关不 mock)。

前置: `.env`/环境变量提供 `REDIS_URL`;缺失时整组 skip(T13 门在验收前必须真跑)。
隔离: 所有 key 落在一次性命名空间 `pgw:test:{uuid}`,teardown 精确清理,
绝不 SCAN/FLUSH 全库,不触碰在用数据。
"""

import json
import os
import uuid

import httpx
import pytest
from dotenv import dotenv_values

from polygateway.backends.redis_cache import RedisCache
from polygateway.middleware.cache import CacheMW, build_cache_key
from polygateway.types import ChatRequest, LLMResponse

_REDIS_URL = os.environ.get("REDIS_URL") or dotenv_values(".env").get("REDIS_URL")

pytestmark = pytest.mark.skipif(
    not _REDIS_URL, reason="需实验室远程 Redis: 在 .env 设置 REDIS_URL(M1 验收前必须真跑)"
)

_MSGS = [{"role": "user", "content": "integration-hi"}]


def _resp(content="cached"):
    return LLMResponse(content, "", "m", "p", 1, 2, 30, None, None, False, "orig")


@pytest.fixture
async def redis_cache():
    import redis.asyncio as aioredis

    client = aioredis.from_url(
        _REDIS_URL, decode_responses=True, socket_connect_timeout=3.0, socket_timeout=3.0
    )
    cache = RedisCache(client)
    used_keys: list[str] = []
    original_set = cache.set

    async def tracking_set(key, value, ttl_s):
        used_keys.append(key)
        await original_set(key, value, ttl_s)

    cache.set = tracking_set  # 记录写入的 key,teardown 精确删除
    yield cache
    if used_keys:
        await client.delete(*used_keys)
    await client.aclose()


@pytest.fixture
def namespace():
    return f"pgw:test:{uuid.uuid4().hex}"


class TestRealRedisRoundtrip:
    async def test_set_get_ttl(self, redis_cache, namespace):
        key = build_cache_key("m", _MSGS, namespace, None)
        assert await redis_cache.get(key) is None
        await redis_cache.set(key, json.dumps({"v": 1}), 60)
        assert json.loads(await redis_cache.get(key)) == {"v": 1}

    async def test_cache_mw_miss_then_hit(self, redis_cache, namespace):
        mw = CacheMW(
            backend=redis_cache,
            model_fingerprint="m",
            default_namespace=namespace,
            ttl_s=60,
        )
        calls = {"n": 0}

        async def terminal(request):
            calls["n"] += 1
            return _resp()

        first = await mw(ChatRequest(messages=_MSGS), terminal)
        second = await mw(ChatRequest(messages=_MSGS), terminal)
        assert first.cache_hit is False and second.cache_hit is True
        assert calls["n"] == 1

    async def test_namespace_isolation_on_shared_instance(self, redis_cache, namespace):
        """多项目共用同一 Redis 时,namespace 不同绝不互相命中(防毒化铁律)。"""
        other = f"pgw:test:{uuid.uuid4().hex}"
        k1 = build_cache_key("m", _MSGS, namespace, None)
        k2 = build_cache_key("m", _MSGS, other, None)
        assert k1 != k2
        await redis_cache.set(k1, json.dumps({"ns": 1}), 60)
        assert await redis_cache.get(k2) is None


class TestDegradationAgainstDeadRedis:
    async def test_unreachable_redis_degrades_to_miss(self):
        """断连方向: 缓存后端挂 → 静默降级,调用照常(铁律,与限流相反)。"""
        import redis.asyncio as aioredis

        dead = RedisCache(
            aioredis.from_url(
                "redis://127.0.0.1:1/0",
                socket_connect_timeout=0.2,
                socket_timeout=0.2,
                decode_responses=True,
            )
        )
        mw = CacheMW(
            backend=dead, model_fingerprint="m", default_namespace="pgw:test:dead", ttl_s=60
        )

        async def terminal(request):
            return _resp("alive")

        resp = await mw(ChatRequest(messages=_MSGS), terminal)
        assert resp.content == "alive" and resp.cache_hit is False


def _sse_response():
    chunk = json.dumps({"choices": [{"delta": {"content": "hello"}}]})
    usage = json.dumps({"choices": [], "usage": {"prompt_tokens": 3, "completion_tokens": 4}})
    body = f"data: {chunk}\n\ndata: {usage}\n\ndata: [DONE]\n\n"
    return httpx.Response(200, content=body.encode(), headers={"content-type": "text/event-stream"})


class TestClientWithRealRedis:
    async def test_from_settings_style_assembly_with_redis_cache(self, redis_cache, namespace):
        """GatewayClient 组装 + 真实 Redis 缓存的端到端命中。"""
        from polygateway import GatewayClient
        from polygateway.backends.memory.breaker import InMemoryGate
        from polygateway.backends.memory.limiter import InMemoryLimiter
        from polygateway.sources import RoundRobinSelector
        from polygateway.structured.json_repair import JsonRepairStrategy
        from polygateway.transports.openai_compat import OpenAICompatTransport
        from polygateway.types import (
            BackpressurePolicy,
            BreakerConfig,
            GlobalLimits,
            RetryPolicy,
            SourceConfig,
        )

        src = SourceConfig(
            name="qwen_1",
            provider="qwen",
            base_url="https://gw.example/v1",
            api_key="sk",
            model="qwen-max",
            timeout_s=5.0,
        )
        client = GatewayClient(
            scope="llm",
            sources=[src],
            selector=RoundRobinSelector(),
            limiter=InMemoryLimiter(
                scope="llm", sources={src.name: src}, global_limits=GlobalLimits(0, 0, 0)
            ),
            breaker=InMemoryGate(config=BreakerConfig(5, 60.0, 120.0)),
            transport=OpenAICompatTransport(
                client_factory=lambda s: httpx.AsyncClient(
                    transport=httpx.MockTransport(lambda req: _sse_response())
                )
            ),
            retry=RetryPolicy(3, 2.0, 30.0),
            backpressure=BackpressurePolicy(300.0, 0.01),
            cache=redis_cache,
            cache_namespace=namespace,
            cache_ttl_s=60,
            structured_strategy=JsonRepairStrategy(),
        )
        first = await client.chat(_MSGS)
        second = await client.chat(_MSGS)
        assert first.cache_hit is False and second.cache_hit is True
        assert second.content == "hello"