projectAIpopular/tests/test_proxy_semcache.py

"""语义缓存测试（T-P6，M2）：精确/n-gram 阈值/TTL/LRU/重建/晋升/singleflight/SSE 回放。"""
import asyncio
import itertools
import json

import pytest

from gateway.proxy.semcache import (
    SingleFlight,
    SemanticCache,
    grams,
    synth_sse_chunks,
    weighted_jaccard,
)
from gateway.proxy.ledger import Ledger


@pytest.fixture()
def cache(tmp_path):
    led = Ledger.init_db(tmp_path / "p.sqlite3")
    return SemanticCache(led, max_entries=100, sim_threshold=0.92,
                         promote_frequency=5)


def _put(cache, key, text, answer="答案A", model="m", doc_version=1, ttl_hours=72):
    cache.put(key, text, answer, model, doc_version=doc_version, ttl_hours=ttl_hours)


def test_grams_and_weighted_jaccard():
    g1 = grams("什么是递归")
    assert any(len(g) == 2 for g in g1) and any(len(g) == 3 for g in g1)
    assert weighted_jaccard(g1, g1) == 1.0
    assert weighted_jaccard(grams("完全不同话题"), g1) == 0.0


def test_exact_hit_and_miss(cache):
    key = "default|1|" + "a" * 16
    _put(cache, key, "什么是递归", "递归是自调用")
    hit = cache.lookup(key, "什么是递归")
    assert hit and hit["level"] == "exact" and hit["answer"] == "递归是自调用"
    assert cache.lookup(key + "-nope", "完全无关的问题") is None


def test_semantic_hit_above_threshold(cache):
    """同义变体：L2 命中（阈值上）。"""
    key = "default|1|b1"
    _put(cache, key, "请解释一下什么叫做递归函数", "递归解释")
    hit = cache.lookup("default|1|b2", "请解释一下什么叫做递归函数", doc_version=1)
    assert hit and hit["level"] == "semantic"
    assert cache.hits_semantic == 1


def test_semantic_miss_below_threshold(cache):
    """完全不同语义：未命中（阈值下）。"""
    _put(cache, "default|1|c1", "请解释一下什么叫做递归函数", "递归解释")
    assert cache.lookup("default|1|c2", "今天股市行情怎么样", doc_version=1) is None


def test_ttl_expiry_fake_clock(cache):
    _put(cache, "k", "某个问题文本", "旧答案", ttl_hours=1)
    cache._clock = cache._now() + 7200                    # 假时钟 +2h
    assert cache.lookup("k", "某个问题文本", doc_version=1) is None  # 过期不可见


def test_lru_eviction(tmp_path):
    led = Ledger.init_db(tmp_path / "p.sqlite3")
    cache = SemanticCache(led, max_entries=3)
    for i in range(5):
        _put(cache, f"k{i}", f"完全不同的问题编号{i}", f"答{i}")
    assert len(cache._l1) == 3                            # LRU 上限
    assert cache.lookup("k0", "完全不同的问题编号0") is None   # 最旧被驱逐
    assert cache.lookup("k4", "完全不同的问题编号4") is not None


def test_rebuild_from_sqlite(tmp_path):
    """启动时由 semcache 表重建倒排索引。"""
    led = Ledger.init_db(tmp_path / "p.sqlite3")
    c1 = SemanticCache(led, max_entries=100)
    c1.put("k", "解释递归的概念", "持久化答案", "m")
    c2 = SemanticCache(led, max_entries=100)              # 新实例：重建
    hit = c2.lookup("k2", "解释递归的概念", doc_version=1)
    assert hit and hit["answer"] == "持久化答案"


def test_promote_after_five_semantic_hits(cache):
    """L2 命中 5 次 -> 晋升 L1（promote 别名键可精确命中）。"""
    key = "default|1|p1"
    _put(cache, key, "请解释一下什么叫做递归函数呢？", "递归解释（变体）")
    promoted = False
    for i in range(5):
        hit = cache.lookup("default|1|p%d" % (i + 2), "请解释一下什么叫做递归函数呢", doc_version=1)
        assert hit and hit["level"] == "semantic"
        if any(k.startswith("default|1|promoted:") for k in cache._l1):
            promoted = True
    assert promoted and cache.hits_semantic == 5


def test_singleflight_merge_and_bypass():
    """两并发同请求：一登记一等待；超限旁路。"""
    async def scenario():
        sf = SingleFlight()
        fut, slot = sf.try_claim("h1")
        assert fut is None and slot is not None           # 首个登记
        fut2, slot2 = sf.try_claim("h1")
        assert fut2 is not None and slot2 is None         # 第二个等待
        sf.release(slot, result="共享答案")
        got = await sf.wait(fut2)
        assert got == "共享答案"
        # 超限旁路
        sf2 = SingleFlight()
        sf2.MAX = 2
        _f, s1 = sf2.try_claim("a")
        _f2, s2 = sf2.try_claim("b")
        f3, s3 = sf2.try_claim("c")
        assert f3 is None and s3 is None                  # 第三个旁路

    asyncio.run(scenario())


def test_synth_sse_chunks_valid():
    """命中回放：合法 SSE 形状（delta 分块 + finish + [DONE]）。"""
    chunks = synth_sse_chunks("你好世界" * 10, chunk_size=20, model="m", request_id="r1")
    text = b"".join(chunks).decode("utf-8")
    assert text.count("chat.completion.chunk") >= 2
    assert '"finish_reason": "stop"' in text or '"finish_reason":"stop"' in text
    assert text.rstrip("\n").endswith("data: [DONE]")
    content = ""
    done = False
    for raw in chunks:
        for line in raw.decode("utf-8").splitlines():
            if not line.startswith("data:"):
                continue
            payload = line[5:].strip()
            if payload == "[DONE]":
                done = True
                continue
            obj = json.loads(payload)
            assert obj["object"] == "chat.completion.chunk"
            content += obj["choices"][0]["delta"].get("content") or ""
    assert done and "你好世界" in content


# ---------------- 2026-09 优化回归：免并集公式 / 规模上界预筛 ----------------

def test_weighted_jaccard_formula_parity():
    """免并集公式（w_inter/(wA+wB−w_inter)）与直接遍历并集逐位一致（整数权重）。"""
    def direct(ga: set, gb: set) -> float:
        inter = ga & gb
        if not inter:
            return 0.0
        w_inter = sum(2 if len(x) == 3 else 1 for x in inter)
        w_union = sum(2 if len(x) == 3 else 1 for x in (ga | gb))
        return w_inter / w_union

    texts = ["什么是递归函数", "请解释一下什么叫做递归函数呢", "今天股市行情怎么样",
             "ab", "abc", "完整题目描述" * 3]
    gs = [grams(t) for t in texts]
    for ga, gb in itertools.product(gs, repeat=2):
        assert weighted_jaccard(ga, gb) == direct(ga, gb)


def test_prefilter_skips_size_mismatched_candidates(cache):
    """规模悬殊的候选被上界预筛排除；结论与全量计分一致（低于阈值 -> 未命中）。"""
    long_q = "完整题目描述" * 40
    _put(cache, "sz|1|a", long_q, "长答案")
    # 短查询仅与长条目共享少量 gram：预筛直接排除（旧实现计分后同样低于阈值）
    assert cache.lookup("sz|1|b", "完整题目", doc_version=1) is None
    assert cache.hits_semantic == 0