121 lines
6.8 KiB
Python
121 lines
6.8 KiB
Python
# ============================================================
|
||
# M3 知识与偏好黄金测试（moduleId: golden-m3, 发版门禁 §14.3）
|
||
# 固化不变式：
|
||
# K1 检索必带出处（assetId/title/version）且未命中返回空（诚实原则 §8.1）
|
||
# K2 报告数字全部可溯源到快照字段（§9.10 硬规则 1）
|
||
# K3 偏好个性化：采用权重 > 试排权重；无信号回退默认（§8.3）
|
||
# K4 知识资产同名重复入库版本递增（§8.1 版本化）
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
import re # 报告数字抽取
|
||
|
||
from server.knowledge.assets import KnowledgeStore # 资产库（临时路径实例化）
|
||
from server.knowledge.preferences import PreferenceStore # 偏好仓
|
||
from server.knowledge.retrieval import search # 混合检索
|
||
from server.aps_domain.reports import build_daily_report, build_version_diff_report # 报告
|
||
from server.state.seed import seed_world # 种子世界
|
||
from server.engines import get_engine # 引擎（造版本用）
|
||
from server.engines.base import EngineParams # 引擎入参
|
||
from server.timeutil import add_minutes, fmt_date, today0 # 日期工具
|
||
|
||
|
||
def _make_version(world, strategy="DELIVERY_FIRST"):
|
||
"""在世界里真实跑一版排产（报告测试的前置数据）。"""
|
||
counters = {} # 独立发号器
|
||
def next_id(kind):
|
||
counters[kind] = counters.get(kind, 0) + 1
|
||
return counters[kind] + 1000
|
||
params = EngineParams(orderIds=[], engineType="RULE", strategyTemplate=strategy,
|
||
planningHorizonDays=14, startDate=fmt_date(add_minutes(today0(), 24 * 60)))
|
||
return get_engine("RULE").solve(world, params, next_id)
|
||
|
||
|
||
# ---------------- K1：检索带出处 + 诚实未命中 ----------------
|
||
def test_search_returns_provenance(tmp_path):
|
||
"""命中必带 assetId/title/version；无关问题返回空列表（不编造）。"""
|
||
kb = KnowledgeStore(path=str(tmp_path / "kn.json")) # 临时库（自动播种）
|
||
hits = search(kb.assets, "换线有什么规定") # SOP 问题
|
||
assert hits, "种子 SOP 应能命中" # 有命中
|
||
top = hits[0] # 最相关
|
||
assert top["title"] == "换线标准SOP" # 命中正确资产
|
||
for key in ("assetId", "title", "version", "snippet"): # 出处字段齐全
|
||
assert top[key], f"命中缺出处字段 {key}"
|
||
miss = search(kb.assets, "quantum blockchain xyzzy") # 无关问题（无重叠词）
|
||
assert miss == [], "无关问题必须诚实返回空" # 未命中为空
|
||
|
||
|
||
# ---------------- K2：报告数字可溯源 ----------------
|
||
def test_daily_report_numbers_from_snapshot():
|
||
"""日报正文中的核心数字必须与冻结快照字段一致。"""
|
||
world = seed_world() # 种子世界
|
||
_make_version(world) # 真实排一版
|
||
report = build_daily_report(world) # 生成日报
|
||
snap = report["snapshot"] # 冻结快照
|
||
assert snap is not None # 有快照
|
||
md = report["markdown"] # 报告正文
|
||
assert f"**{snap['poCount']}** 个" in md # 订单数可溯源
|
||
assert f"**{snap['woCount']}** 个" in md # 工单数可溯源
|
||
assert f"**{snap['conflictCount']}** 项" in md # 冲突数可溯源
|
||
assert snap["versionNo"] in md # 版本号出现在出处行
|
||
|
||
|
||
def test_version_diff_report_needs_two_versions():
|
||
"""不足两版时诚实拒绝；两版后 KPI 表出现新旧版本号。"""
|
||
world = seed_world() # 种子世界
|
||
r0 = build_version_diff_report(world) # 无版本
|
||
assert r0["reportId"] is None # 拒绝生成
|
||
_make_version(world, "DELIVERY_FIRST") # 第一版
|
||
_make_version(world, "CAPACITY_BALANCE") # 第二版
|
||
r = build_version_diff_report(world) # 生成对比
|
||
assert r["reportId"] # 成功
|
||
assert r["snapshot"]["new"]["versionNo"] in r["markdown"] # 新版号在正文
|
||
assert r["snapshot"]["old"]["versionNo"] in r["markdown"] # 旧版号在正文
|
||
|
||
|
||
# ---------------- K3：偏好个性化 ----------------
|
||
def test_preference_weighting(tmp_path):
|
||
"""采用（2分）> 试排（1分）；无信号回退系统默认。"""
|
||
ps = PreferenceStore(path=str(tmp_path / "pref.json")) # 临时仓
|
||
default, scores = ps.preferred_strategy() # 无样本
|
||
assert default == "COMPREHENSIVE" and scores == {} # 回退默认
|
||
ps.record("DELIVERY_FIRST", source="schedule.run") # 试排交期 ×2 = 2分
|
||
ps.record("DELIVERY_FIRST", source="schedule.run")
|
||
ps.record("CAPACITY_BALANCE", source="scenario.apply") # 采用均衡 ×1 = 2分…
|
||
ps.record("CAPACITY_BALANCE", source="scenario.apply") # ×2 = 4分（采用权重更高）
|
||
best, scores = ps.preferred_strategy() # 求偏好
|
||
assert best == "CAPACITY_BALANCE", f"采用权重应更高，got {scores}" # 采用胜出
|
||
assert scores["DELIVERY_FIRST"] == 2 and scores["CAPACITY_BALANCE"] == 4 # 得分可解释
|
||
|
||
|
||
# ---------------- K4：资产版本化 ----------------
|
||
def test_asset_versioning(tmp_path):
|
||
"""同名资产重复入库版本号递增（v1 → v2）。"""
|
||
kb = KnowledgeStore(path=str(tmp_path / "kn.json")) # 临时库
|
||
a1 = kb.add("sop", "测试SOP", "第一版内容") # 首次入库
|
||
assert a1["version"] == "v1" # v1
|
||
a2 = kb.add("sop", "测试SOP", "第二版内容") # 同名再入
|
||
assert a2["version"] == "v2" # 版本递增
|
||
assert kb.find_by_title("测试SOP") is not None # 标题可检索
|
||
|
||
|
||
def test_hybrid_search_units_with_chunks(tmp_path):
|
||
"""带 chunks 的资产按单元检索，仍强制出处。"""
|
||
kb = KnowledgeStore(path=str(tmp_path / "kn.json"))
|
||
kb.assets = []
|
||
kb._write()
|
||
kb.add_with_chunks(
|
||
kind="sop", title="夜班换线禁令", content="摘要",
|
||
chunks=[{"chunkId": "c0", "text": "每日十六点后禁止跨产品族换线作业。", "seq": 0}],
|
||
tags=["换线", "夜班"], approved=True,
|
||
)
|
||
hits = search(kb.assets, "夜班换线")
|
||
# search() 走整篇；再用 hybrid on units
|
||
from server.knowledge.retrieval import hybrid_search
|
||
units = kb.iter_search_units()
|
||
hits2 = hybrid_search(units, "禁止跨产品族换线", top_k=2)
|
||
assert hits2
|
||
for key in ("assetId", "title", "version", "snippet"):
|
||
assert hits2[0][key]
|
||
assert hits2[0].get("chunkId") == "c0"
|