327 lines
12 KiB
Python
327 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""正文 Gate A/B 逐样本唯一判定器的表驱动测试。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import pathlib
|
||
import sys
|
||
import unittest
|
||
|
||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
|
||
|
||
from writer_gate import ( # noqa: E402
|
||
CONFOUND_LABELS,
|
||
REQUIRED_SCENARIOS,
|
||
decide_gate_a,
|
||
decide_gate_b,
|
||
)
|
||
from writer_rubric import DIMENSIONS # noqa: E402
|
||
|
||
|
||
SCENARIOS = tuple(sorted(REQUIRED_SCENARIOS))
|
||
|
||
|
||
def scores(
|
||
*,
|
||
b_fidelity_delta: float = 0.0,
|
||
c_fidelity_delta: float = 0.5,
|
||
) -> dict[str, dict[str, float]]:
|
||
"""构造三臂五维终分，分别控制 B-A 与 C-A 的保真增量。"""
|
||
|
||
a_scores = {dimension: 7.0 for dimension in DIMENSIONS}
|
||
b_scores = dict(a_scores)
|
||
c_scores = dict(a_scores)
|
||
b_scores["setting_entity_fidelity"] += b_fidelity_delta
|
||
c_scores["setting_entity_fidelity"] += c_fidelity_delta
|
||
return {"A": a_scores, "B": b_scores, "C": c_scores}
|
||
|
||
|
||
def sample(
|
||
number: int,
|
||
*,
|
||
work_id: int = 8,
|
||
scenario: str | None = None,
|
||
**overrides,
|
||
) -> dict[str, object]:
|
||
"""构造合法的逐样本脱敏结果。"""
|
||
|
||
result: dict[str, object] = {
|
||
"sampleId": f"{work_id}-{number}",
|
||
"workId": work_id,
|
||
"scenario": scenario or SCENARIOS[(number - 1) % len(SCENARIOS)],
|
||
"schemaValid": True,
|
||
"futureLeakage": False,
|
||
"systemFailure": False,
|
||
"reviewStatus": "stable_report",
|
||
"cArm": {
|
||
"hardConstraintCoverage": 1.0,
|
||
"highSeverityResidualCount": 0,
|
||
},
|
||
"scores": scores(),
|
||
"confounders": {category: [] for category in CONFOUND_LABELS},
|
||
}
|
||
result.update(overrides)
|
||
return result
|
||
|
||
|
||
def gate_a_input(samples: list[dict[str, object]], **forged) -> dict[str, object]:
|
||
"""构造 Gate A 输入；forged 用于证明顶层聚合值无效。"""
|
||
|
||
return {"gate": "A", "samples": samples, **forged}
|
||
|
||
|
||
def gate_b_input(samples: list[dict[str, object]], **overrides) -> dict[str, object]:
|
||
"""构造 Gate B 输入。"""
|
||
|
||
return {"gate": "B", "gateAStatus": "passed", "samples": samples, **overrides}
|
||
|
||
|
||
def ten_samples() -> list[dict[str, object]]:
|
||
"""构造两书各五章并覆盖五类场景的 Gate B 基线。"""
|
||
|
||
return [
|
||
sample(index + 1, work_id=work_id, scenario=SCENARIOS[index])
|
||
for work_id in (8, 4)
|
||
for index in range(5)
|
||
]
|
||
|
||
|
||
class GateATest(unittest.TestCase):
|
||
"""验证 Gate A 的逐样本统计与唯一短路顺序。"""
|
||
|
||
def test_table_driven_terminal_priority(self):
|
||
base = [sample(index) for index in range(1, 6)]
|
||
cases = []
|
||
|
||
insufficient = copy.deepcopy(base[:4])
|
||
insufficient[0]["futureLeakage"] = True
|
||
cases.append(("不足优先于泄露", insufficient, "insufficient_evidence"))
|
||
|
||
for label, field, value in (
|
||
("schema 非法", "schemaValid", False),
|
||
("未来泄露", "futureLeakage", True),
|
||
("系统失败", "systemFailure", True),
|
||
):
|
||
samples = copy.deepcopy(base)
|
||
samples.append(sample(6, **{field: value}))
|
||
cases.append((label, samples, "failed"))
|
||
|
||
high_severity = copy.deepcopy(base)
|
||
high_severity[0]["cArm"]["highSeverityResidualCount"] = 1
|
||
cases.append(("C 高严重度", high_severity, "failed"))
|
||
|
||
low_coverage = copy.deepcopy(base)
|
||
low_coverage[0]["cArm"]["hardConstraintCoverage"] = 0.99
|
||
cases.append(("覆盖不足", low_coverage, "failed"))
|
||
cases.append(("全部通过", base, "passed"))
|
||
|
||
for label, samples, expected in cases:
|
||
with self.subTest(label=label):
|
||
self.assertEqual(decide_gate_a(gate_a_input(samples))["status"], expected)
|
||
|
||
def test_pre_c_arm_failures_do_not_require_c_arm_results(self):
|
||
"""前置机械或系统失败没有 C 臂结果时仍按唯一顺序裁决。"""
|
||
|
||
base = [sample(index) for index in range(1, 6)]
|
||
cases = []
|
||
for label, field, value in (
|
||
("schema 非法", "schemaValid", False),
|
||
("未来泄露", "futureLeakage", True),
|
||
("系统失败", "systemFailure", True),
|
||
):
|
||
failed_sample = sample(6, **{field: value})
|
||
failed_sample.pop("cArm")
|
||
cases.append((f"五有效加{label}", [*copy.deepcopy(base), failed_sample], "failed"))
|
||
cases.append((f"四有效加{label}", [*copy.deepcopy(base[:4]), failed_sample], "insufficient_evidence"))
|
||
|
||
for label, samples, expected in cases:
|
||
with self.subTest(label=label):
|
||
self.assertEqual(decide_gate_a(gate_a_input(samples))["status"], expected)
|
||
|
||
def test_forged_upstream_aggregates_cannot_change_result(self):
|
||
forged = gate_a_input(
|
||
[sample(index) for index in range(1, 5)],
|
||
validSampleCount=999,
|
||
invalidSchemaCount=0,
|
||
futureLeakageCount=0,
|
||
cArmHardConstraintCoverage=1.0,
|
||
)
|
||
report = decide_gate_a(forged)
|
||
self.assertEqual(report["status"], "insufficient_evidence")
|
||
self.assertEqual(report["metrics"]["validSampleCount"], 4)
|
||
|
||
def test_valid_sample_boundary_counts_each_sample_once(self):
|
||
"""四个有效样本仍不足，补到五个后才进入硬门判断。"""
|
||
|
||
four_valid = [sample(index) for index in range(1, 5)]
|
||
multi_failed = sample(
|
||
5,
|
||
schemaValid=False,
|
||
futureLeakage=True,
|
||
systemFailure=True,
|
||
reviewStatus="invalid_unstable",
|
||
)
|
||
multi_failed.pop("cArm")
|
||
report = decide_gate_a(gate_a_input([*four_valid, multi_failed]))
|
||
self.assertEqual(report["status"], "insufficient_evidence")
|
||
self.assertEqual(report["metrics"]["sampleCount"], 5)
|
||
self.assertEqual(report["metrics"]["validSampleCount"], 4)
|
||
|
||
report = decide_gate_a(
|
||
gate_a_input([*four_valid, sample(6), multi_failed])
|
||
)
|
||
self.assertEqual(report["status"], "failed")
|
||
self.assertEqual(report["metrics"]["validSampleCount"], 5)
|
||
|
||
def test_five_confounder_categories_are_aggregated_per_sample(self):
|
||
samples = [sample(index) for index in range(1, 7)]
|
||
categories = list(CONFOUND_LABELS)
|
||
for index, category in enumerate(categories):
|
||
samples[index]["confounders"][category] = [{"code": f"case-{index}"}]
|
||
samples[5]["futureLeakage"] = True
|
||
samples[4]["reviewStatus"] = "invalid_unstable"
|
||
|
||
report = decide_gate_a(
|
||
gate_a_input(samples, confounds={"falseNegatives": ["伪造聚合"]})
|
||
)
|
||
self.assertEqual(set(report["confounds"]), set(CONFOUND_LABELS))
|
||
self.assertNotIn("伪造聚合", str(report["confounds"]))
|
||
for index, category in enumerate(categories):
|
||
self.assertTrue(
|
||
any(item["sampleId"] == f"8-{index + 1}" for item in report["confounds"][category])
|
||
)
|
||
self.assertTrue(
|
||
any(item["sampleId"] == "8-6" for item in report["confounds"]["leakage"])
|
||
)
|
||
|
||
|
||
class GateBTest(unittest.TestCase):
|
||
"""验证 Gate B 的充分性、退化与增益短路顺序。"""
|
||
|
||
def test_gate_a_terminal_has_highest_priority(self):
|
||
insufficient = [sample(index) for index in range(1, 5)]
|
||
report = decide_gate_b(gate_b_input(insufficient, gateAStatus="passed"))
|
||
self.assertEqual(report["status"], "insufficient_evidence")
|
||
self.assertEqual(report["metrics"]["recomputedGateAStatus"], "insufficient_evidence")
|
||
|
||
failed = [sample(index) for index in range(1, 6)]
|
||
failed.append(sample(6, futureLeakage=True))
|
||
report = decide_gate_b(gate_b_input(failed, gateAStatus="passed"))
|
||
self.assertEqual(report["status"], "failed")
|
||
self.assertEqual(report["metrics"]["recomputedGateAStatus"], "failed")
|
||
|
||
def test_forged_gate_a_status_cannot_override_recomputed_result(self):
|
||
"""顶层 Gate A 终态只是非可信输入，不能放行或阻断同批样本。"""
|
||
|
||
passed = decide_gate_b(gate_b_input(ten_samples(), gateAStatus="failed"))
|
||
self.assertEqual(passed["status"], "passed")
|
||
self.assertEqual(passed["metrics"]["recomputedGateAStatus"], "passed")
|
||
|
||
insufficient = decide_gate_b(
|
||
gate_b_input([sample(index) for index in range(1, 5)], gateAStatus="passed")
|
||
)
|
||
self.assertEqual(insufficient["status"], "insufficient_evidence")
|
||
|
||
def test_gate_a_short_circuit_keeps_sample_confounds_without_changing_status(self):
|
||
"""Gate A 终态优先，但合法逐样本混淆项仍应进入 Gate B 报告。"""
|
||
|
||
valid = [sample(index) for index in range(1, 6)]
|
||
leaked = sample(6, futureLeakage=True)
|
||
leaked.pop("cArm")
|
||
leaked["confounders"]["falsePositives"] = ["detector 误报候选"]
|
||
report = decide_gate_b(
|
||
{
|
||
"gateAStatus": "failed",
|
||
"samples": [*valid, leaked],
|
||
"confounds": {"falsePositives": ["伪造聚合"]},
|
||
}
|
||
)
|
||
self.assertEqual(report["status"], "failed")
|
||
self.assertIn("detector 误报候选", str(report["confounds"]["falsePositives"]))
|
||
self.assertTrue(report["confounds"]["leakage"])
|
||
self.assertNotIn("伪造聚合", str(report["confounds"]))
|
||
|
||
def test_sample_sufficiency_precedes_quality(self):
|
||
samples = ten_samples()[:-1]
|
||
for item in samples[:3]:
|
||
item["scores"]["C"]["style_consistency"] = 6.0
|
||
self.assertEqual(decide_gate_b(gate_b_input(samples))["status"], "insufficient_evidence")
|
||
|
||
unstable = ten_samples()
|
||
for index in range(3):
|
||
unstable[index]["reviewStatus"] = "invalid_unstable"
|
||
unstable[index].pop("scores")
|
||
self.assertEqual(
|
||
decide_gate_b(gate_b_input(unstable))["status"], "insufficient_evidence"
|
||
)
|
||
|
||
def test_missing_scene_or_single_work_is_insufficient(self):
|
||
missing_scene = ten_samples()
|
||
for item in missing_scene:
|
||
if item["scenario"] == "returning_character":
|
||
item["scenario"] = "battle"
|
||
self.assertEqual(
|
||
decide_gate_b(gate_b_input(missing_scene))["status"],
|
||
"insufficient_evidence",
|
||
)
|
||
one_work = [sample(index) for index in range(1, 11)]
|
||
self.assertEqual(
|
||
decide_gate_b(gate_b_input(one_work))["status"], "insufficient_evidence"
|
||
)
|
||
|
||
def test_hard_failure_and_quality_regression_fail(self):
|
||
low_coverage = ten_samples()
|
||
low_coverage[0]["cArm"]["hardConstraintCoverage"] = 0.9
|
||
self.assertEqual(decide_gate_b(gate_b_input(low_coverage))["status"], "failed")
|
||
|
||
decline = ten_samples()
|
||
for item in decline[:3]:
|
||
item["scores"]["C"]["style_consistency"] = 6.0
|
||
report = decide_gate_b(
|
||
gate_b_input(
|
||
decline,
|
||
anyDimensionDeclineOverHalfRatio=0.0,
|
||
averageDeltas={"setting_entity_fidelity": 999},
|
||
)
|
||
)
|
||
self.assertEqual(report["status"], "failed")
|
||
self.assertEqual(report["metrics"]["anyDimensionDeclineOverHalfRatio"], 0.3)
|
||
|
||
def test_gain_threshold_distinguishes_passed_and_no_gain(self):
|
||
self.assertEqual(decide_gate_b(gate_b_input(ten_samples()))["status"], "passed")
|
||
no_gain = ten_samples()
|
||
for item in no_gain:
|
||
item["scores"] = scores(c_fidelity_delta=0.0)
|
||
self.assertEqual(decide_gate_b(gate_b_input(no_gain))["status"], "no_gain")
|
||
|
||
def test_stable_sample_requires_complete_a_b_c_scores(self):
|
||
missing_b = ten_samples()
|
||
missing_b[0]["scores"].pop("B")
|
||
with self.assertRaisesRegex(ValueError, "精确包含 A/B/C"):
|
||
decide_gate_b(gate_b_input(missing_b))
|
||
|
||
def test_b_minus_a_is_diagnostic_and_c_minus_a_drives_gate(self):
|
||
"""即使 B-A 明显退化，只要 C-A 达标，正式 Gate B 仍按 C-A 通过。"""
|
||
|
||
samples = ten_samples()
|
||
for item in samples:
|
||
item["scores"] = scores(
|
||
b_fidelity_delta=-2.0,
|
||
c_fidelity_delta=0.5,
|
||
)
|
||
report = decide_gate_b(gate_b_input(samples))
|
||
self.assertEqual(report["status"], "passed")
|
||
self.assertEqual(
|
||
report["metrics"]["averageDeltas"]["B-A"]["setting_entity_fidelity"],
|
||
-2.0,
|
||
)
|
||
self.assertEqual(
|
||
report["metrics"]["averageDeltas"]["C-A"]["setting_entity_fidelity"],
|
||
0.5,
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|