Files
PolyGateway/tests/unit/test_cache.py
T
iomgaa bd9da4c911 fix: bring a cached tier back as a tier, not as a bare string
Adding applied_effort to LLMResponse put it through the cache round
trip, where JSON stores a StrEnum as its plain value. Rehydrated raw, a
hit would hand downstream a str while the annotation says Effort, and
every `is Effort.LOW` in the library would quietly answer False on the
hit path only -- the same trap thinking_observation already has a
coercion for.

A value outside this version's vocabulary degrades to None rather than
failing the entry: projects sharing one Redis would otherwise keep
invalidating each other's writes over an attribution field, and None is
the honest reading of a tier this version cannot name.
2026-09-05 05:05:28 -04:00

594 lines
25 KiB
Python

"""CacheMW 与缓存 key 公式测试(ARCH §7.5: 防毒化 key、命中重建、静默降级)。"""
import dataclasses
import hashlib
import json
import pytest
from loguru import logger
from polygateway.backends.memory.cache import InMemoryCache
from polygateway.errors import ResultInvalidError, TransientError
from polygateway.middleware.cache import CacheMW, build_cache_key, digest_messages
from polygateway.middleware.telemetry import TelemetryEmitter
from polygateway.types import (
ChatRequest,
Effort,
LLMResponse,
SourceConfig,
ThinkingObservation,
)
_MSGS = [{"role": "user", "content": "hi"}]
def _resp(content="cached", **overrides):
base = {
"content": content,
"thinking": "",
"model": "m",
"provider": "p",
"prompt_tokens": 1,
"completion_tokens": 2,
"latency_ms": 30,
"ttft_ms": 5.0,
"max_inter_token_ms": 2.0,
"cache_hit": False,
"call_id": "orig",
"source_name": "s1",
"usage_source": "measured",
}
base.update(overrides)
return LLMResponse(**base)
class TestKeyFormula:
def test_same_request_same_key(self):
k1 = build_cache_key("m", _MSGS, "proj", None)
k2 = build_cache_key("m", _MSGS, "proj", None)
assert k1 == k2 and k1.startswith("pgw:cache:")
@pytest.mark.parametrize(
("a", "b"),
[
(("m1", _MSGS, "proj", None), ("m2", _MSGS, "proj", None)),
(("m", _MSGS, "proj", None), ("m", _MSGS, "tenant2", None)),
(("m", _MSGS, "proj", None), ("m", _MSGS, "proj", "epoch2")),
(("m", _MSGS, "proj", "s1"), ("m", _MSGS, "proj", "s2")),
(("m", _MSGS, "proj", None), ("m", [{"role": "user", "content": "yo"}], "proj", None)),
],
)
def test_any_dimension_change_changes_key(self, a, b):
assert build_cache_key(*a) != build_cache_key(*b)
def test_empty_sampling_keeps_legacy_key(self):
"""空采样参数时键形逐字不变,存量缓存不被全量作废(issue #4 决策 C)。
golden 值取自加 sampling 维度之前的实现,不得随实现漂移。
"""
assert build_cache_key("qwen-max", [{"role": "user", "content": "hi"}], "proj", None) == (
"pgw:cache:c54544e8672f4c91373b4a72716a88497445b440b89445aa5379b356b228f58b"
)
assert build_cache_key("qwen-max", [{"role": "user", "content": "hi"}], "proj", "s1") == (
"pgw:cache:eed9cd9cc06acc0dedf4f337b74e06ed3482afdc30fa2acedd194f6cc1df33bf"
)
def test_differing_seed_changes_key(self):
"""issue #4 的直接回归: 5 个 seed 若共用一个 key,标准差会恒为 0。"""
k1 = build_cache_key("m", _MSGS, "proj", None, sampling={"seed": 1})
k2 = build_cache_key("m", _MSGS, "proj", None, sampling={"seed": 2})
assert k1 != k2
def test_sampling_key_order_irrelevant(self):
k1 = build_cache_key("m", _MSGS, "proj", None, sampling={"seed": 1, "temperature": 0})
k2 = build_cache_key("m", _MSGS, "proj", None, sampling={"temperature": 0, "seed": 1})
assert k1 == k2
def test_empty_sampling_equals_omitted(self):
"""空 dict 与不传须同键,否则升级后存量缓存全部 miss。"""
assert build_cache_key("m", _MSGS, "proj", None, sampling={}) == build_cache_key(
"m", _MSGS, "proj", None
)
def test_multimodal_part_digested_not_inlined(self):
big_b64 = "data:image/png;base64," + "A" * 1_000_000
messages = [
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": big_b64}},
{"type": "text", "text": "describe"},
],
}
]
digested = digest_messages(messages)
payload = json.dumps(digested, ensure_ascii=False)
assert len(payload) < 500 # 大图不进 canonical_json
expected = hashlib.sha256(big_b64.encode()).hexdigest()
assert expected in payload # 但字节变化仍改变 key
# 图像字节变化 → key 变
messages2 = [
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": big_b64[:-1] + "B"}},
{"type": "text", "text": "describe"},
],
}
]
assert build_cache_key("m", messages, "p", None) != build_cache_key(
"m", messages2, "p", None
)
def test_request_tier_changes_key(self):
"""同 messages 跑 low 与 max 不得互相命中(issue #20;issue #4 的逐字翻版)。
请求级档位必须**独立于** `model_fingerprint` 进 key: 后者是装配期算出的
集合级指纹,一次调用改档位不会让它变一个字节。
"""
k_low = build_cache_key("m", _MSGS, "proj", None, reasoning_effort=Effort.LOW)
k_max = build_cache_key("m", _MSGS, "proj", None, reasoning_effort=Effort.MAX)
assert k_low != k_max
def test_explicit_none_tier_is_not_the_absent_tier(self):
"""`None`(不表态)与 `Effort.NONE`(要求不推理)是两个 key。
二者合并即毒化: "没写档位"的调用会读到"明确关掉推理"那次的响应,
而后者的内容恰恰是缺推理过程的。
"""
assert build_cache_key("m", _MSGS, "proj", None) != build_cache_key(
"m", _MSGS, "proj", None, reasoning_effort=Effort.NONE
)
def test_absent_tier_keeps_legacy_key(self):
"""不表态档位时键形逐字不变,存量缓存不被本次升级全量作废。
golden 值与 `test_empty_sampling_keeps_legacy_key` 同源,取自加
`reasoning_effort` 维度之前的实现,不得随实现漂移。
"""
assert build_cache_key(
"qwen-max",
[{"role": "user", "content": "hi"}],
"proj",
None,
reasoning_effort=None,
) == ("pgw:cache:c54544e8672f4c91373b4a72716a88497445b440b89445aa5379b356b228f58b")
def test_declared_tier_key_is_a_golden(self):
"""配了档位那一侧同样要有 golden: 字面量变了就是所有该档缓存冷启动。
存量(不表态)那侧的 golden 由 `test_absent_tier_keeps_legacy_key` 守着,
而"档位怎么写进 key"此前没有任何字面量断言——变异实测把 `str(...)` 换成
`repr(...)`,全套件依然全绿(2026-09-05 独立验证查出)。
"""
assert build_cache_key(
"qwen-max",
[{"role": "user", "content": "hi"}],
"proj",
None,
reasoning_effort=Effort.LOW,
) == ("pgw:cache:21d7be93729635b27d4ee54e0e7e7310554bb04faf08919f8ce83794ac33f575")
assert build_cache_key(
"qwen-max",
[{"role": "user", "content": "hi"}],
"proj",
"s1",
reasoning_effort=Effort.NONE,
) == ("pgw:cache:44f4f1ce1ee39e5003a27f4f21a531554cb54d1c66e936d11008d4ff06ddc5e6")
class _Terminal:
def __init__(self, response):
self.response = response
self.calls = 0
async def __call__(self, request):
self.calls += 1
if isinstance(self.response, Exception):
raise self.response
return self.response
def _mw(backend, **kwargs):
defaults = {
"backend": backend,
"model_fingerprint": "m",
"default_namespace": "proj",
"ttl_s": 3600,
}
defaults.update(kwargs)
return CacheMW(**defaults)
class TestCacheFlow:
async def test_miss_then_hit_with_fresh_call_id(self):
backend = InMemoryCache()
mw = _mw(backend)
terminal = _Terminal(_resp())
first = await mw(ChatRequest(messages=_MSGS), terminal)
assert first.cache_hit is False and terminal.calls == 1
second = await mw(ChatRequest(messages=_MSGS), terminal)
assert second.cache_hit is True and second.latency_ms == 0
assert second.content == "cached"
assert second.call_id != first.call_id # 命中生成独立 cache_call_id
assert terminal.calls == 1 # 未再触达内层
async def test_differing_sampling_does_not_hit(self):
"""issue #4 的中间件层回归: 逐 rollout 变 seed 必须回源,不得复用响应。"""
backend = InMemoryCache()
mw = _mw(backend)
terminal = _Terminal(_resp())
await mw(ChatRequest(messages=_MSGS, sampling={"seed": 1}), terminal)
await mw(ChatRequest(messages=_MSGS, sampling={"seed": 2}), terminal)
assert terminal.calls == 2 # 两次都回源
# 同 seed 才命中
third = await mw(ChatRequest(messages=_MSGS, sampling={"seed": 1}), terminal)
assert third.cache_hit is True and terminal.calls == 2
async def test_differing_reasoning_effort_does_not_hit(self):
"""接线门: `CacheMW` 必须把 `request.reasoning_effort` 传进 key 公式。
只测 `build_cache_key` 不够——参数加了却没人传是本改动最可能的落地方式,
那种缺口在公式层的用例里完全看不见。
"""
backend = InMemoryCache()
mw = _mw(backend)
terminal = _Terminal(_resp())
await mw(ChatRequest(messages=_MSGS, reasoning_effort=Effort.LOW), terminal)
await mw(ChatRequest(messages=_MSGS, reasoning_effort=Effort.MAX), terminal)
assert terminal.calls == 2 # 两档各自回源
third = await mw(ChatRequest(messages=_MSGS, reasoning_effort=Effort.LOW), terminal)
assert third.cache_hit is True and terminal.calls == 2 # 同档才命中
async def test_structured_injection_does_not_pollute_key(self):
"""CacheMW 读 sampling 而非 overlay: 结构化注入不该改变缓存身份。"""
backend = InMemoryCache()
mw = _mw(backend)
terminal = _Terminal(_resp())
await mw(ChatRequest(messages=_MSGS, sampling={"seed": 1}), terminal)
polluted = ChatRequest(
messages=_MSGS,
sampling={"seed": 1},
overlay={"seed": 1, "response_format": {"type": "json_object"}},
)
assert (await mw(polluted, terminal)).cache_hit is True
assert terminal.calls == 1
async def test_per_call_namespace_overrides_default(self):
backend = InMemoryCache()
mw = _mw(backend)
terminal = _Terminal(_resp())
await mw(ChatRequest(messages=_MSGS, cache_namespace="tenant-a"), terminal)
# 另一租户不得命中
await mw(ChatRequest(messages=_MSGS, cache_namespace="tenant-b"), terminal)
assert terminal.calls == 2
async def test_failure_not_cached(self):
backend = InMemoryCache()
mw = _mw(backend)
failing = _Terminal(TransientError("boom"))
with pytest.raises(TransientError):
await mw(ChatRequest(messages=_MSGS), failing)
ok = _Terminal(_resp())
await mw(ChatRequest(messages=_MSGS), ok)
assert ok.calls == 1 # 失败未被固化,正常回源
async def test_ttl_expiry(self):
t = {"now": 0.0}
backend = InMemoryCache(now=lambda: t["now"])
mw = _mw(backend, ttl_s=100)
terminal = _Terminal(_resp())
await mw(ChatRequest(messages=_MSGS), terminal)
t["now"] = 101.0
await mw(ChatRequest(messages=_MSGS), terminal)
assert terminal.calls == 2
class TestObservabilityFieldsOnHit:
"""issue #3 决策 B1: 命中行原样回放,与 model/prompt_tokens 同一口径。"""
async def test_fields_replayed_on_hit(self):
backend = InMemoryCache()
mw = _mw(backend)
terminal = _Terminal(
_resp(cached_prompt_tokens=64, model_reported="MiniMax-Text-01-250321")
)
await mw(ChatRequest(messages=_MSGS), terminal)
hit = await mw(ChatRequest(messages=_MSGS), terminal)
assert hit.cache_hit is True
assert hit.cached_prompt_tokens == 64
assert hit.model_reported == "MiniMax-Text-01-250321"
async def test_legacy_cache_entry_without_new_keys_rehydrates(self):
"""旧格式条目(无这两个键)必须照常重建为 None,不得抛异常回源。"""
backend = InMemoryCache()
mw = _mw(backend)
key = build_cache_key("m", _MSGS, "proj", None)
legacy = {
"content": "legacy",
"thinking": "",
"model": "m",
"provider": "p",
"prompt_tokens": 1,
"completion_tokens": 2,
"latency_ms": 30,
"ttft_ms": 5.0,
"max_inter_token_ms": 2.0,
"cache_hit": False,
"call_id": "orig",
"source_name": "s1",
"cost": None,
"usage_source": "measured",
}
await backend.set(key, json.dumps(legacy), ttl_s=100)
terminal = _Terminal(_resp())
hit = await mw(ChatRequest(messages=_MSGS), terminal)
assert hit.content == "legacy" and terminal.calls == 0 # 真的走了缓存
assert hit.cached_prompt_tokens is None and hit.model_reported is None
class TestThinkingObservationRehydration:
"""issue #16/#17: 命中回放必须复活成枚举实例,而不是 JSON 里的裸 str。
裸 str 与字段注解分叉,下游拿 `resp.thinking_observation is
ThinkingObservation.OBSERVED` 判等会在缓存命中路径上静默为 False。
"""
async def test_hit_replays_enum_instance_not_bare_str(self):
backend = InMemoryCache()
mw = _mw(backend)
terminal = _Terminal(_resp(thinking_observation=ThinkingObservation.OBSERVED))
await mw(ChatRequest(messages=_MSGS), terminal)
hit = await mw(ChatRequest(messages=_MSGS), terminal)
assert hit.cache_hit is True and terminal.calls == 1
assert isinstance(hit.thinking_observation, ThinkingObservation)
assert hit.thinking_observation is ThinkingObservation.OBSERVED
async def test_unknown_value_degrades_to_unknown_and_still_hits(self):
"""域外取值降级为 UNKNOWN,内容照常复活——不得因此作废整条缓存。
真实场景: 三项目共用一个 Redis,先升级的项目写入了本版没有的第四态,
未升级的两个项目若把它判成未命中,就会在这些 key 上每次真打网关、随后
覆写回旧值,两个版本互相打对方的缓存(表现是命中率莫名腰斩)。一个纯
可观测性字段不该有能力废掉内容完好的缓存响应。
"""
backend = InMemoryCache()
mw = _mw(backend)
key = build_cache_key("m", _MSGS, "proj", None)
poisoned = dataclasses.asdict(_resp(content="from-a-newer-version"))
poisoned["thinking_observation"] = "partially_observed"
poisoned.pop("structured_data", None)
await backend.set(key, json.dumps(poisoned), 3600)
terminal = _Terminal(_resp())
messages: list[str] = []
sink_id = logger.add(messages.append, level="WARNING")
try:
resp = await mw(ChatRequest(messages=_MSGS), terminal)
finally:
logger.remove(sink_id)
assert terminal.calls == 0 and resp.cache_hit is True
assert resp.content == "from-a-newer-version" # 内容完好,照常复活
assert resp.thinking_observation is ThinkingObservation.UNKNOWN
# 单独一条讲清原因的 warning: 通用的"重建失败"没有任何线索指向真因
hits = [m for m in messages if "partially_observed" in m]
assert len(hits) == 1, f"域外取值必须单独告警一次,实得 {len(hits)} 条: {messages}"
assert "thinking_observation" in hits[0]
assert [m for m in messages if "重建失败" in m] == []
async def test_a_broken_payload_still_falls_back_to_source(self):
"""对照组: 内容完整性真被破坏时,仍必须按未命中回源(降级方向不变)。"""
backend = InMemoryCache()
mw = _mw(backend)
key = build_cache_key("m", _MSGS, "proj", None)
await backend.set(key, "{not json at all", 3600)
terminal = _Terminal(_resp())
resp = await mw(ChatRequest(messages=_MSGS), terminal)
assert terminal.calls == 1 and resp.cache_hit is False
assert resp.content == "cached"
async def test_legacy_entry_without_key_rehydrates_to_default(self):
"""升级前写入的条目没有该键,必须照常复活并落到默认 UNKNOWN。"""
backend = InMemoryCache()
mw = _mw(backend)
key = build_cache_key("m", _MSGS, "proj", None)
legacy = dataclasses.asdict(_resp(content="legacy"))
legacy.pop("thinking_observation")
legacy.pop("structured_data", None)
await backend.set(key, json.dumps(legacy), 3600)
terminal = _Terminal(_resp())
hit = await mw(ChatRequest(messages=_MSGS), terminal)
assert hit.content == "legacy" and terminal.calls == 0
assert hit.thinking_observation is ThinkingObservation.UNKNOWN
class TestAppliedEffortRehydration:
"""issue #20: 实际档同样必须复活成枚举,理由与 `thinking_observation` 逐条相同。
JSON 里存的是 `StrEnum` 的字符串值;不转就复活成裸 str,而库内一路是
`is Effort.LOW` 的身份比较——命中路径上会静默判否,且下游拿到的类型与字段
注解分叉。缓存是档位的**第三条入口**(另两条是 `.env` 解析与 `chat()` 参数),
归一化不变式必须在这里也闭合。
"""
async def test_hit_replays_enum_instance_not_bare_str(self):
backend = InMemoryCache()
mw = _mw(backend)
terminal = _Terminal(_resp(applied_effort=Effort.LOW))
await mw(ChatRequest(messages=_MSGS), terminal)
hit = await mw(ChatRequest(messages=_MSGS), terminal)
assert hit.cache_hit is True and terminal.calls == 1
assert isinstance(hit.applied_effort, Effort)
assert hit.applied_effort is Effort.LOW
async def test_unknown_tier_degrades_to_none_and_still_hits(self):
"""域外档位降级为 None(=不知道这次跑在哪档),不作废内容完好的条目。
降级方向与 `thinking_observation` 同源: 共用一个 Redis 的项目里,先升级
的那个可能写入本版没有的档位名,未升级的项目若判成未命中,两个版本就会
互相打对方的缓存。归因字段不该有能力废掉一条内容完好的响应。
"""
backend = InMemoryCache()
mw = _mw(backend)
key = build_cache_key("m", _MSGS, "proj", None)
poisoned = dataclasses.asdict(_resp(content="from-a-newer-version"))
poisoned["applied_effort"] = "ultra"
poisoned.pop("structured_data", None)
await backend.set(key, json.dumps(poisoned), 3600)
terminal = _Terminal(_resp())
messages: list[str] = []
sink_id = logger.add(messages.append, level="WARNING")
try:
resp = await mw(ChatRequest(messages=_MSGS), terminal)
finally:
logger.remove(sink_id)
assert terminal.calls == 0 and resp.cache_hit is True
assert resp.content == "from-a-newer-version"
assert resp.applied_effort is None
hits = [m for m in messages if "ultra" in m]
assert len(hits) == 1, f"域外档位必须单独告警一次,实得 {len(hits)} 条: {messages}"
assert "applied_effort" in hits[0]
assert [m for m in messages if "重建失败" in m] == []
async def test_legacy_entry_without_key_rehydrates_to_none(self):
"""升级前写入的条目没有该键,必须照常复活并落到默认 None。"""
backend = InMemoryCache()
mw = _mw(backend)
key = build_cache_key("m", _MSGS, "proj", None)
legacy = dataclasses.asdict(_resp(content="legacy"))
legacy.pop("applied_effort")
legacy.pop("structured_data", None)
await backend.set(key, json.dumps(legacy), 3600)
terminal = _Terminal(_resp())
hit = await mw(ChatRequest(messages=_MSGS), terminal)
assert hit.content == "legacy" and terminal.calls == 0
assert hit.applied_effort is None
class _BrokenBackend:
async def get(self, key):
raise ConnectionError("redis down")
async def set(self, key, value, ttl_s):
raise ConnectionError("redis down")
class TestDegradation:
async def test_backend_failure_degrades_silently(self):
mw = _mw(_BrokenBackend())
terminal = _Terminal(_resp())
resp = await mw(ChatRequest(messages=_MSGS), terminal)
assert resp.content == "cached" and terminal.calls == 1 # 读写全降级,调用照常
async def test_corrupt_cache_value_treated_as_miss(self):
backend = InMemoryCache()
mw = _mw(backend)
key = build_cache_key("m", _MSGS, "proj", None)
await backend.set(key, "{not json", 3600)
terminal = _Terminal(_resp())
resp = await mw(ChatRequest(messages=_MSGS), terminal)
assert terminal.calls == 1 and resp.cache_hit is False
class _FakeStrategy:
"""fake StructuredOutputStrategy(T2 冻结的 Protocol,不依赖 T11)。"""
def request_overlay(self, schema):
return {}
def parse(self, text):
data = json.loads(text) # 简化: 直接 json.loads
if not isinstance(data, dict):
raise ResultInvalidError("非对象", raw_text=text)
return data
class _StrictModel:
"""鸭子型 pydantic 模型: model_validate 要求含 answer 键。"""
@classmethod
def model_validate(cls, data):
if "answer" not in data:
raise ValueError("missing answer")
return {"validated": data["answer"]}
class TestStructuredRehydration:
async def test_hit_rebuilds_structured_data(self):
backend = InMemoryCache()
mw = _mw(backend, strategy=_FakeStrategy())
terminal = _Terminal(_resp(content='{"answer": 42}'))
req = ChatRequest(messages=_MSGS, structured=_StrictModel)
await mw(req, terminal)
hit = await mw(req, terminal)
assert hit.cache_hit is True
assert hit.structured_data == {"validated": 42}
assert terminal.calls == 1
async def test_schema_change_revalidation_failure_falls_back_to_source(self):
backend = InMemoryCache()
mw = _mw(backend, strategy=_FakeStrategy())
terminal = _Terminal(_resp(content='{"other": 1}')) # 缓存内容不含 answer
await mw(ChatRequest(messages=_MSGS), terminal) # 无 structured 写入
# 换 schema 读: 重校验失败 → 按未命中回源
again = await mw(ChatRequest(messages=_MSGS, structured=_StrictModel), terminal)
assert terminal.calls == 2 and again.cache_hit is False
async def test_structured_data_not_serialized_into_cache(self):
backend = InMemoryCache()
mw = _mw(backend, strategy=_FakeStrategy())
terminal = _Terminal(
dataclasses.replace(_resp(content='{"answer": 1}'), structured_data={"x": object()})
)
await mw(ChatRequest(messages=_MSGS), terminal) # 不可 JSON 的 structured_data 不阻塞写缓存
key = build_cache_key("m", _MSGS, "proj", None)
raw = await backend.get(key)
assert raw is not None and "structured_data" not in json.loads(raw)
class TestTelemetryCapDoesNotPoisonTheCacheKey:
"""红线之一(issue #12): 遥测截断绝不能改到缓存 key。
`digest_messages` 对 content 非 list 的消息**原样透传同一个 dict 对象**
(本文件上方公式测试依赖的也是这份对象),遥测拿到的与算 key 用的是同一份。
就地截断会让同一组 messages 在遥测前后算出两个不同的 key——全量 miss、
且没有任何报错。故这里测的是"截断没有就地改掉调用方的对象",不只是
"截断函数是纯的"。
"""
class _Rows:
def __init__(self):
self.rows = []
async def record_llm_call(self, **fields):
self.rows.append(fields)
async def test_key_is_byte_identical_across_a_capped_emit(self):
messages = [
{"role": "user", "content": "合同正文" * 31},
{"role": "user", "content": [{"type": "text", "text": "标书正文" * 30}]},
]
before = build_cache_key("m", messages, "proj", None)
rec = self._Rows()
await TelemetryEmitter(rec, text_cap=8).emit_attempt(
request=ChatRequest(messages=messages),
source=SourceConfig(
name="s1",
provider="p",
base_url="https://gw.example/v1",
api_key="sk",
model="m",
timeout_s=10.0,
),
call_id="c",
latency_ms=1,
response=_resp(),
error=None,
)
# 截断确实发生了(否则本用例恒真)
logged = json.loads(rec.rows[0]["messages"])
assert "(略 116 字)" in logged[0]["content"]
assert "(略 112 字)" in logged[1]["content"][0]["text"]
assert build_cache_key("m", messages, "proj", None) == before