tzt
3250e4e099
- normalizer.py：resolve_bucket（X-Campus-Bucket 头 > 映射 > default，D-P2）； normalize_messages（role\u0001content\u0002 串接，剔易变字段与时间戳行， 多模态 content 取 text）；is_cacheable（system 外 >1 条 = 多轮不缓存，D-P5）； canonical_hash（bucket|doc_version|sha256(norm)——模板/资料不参与哈希， 资料更新 = 版本+1 旧键失效）；shape（[canonical_system(无才注入)] -> [课程资料前缀(文件读取,不入git)] -> [原 messages]，易变顶层字段剔除） - 测试 +8：同义同哈希/易变剔除/模板不入哈希/doc_version 失效/多轮判定/ 整形顺序+资料注入/桶解析回落/序列化形态，全量 370 passed
107 lines
4.5 KiB
Python
107 lines
4.5 KiB
Python
"""规范化测试（T-P5）：同义同哈希 / 易变剔除 / 顺序固定 / doc_version 失效 / 多轮不缓存。"""
|
||
import hashlib
|
||
|
||
from gateway.proxy.config import BucketCfg
|
||
from gateway.proxy.normalizer import (
|
||
canonical_hash,
|
||
is_cacheable,
|
||
normalize_messages,
|
||
resolve_bucket,
|
||
shape,
|
||
)
|
||
|
||
|
||
def _body(content="什么是递归？", role="user", n_extra=0, **over):
|
||
msgs = [{"role": "system", "content": "你是助教"},
|
||
{"role": role, "content": content}]
|
||
for i in range(n_extra):
|
||
msgs.append({"role": "assistant", "content": f"答 {i}"})
|
||
b = {"model": "deepseek-chat", "messages": msgs}
|
||
b.update(over)
|
||
return b
|
||
|
||
|
||
def test_same_semantics_same_hash():
|
||
"""同输入同哈希（确定性）+ content 顺序敏感。"""
|
||
b1 = _body()
|
||
b2 = _body()
|
||
assert canonical_hash("default", 1, b1) == canonical_hash("default", 1, b2)
|
||
b3 = _body("什么是递归") # 差一个问号 -> 不同哈希
|
||
assert canonical_hash("default", 1, b1) != canonical_hash("default", 1, b3)
|
||
|
||
|
||
def test_volatile_fields_stripped():
|
||
"""规则 2：temperature/seed 等易变字段不参与哈希；时间戳内容行剔除。"""
|
||
b1 = _body(temperature=0.2, seed=1)
|
||
b2 = _body(temperature=0.9, seed=42)
|
||
assert canonical_hash("default", 1, b1) == canonical_hash("default", 1, b2)
|
||
b3 = _body("当前时间：2026-09-05 10:00:00\n什么是递归？")
|
||
b4 = _body("当前时间：2026-09-06 23:59:59\n什么是递归？")
|
||
assert canonical_hash("default", 1, b3) == canonical_hash("default", 1, b4)
|
||
|
||
|
||
def test_system_template_not_in_hash():
|
||
"""规则 3：桶模板/资料前缀不参与哈希——不同模板同哈希。"""
|
||
b = _body()
|
||
assert canonical_hash("python24", 1, b) == canonical_hash("python24", 1, b)
|
||
# 桶不同 -> 键不同（隔离）
|
||
assert canonical_hash("default", 1, b) != canonical_hash("python24", 1, b)
|
||
|
||
|
||
def test_doc_version_bumps_key():
|
||
"""doc_version+1 后旧缓存键不可见（资料更新失效）。"""
|
||
b = _body()
|
||
assert canonical_hash("course", 1, b) != canonical_hash("course", 2, b)
|
||
|
||
|
||
def test_multiturn_not_cacheable():
|
||
"""规则 5：system 外 >1 条 -> 多轮 cacheable=False（D-P5）。"""
|
||
assert is_cacheable(_body()) is True # system+1 user
|
||
assert is_cacheable(_body(n_extra=1)) is False # +assistant 历史
|
||
assert is_cacheable({"messages": [{"role": "user", "content": "x"}]}) is True
|
||
|
||
|
||
def test_shape_fixed_order_and_template_injection(tmp_path):
|
||
"""规则 4：整形顺序 [canonical_system] -> [doc_prefix] -> [原 messages]；
|
||
自带 system 时不重复注入模板；资料前缀文件内容插入。"""
|
||
prefix_file = tmp_path / "python24.txt"
|
||
prefix_file.write_text("第一章：变量与类型", encoding="utf-8")
|
||
bucket = BucketCfg(name="python24", system_template="你是 Python 助教。",
|
||
doc_prefix_file=str(prefix_file), doc_version=3)
|
||
# 无 system：注入模板 + 资料
|
||
b = {"model": "m", "messages": [{"role": "user", "content": "q"}]}
|
||
shaped = shape(b, bucket)
|
||
roles = [m["role"] for m in shaped["messages"]]
|
||
assert roles == ["system", "system", "user"]
|
||
assert "Python 助教" in shaped["messages"][0]["content"]
|
||
assert "课程资料 v3" in shaped["messages"][1]["content"]
|
||
assert "第一章" in shaped["messages"][1]["content"]
|
||
# 自带 system：模板不重复注入；资料前缀在原 messages 之前（规则 4 顺序）
|
||
b2 = {"model": "m", "messages": [{"role": "system", "content": "自定义"},
|
||
{"role": "user", "content": "q"}]}
|
||
shaped2 = shape(b2, bucket)
|
||
assert "课程资料" in shaped2["messages"][0]["content"]
|
||
assert shaped2["messages"][1]["content"] == "自定义"
|
||
# 易变字段剔除
|
||
assert "temperature" not in shape(_body(temperature=0.7), bucket)
|
||
|
||
|
||
def test_resolve_bucket_header_and_fallback():
|
||
"""D-P2：X-Campus-Bucket 头优先；未知名回落 default。"""
|
||
from types import SimpleNamespace
|
||
|
||
class Cfg:
|
||
def bucket(self, name):
|
||
return SimpleNamespace(name=name or "default")
|
||
|
||
r = resolve_bucket(_body(), {"x-campus-bucket": "python24"}, Cfg())
|
||
assert r.name == "python24"
|
||
r2 = resolve_bucket(_body(), {}, Cfg())
|
||
assert r2.name == "default"
|
||
|
||
|
||
def test_normalize_serialization_format():
|
||
"""规则 1：role\\u0001content\\u0002 串接形态。"""
|
||
b = {"messages": [{"role": "user", "content": "hi"}]}
|
||
assert normalize_messages(b) == "user\u0001hi\u0002"
|