421 lines
22 KiB
Python
421 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""合同负路测试 + U0 端到端回放。
|
||
|
||
覆盖专题-09 §12 一级验收中可自动化的条目：
|
||
- 验收 4：诊断产物头缺项时修订拒绝执行
|
||
- 验收 6：事实快照缺失时 Patch 自动降为 Audit（无显式授权不得继续）
|
||
- 验收 7：成对选择模型与改写模型相同时视为阶段未执行
|
||
其余为合同负路（无发现禁改、唯一匹配、样例不齐拒绝激活、结构校验、事实增量、口癖保护）。
|
||
"""
|
||
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
# 内网环境不 pip install：直接把 src/ 加入导入路径
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
|
||
|
||
from deai import cards, diagnose, gates, load, pairwise, patch, report
|
||
from deai.pipeline import DowngradedToAudit, resolve_mode, run_audit, run_patch
|
||
|
||
|
||
class TestAssetCompleteness(unittest.TestCase):
|
||
"""规则库/样例库装载：active 规则必须配齐四类样例（专题-09 §4.1）。"""
|
||
|
||
def test_all_shipped_rules_load_with_complete_samples(self):
|
||
samples = load.load_samples()
|
||
rules = load.load_rules(samples=samples)
|
||
actives = load.active_rules(rules)
|
||
# 一级目标：至少 10 条 active 规则，每条四类样例齐全（装载器已强制）
|
||
self.assertGreaterEqual(len(actives), 10)
|
||
for rule in actives:
|
||
for stype in load.SAMPLE_TYPES:
|
||
self.assertTrue(rule["samples"][stype], f"{rule['id']} 缺 {stype}")
|
||
for sample_id in rule["samples"][stype]:
|
||
self.assertIn(rule["id"], samples[sample_id].get("rules", []))
|
||
|
||
def test_rule_without_four_sample_types_rejected(self):
|
||
samples = load.load_samples()
|
||
rules = load.load_rules(samples=samples)
|
||
broken = dict(rules["l002"])
|
||
broken["samples"] = dict(broken["samples"])
|
||
broken["samples"]["regression"] = [] # 缺回归陷阱样例
|
||
with self.assertRaises(load.LoadError):
|
||
load.check_activation(broken, samples)
|
||
|
||
def test_rule_referencing_missing_sample_rejected(self):
|
||
samples = load.load_samples()
|
||
rules = load.load_rules(samples=samples)
|
||
broken = dict(rules["l002"])
|
||
broken["samples"] = dict(broken["samples"])
|
||
broken["samples"]["regression"] = ["不存在的样例id"]
|
||
with self.assertRaises(load.LoadError):
|
||
load.check_activation(broken, samples)
|
||
|
||
|
||
class TestCaseCards(unittest.TestCase):
|
||
"""反向回填/创作反馈的案例卡：快采集，慢确认，禁止越过资产门。"""
|
||
|
||
def _capture(self, license_name="owned", label="sf", mode="backfill"):
|
||
return cards.capture_case_card(
|
||
card_id=f"card-{license_name}-{label}-{mode}",
|
||
label=label,
|
||
layer="lexical",
|
||
carrier="narration",
|
||
source_kind="existing_work" if mode == "backfill" else "creation_feedback",
|
||
source_license=license_name,
|
||
source_text="来源全文：值得注意的是，门外下雨了。",
|
||
excerpt="值得注意的是，门外下雨了。",
|
||
context="她抬头。值得注意的是，门外下雨了。",
|
||
location="chapter-1:paragraph-2",
|
||
pattern="无功能元话语",
|
||
rationale="待复核的表面模式观察",
|
||
function_check=["是否承担转折"],
|
||
risk_if_changed="可能误删节奏或信息",
|
||
capture_mode=mode,
|
||
)
|
||
|
||
def test_shipped_case_card_fixture_uses_current_schema(self):
|
||
loaded = cards.load_case_cards()
|
||
self.assertIn("card-backfill-l002-001", loaded)
|
||
self.assertEqual(loaded["card-backfill-l002-001"]["schema_version"], "ai-flavor-case-v1")
|
||
|
||
def test_backfill_starts_as_shadow_and_confirmation_is_explicit(self):
|
||
shadow = self._capture()
|
||
self.assertEqual(shadow["state"], "shadow")
|
||
with self.assertRaises(cards.CardError):
|
||
cards.promote_card_to_sample(shadow)
|
||
canonical = cards.confirm_case_card(
|
||
shadow, label="sf", review_note="作者确认：此处无功能"
|
||
)
|
||
sample = cards.promote_card_to_sample(canonical)
|
||
self.assertEqual(sample["case_card_id"], canonical["id"])
|
||
self.assertEqual(sample["type"], "sf")
|
||
|
||
def test_unlicensed_source_is_hash_only_and_cannot_be_confirmed(self):
|
||
card = self._capture(license_name="unauthorized")
|
||
self.assertEqual(card["excerpt"], "")
|
||
self.assertEqual(card["context"], "")
|
||
self.assertIn("excerpt_hash", card["source"])
|
||
with self.assertRaises(cards.CardError):
|
||
cards.confirm_case_card(card, label="sf", review_note="不应确认")
|
||
|
||
def test_hash_only_card_rejects_text_injection(self):
|
||
card = self._capture(license_name="research_only")
|
||
broken = dict(card)
|
||
broken["excerpt"] = "偷偷保存的第三方原文"
|
||
with self.assertRaises(cards.CardError):
|
||
cards.validate_card(broken)
|
||
|
||
def test_rule_candidate_is_always_candidate_and_needs_opposing_evidence(self):
|
||
sf = self._capture(label="sf")
|
||
snf = self._capture(label="snf")
|
||
candidate = cards.propose_rule_from_cards(
|
||
[sf, snf],
|
||
rule_id="candidate-l999",
|
||
name="待评估元话语",
|
||
layer="lexical",
|
||
carrier_scope="narration",
|
||
trigger={"type": "model_judgment", "criteria": "上下文功能判断"},
|
||
fix_hint="先仲裁再决定",
|
||
function_check=["是否承担叙事功能"],
|
||
)
|
||
self.assertEqual(candidate["status"], "candidate")
|
||
self.assertEqual(candidate["default_disposition"], "candidate")
|
||
# 候选规则可以先引用 shadow 卡，供后续人工确认；不得因此变 active。
|
||
load.check_case_card_refs(candidate, {sf["id"]: sf, snf["id"]: snf})
|
||
active = dict(candidate)
|
||
active["status"] = "active"
|
||
with self.assertRaises(load.LoadError):
|
||
load.check_case_card_refs(active, {sf["id"]: sf, snf["id"]: snf})
|
||
with self.assertRaises(cards.CardError):
|
||
cards.propose_rule_from_cards(
|
||
[sf],
|
||
rule_id="candidate-l998",
|
||
name="单样本禁令",
|
||
layer="lexical",
|
||
carrier_scope="narration",
|
||
trigger={"type": "model_judgment", "criteria": "读感"},
|
||
fix_hint="删除",
|
||
function_check=["是否有功能"],
|
||
)
|
||
|
||
def test_live_feedback_requires_feedback_source(self):
|
||
card = self._capture(mode="live_feedback")
|
||
self.assertEqual(card["source"]["kind"], "creation_feedback")
|
||
broken = dict(card)
|
||
broken["source"] = dict(card["source"])
|
||
broken["source"]["kind"] = "existing_work"
|
||
with self.assertRaises(cards.CardError):
|
||
cards.validate_card(broken)
|
||
|
||
|
||
class TestNegativePaths(unittest.TestCase):
|
||
"""一级验收 4/6/7 的负路（必须可测、必须红）。"""
|
||
|
||
def test_artifact_header_missing_blocks_patch(self):
|
||
# 验收 4：产物头缺项 → 视为诊断未发生
|
||
artifact = {"text_hash": "sha256:x", "findings": []} # 缺 mode/规则库版本
|
||
with self.assertRaises(diagnose.ArtifactIncomplete):
|
||
diagnose.validate_artifact(artifact)
|
||
|
||
def test_snapshot_missing_downgrades_to_audit(self):
|
||
# 验收 6：快照缺失且无授权 → Patch 强制降为 Audit
|
||
with self.assertRaises(DowngradedToAudit):
|
||
resolve_mode({"mode": "Patch", "author_approved_revision": True, "allowed_scope": "全文"}, fact_snapshot=None)
|
||
|
||
def test_snapshot_missing_with_explicit_authorization_proceeds(self):
|
||
mode = resolve_mode(
|
||
{"mode": "Patch", "author_approved_revision": True, "allowed_scope": "全文", "no_snapshot_authorization": True}, fact_snapshot=None)
|
||
self.assertEqual(mode, "Patch")
|
||
|
||
def test_same_model_pairwise_treated_as_not_executed(self):
|
||
# 验收 7：选择模型 == 改写模型 → 阶段 10 视为未执行
|
||
record = {"choice": "candidate", "rationale": "理由", "selection_model": "claude"}
|
||
with self.assertRaises(pairwise.PairwiseNotExecuted):
|
||
pairwise.validate_choice(record, rewrite_model="claude")
|
||
|
||
def test_pairwise_without_rationale_rejected(self):
|
||
record = {"choice": "candidate", "rationale": "", "selection_model": "gpt-5.6-sol"}
|
||
with self.assertRaises(pairwise.PairwiseNotExecuted):
|
||
pairwise.validate_choice(record, rewrite_model="claude")
|
||
|
||
|
||
class TestPatchContracts(unittest.TestCase):
|
||
def _finding(self, span):
|
||
return {
|
||
"id": "f1", "text_hash": "sha256:0000000000000000",
|
||
"rule_id": "l002", "rule_version": 1, "spans": [span],
|
||
"context_window": span, "layer": "lexical", "evidence": "测试",
|
||
"possible_function": "none", "confidence": "high",
|
||
"decision_proposal": "repair",
|
||
}
|
||
|
||
def _patch(self, finding_id="f1", original="值得注意的是，", replacement="", action="delete"):
|
||
return {"finding_id": finding_id, "action": action,
|
||
"original_exact": original, "replacement": replacement,
|
||
"rationale": "测试删除", "protected_invariants": []}
|
||
|
||
def test_patch_without_finding_rejected(self):
|
||
with self.assertRaises(patch.PatchError):
|
||
patch.apply_patches("值得注意的是，他来了。", [self._patch()], {})
|
||
|
||
def test_patch_nonunique_match_rejected(self):
|
||
text = "值得注意的是，值得注意的是，他来了。"
|
||
findings = {"f1": self._finding("值得注意的是")}
|
||
with self.assertRaises(patch.PatchError):
|
||
patch.apply_patches(text, [self._patch()], findings)
|
||
|
||
def test_patch_not_covering_span_rejected(self):
|
||
# original_exact 与发现 span 无关 → 禁止（防止借诊断之名改别处）
|
||
findings = {"f1": self._finding("他来了")}
|
||
with self.assertRaises(patch.PatchError):
|
||
patch.apply_patches("值得注意的是，他来了。", [self._patch()], findings)
|
||
|
||
def test_skip_patch_cannot_smuggle_replacement(self):
|
||
finding = self._finding("值得注意的是")
|
||
malicious = self._patch(action="skip", replacement="他突然笑了")
|
||
with self.assertRaises(patch.PatchError):
|
||
patch.apply_patches("值得注意的是，他来了。", [malicious], {"f1": finding})
|
||
|
||
def test_patch_does_not_normalize_unrelated_blank_lines(self):
|
||
finding = self._finding("值得注意的是")
|
||
text = "前文。\n\n\n值得注意的是，他来了。"
|
||
candidate, _ = patch.apply_patches(
|
||
text, [self._patch(original="值得注意的是，", replacement="")], {"f1": finding}
|
||
)
|
||
self.assertTrue(candidate.startswith("前文。\n\n\n"))
|
||
|
||
def test_adjacent_punctuation_extension_allowed(self):
|
||
# U0 修正 #6：span 允许紧邻标点的最小扩展（避免删除后标点粘连）
|
||
text = "他打量她，嘴角微微上扬。「姑娘」"
|
||
finding = self._finding("嘴角微微上扬")
|
||
p = self._patch(original="，嘴角微微上扬", replacement="")
|
||
candidate, _ = patch.apply_patches(text, [p], {"f1": finding})
|
||
self.assertEqual(candidate, "他打量她。「姑娘」")
|
||
|
||
|
||
class TestHardGate(unittest.TestCase):
|
||
def test_structural_verify_detects_outside_change(self):
|
||
original = "她关上门。值得注意的是，天黑了。"
|
||
patches = [{"finding_id": "f1", "action": "delete",
|
||
"original_exact": "值得注意的是，", "replacement": "",
|
||
"rationale": "测试", "protected_invariants": []}]
|
||
findings = {"f1": {"id": "f1", "spans": ["值得注意的是，"]}}
|
||
# 候选稿在 patch 之外还被偷偷改了一处 → 结构校验必须红
|
||
tampered = "她关上门。天黑了。另外多出一句。"
|
||
result = gates.structural_verify(original, tampered, patches, findings)
|
||
self.assertFalse(result["pass"])
|
||
|
||
def test_fact_delta_blocks_new_number(self):
|
||
patches = [{"finding_id": "f1", "action": "local_rewrite",
|
||
"original_exact": "值三百两银子", "replacement": "值三百五十两银子",
|
||
"rationale": "测试", "protected_invariants": []}]
|
||
result = gates.fact_delta(patches)
|
||
self.assertFalse(result["pass"])
|
||
|
||
def test_number_attribution_allows_approved_deletion(self):
|
||
# 数字随批准删除消失是合法的（U0 修正 #1：不做计数守恒）
|
||
original = "值三百两银子。值得一提的是，三天后开门。"
|
||
candidate = "值三百两银子。开门。"
|
||
patches = [{"finding_id": "f1", "action": "delete",
|
||
"original_exact": "值得一提的是，三天后", "replacement": "",
|
||
"rationale": "测试", "protected_invariants": []}]
|
||
result = gates.number_attribution(original, candidate, patches)
|
||
self.assertTrue(result["pass"], result["failures"])
|
||
|
||
def test_number_attribution_blocks_unexplained_disappearance(self):
|
||
original = "值三百两银子。"
|
||
candidate = "值银子。" # 数字凭空消失，无对应 patch
|
||
result = gates.number_attribution(original, candidate, [])
|
||
self.assertFalse(result["pass"])
|
||
|
||
def test_quote_tamper_fails(self):
|
||
# 引文/场内文本被篡改（未在任何批准删除片段内）→ 必须红
|
||
result = gates.quote_attribution(
|
||
"面板显示：【等级：待定】。", "面板显示：【等级：最高】。", [])
|
||
self.assertFalse(result["pass"])
|
||
|
||
def test_quote_duplicate_loss_cannot_hide_behind_one_deleted_quote(self):
|
||
patches = [{"finding_id": "f1", "action": "delete",
|
||
"original_exact": "前文【同一条】", "replacement": "",
|
||
"rationale": "测试", "protected_invariants": []}]
|
||
result = gates.quote_attribution(
|
||
"前文【同一条】。后文【同一条】。",
|
||
"前文。后文。",
|
||
patches,
|
||
)
|
||
self.assertFalse(result["pass"])
|
||
|
||
def test_quote_approved_deletion_passes(self):
|
||
# 引文整段位于批准删除片段内 → 合法消失
|
||
patches = [{"finding_id": "f1", "action": "delete",
|
||
"original_exact": "面板显示：【等级：待定】。", "replacement": "",
|
||
"rationale": "测试", "protected_invariants": []}]
|
||
result = gates.quote_attribution(
|
||
"前文。面板显示：【等级：待定】。后文。", "前文。后文。", patches)
|
||
self.assertTrue(result["pass"], result["failures"])
|
||
|
||
def test_protected_tic_deletion_fails(self):
|
||
voice_thin = {"untouchable_verbal_tics": {"老周": ["我说小子"]}}
|
||
result = gates.protected_checks(
|
||
"「我说小子，你来了。」", "「你来了。」", voice_thin)
|
||
self.assertFalse(result["pass"])
|
||
|
||
|
||
class TestU0Replay(unittest.TestCase):
|
||
"""端到端回放 U0 演练：真实规则库 + 真实目标文本 + 8 条 patch + 跨模型选择记录。"""
|
||
|
||
TEXT = """林晚儿站在万宝阁的柜台前，手指无意识地摩挲着那枚铜钱的边缘。
|
||
|
||
**那伙计上下打量了她一眼，**嘴角微微上扬。「姑娘，这东西值三百两银子，不是您一枚铜钱就能当的。」
|
||
|
||
值得注意的是，万宝阁在青云城立了三百年，从没人敢在这里讨价还价。研究表明，能进这种地方的修士，多半背景不凡。伙计眼中闪过一丝精光，显然已经把林晚儿的来历掂量了一遍。
|
||
|
||
「三百年？我说小子，你这铺子也就立了三百年，可我手里这枚铜钱——」老周把铜钱往柜台上一推，声音忽然压低，「它传了三千年。」
|
||
|
||
[占位：此处揭示铜钱来历]
|
||
|
||
万宝阁的大堂里安静了一瞬。系统面板的流光在半空停住，鉴定结果的金字明灭不定：
|
||
|
||
【物品：无法识别｜等级：待定｜建议操作：上报上峰】
|
||
|
||
无论是伙计的讥笑，还是看客的沉默，似乎都在这一刻定格了。三天后，这件事被摆进了执法堂的晨会。那时谁也不知道，这枚铜钱会掀翻整座青云城。也许这就是命运吧，总在人不经意的时候，悄悄安排好了一切。"""
|
||
|
||
PATCH_SPANS = [
|
||
# (original_exact, replacement, 对应发现 span 的识别片段)
|
||
("**那伙计上下打量了她一眼，**", "那伙计上下打量了她一眼，", "**那伙计"),
|
||
("，嘴角微微上扬", "", "嘴角微微上扬"),
|
||
("眼中闪过一丝精光，", "", "眼中闪过一丝精光"),
|
||
("研究表明，", "", "研究表明"),
|
||
("值得注意的是，", "", "值得注意的是"),
|
||
("无论是伙计的讥笑，还是看客的沉默，似乎都在这一刻定格了。", "", "无论是伙计的讥笑"),
|
||
("也许这就是命运吧，总在人不经意的时候，悄悄安排好了一切。", "", "也许这就是命运吧"),
|
||
("[占位：此处揭示铜钱来历]", "", "[占位：此处揭示铜钱来历]"),
|
||
]
|
||
|
||
def _build_artifact(self, rules):
|
||
artifact = run_audit(self.TEXT, rules, load.rule_library_version(rules))
|
||
artifact["mode"] = "Patch"
|
||
# 语义层发现由外部判定注入（U0 阶段 4 人工判定的代码化形态）
|
||
sem_finding = {
|
||
"id": "f-sem", "text_hash": artifact["text_hash"],
|
||
"rule_id": "sem001", "rule_version": 1,
|
||
"spans": ["也许这就是命运吧，总在人不经意的时候，悄悄安排好了一切。"],
|
||
"context_window": "那时谁也不知道…悄悄安排好了一切。",
|
||
"layer": "semantic",
|
||
"evidence": "段尾脱离场景的命运评注，钩子由前句承担",
|
||
"possible_function": "none", "confidence": "high",
|
||
"decision_proposal": "repair",
|
||
"arbitration_note": "五问皆否；前句为伏笔钩子，升华句稀释钩子",
|
||
}
|
||
diagnose.merge_model_findings(artifact, [sem_finding])
|
||
# 模拟仲裁完成：全部 repair（U0 阶段 4 结论）
|
||
for f in artifact["findings"]:
|
||
f["decision_proposal"] = "repair"
|
||
f["possible_function"] = f.get("possible_function", "none")
|
||
f.setdefault("arbitration_note", "演练回放：五问仲裁通过")
|
||
return artifact
|
||
|
||
def test_full_pipeline_replay(self):
|
||
samples = load.load_samples()
|
||
rules = load.load_rules(samples=samples)
|
||
artifact = self._build_artifact(rules)
|
||
diagnose.validate_artifact(artifact)
|
||
|
||
# patch 的 finding_id 按 span 覆盖关系解析（回放时不硬编码顺序）
|
||
patches = []
|
||
for original, replacement, marker in self.PATCH_SPANS:
|
||
fid = next(f["id"] for f in artifact["findings"] if marker in f["spans"][0]
|
||
or f["spans"][0] in original)
|
||
patches.append({"finding_id": fid, "action": "delete" if not replacement else "patch",
|
||
"original_exact": original, "replacement": replacement,
|
||
"rationale": "U0 回放", "protected_invariants": []})
|
||
|
||
result = run_patch(
|
||
self.TEXT, rules, artifact, patches,
|
||
task_contract={"mode": "Patch", "allowed_scope": "全文", "author_approved_revision": True},
|
||
fact_snapshot={"entities": ["林晚儿", "老周", "伙计", "万宝阁", "青云城", "执法堂"]},
|
||
voice_thin={"untouchable_verbal_tics": {"老周": ["我说小子"]},
|
||
"protected_spans": ["【物品：无法识别｜等级：待定｜建议操作：上报上峰】"]},
|
||
rewrite_model="claude",
|
||
pairwise_record={"choice": "candidate",
|
||
"rationale": "保真完整，钩子有力（U0 gpt-5.6-sol 判定）",
|
||
"selection_model": "gpt-5.6-sol"},
|
||
rule_library_version=load.rule_library_version(rules))
|
||
|
||
# 硬门全绿（结构/事实增量/数字归因/保护项）
|
||
self.assertTrue(result["hard_gate"]["pass"], result["hard_gate"])
|
||
# 复扫零残留
|
||
self.assertTrue(result["regression_gate"]["pass"],
|
||
result["regression_gate"]["residual_hits"])
|
||
# 跨模型选择成立
|
||
self.assertEqual(result["pairwise_choice"]["choice"], "candidate")
|
||
# 候选稿不再含任何确定性命中点
|
||
for marker in ("**", "值得注意的是", "研究表明", "嘴角微微上扬", "[占位", "也许这就是命运"):
|
||
self.assertNotIn(marker, result["candidate_text"])
|
||
# 保护项仍在
|
||
self.assertIn("我说小子", result["candidate_text"])
|
||
self.assertIn("【物品：无法识别｜等级：待定｜建议操作：上报上峰】", result["candidate_text"])
|
||
|
||
# 审计报告可组装且过合同（含禁止总分检查）
|
||
rep = report.assemble(
|
||
artifact, patches, result["candidate_text"], result["hard_gate"],
|
||
result["voice_gate"], result["regression_gate"], result["pairwise_choice"],
|
||
unresolved_risks=result["hard_gate"]["unverified"])
|
||
self.assertIn("pairwise_choice", rep["final"])
|
||
|
||
def test_report_rejects_score_fields(self):
|
||
samples = load.load_samples()
|
||
rules = load.load_rules(samples=samples)
|
||
artifact = run_audit(self.TEXT, rules, load.rule_library_version(rules))
|
||
with self.assertRaises(report.ForbiddenScoreError):
|
||
report.assemble(
|
||
artifact, [], None,
|
||
{"pass": True, "checks": {}, "unverified": [], "human_score": 0.9},
|
||
{"status": "unverified", "pass": None, "note": ""},
|
||
{"pass": True, "residual_hits": []}, None, [])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|