265 lines
13 KiB
Python
265 lines
13 KiB
Python
"""v7 机制第 7 项 user_side_items 定向测试。
|
||
|
||
钉住的验收(只测本项,不碰其它 7 项):
|
||
- B/C 全 pending 时:7 维 overall 与代码表仍可 pass,
|
||
生产表强制 not_pass,原因明确指向用户侧。
|
||
- user_side_items 为空/缺失时:标 user_side_unknown=True,
|
||
绝不写成用户侧已完成。
|
||
- 非法项不静默丢弃、不编造状态:记入 invalid_items。
|
||
- 阈值仍为 test_default_unapproved;本项不做 B/C 拍板。
|
||
"""
|
||
|
||
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.audit_scorecard import (
|
||
DEFAULT_THRESHOLDS, USER_SIDE_ITEM_IDS,
|
||
apply_user_side_three_table_rule, build_scorecard,
|
||
normalize_user_side_items)
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.audit_scorecard import (
|
||
DEFAULT_THRESHOLDS, USER_SIDE_ITEM_IDS,
|
||
apply_user_side_three_table_rule, build_scorecard,
|
||
normalize_user_side_items)
|
||
|
||
|
||
def _full_evidence():
|
||
return {
|
||
"delivery_quality": {
|
||
"first_pass_rate": 0.96, "defect_rate": 0.01,
|
||
"rework_rate": 0.04, "first_pass_known": True,
|
||
"defect_known": True, "rework_known": True,
|
||
"attempts_present": True, "evidence_kind": "real",
|
||
},
|
||
"gate_audit": {
|
||
"hard_gate_enforced": True, "gate_items_passed": 5,
|
||
"gate_items_total": 5, "merge_blocked_verified": True,
|
||
"loop_end_to_end_verified": True,
|
||
"sampling_misjudgment_rate": 0.0,
|
||
},
|
||
"arch_contract": {
|
||
"contract_frozen": True, "arch_violations": 0,
|
||
"directory_violations": 0, "import_violations": 0,
|
||
"stack_changed_without_approval": False,
|
||
},
|
||
"test_coverage": {
|
||
"prd_coverage_rate": 1.0, "cross_matrix_coverage_rate": 1.0,
|
||
"coverage_pct": 0.92, "cross_matrix_kind": "real",
|
||
"real_pg": True, "real_service": True,
|
||
"agent_six_complete": True,
|
||
"lint_items": 0, "mypy_items": 0, "bandit_high": 0,
|
||
},
|
||
"security_ledger": {
|
||
"ledger_verify_ok": True, "ledger_coverage_rate": 1.0,
|
||
"branch_protection": "protected",
|
||
"shell_whitelist_pass_rate": 1.0, "anchor_check_recent": True,
|
||
},
|
||
"cost_quota": {
|
||
"cost_records_complete": True, "unpriced_as_zero": False,
|
||
"per_creator_present": True, "quota_cutoff_verified": True,
|
||
},
|
||
"ops_scale": {
|
||
"ci_green_rate": 1.0, "deploy_success_rate": 1.0,
|
||
"rollback_drill_done": True, "ha_status": "second_host",
|
||
"parallel_scale_proven": 40,
|
||
},
|
||
}
|
||
|
||
|
||
# v7 审计 B/C 清单全 pending 证据(blocked_by 按阻塞方示例分三组)。
|
||
def _all_bc_pending():
|
||
blocked_by = {
|
||
"B1": "user", "B2": "user", "B3": "contract_decision",
|
||
"B4": "user", "B5": "host_env", "B6": "host_env",
|
||
"B7": "user", "B8": "contract_decision", "B9": "user",
|
||
"C1": "contract_decision", "C2": "host_env",
|
||
"C3": "contract_decision", "C4": "contract_decision",
|
||
}
|
||
return [{"id": item_id, "status": "pending",
|
||
"blocked_by": blocked_by[item_id]}
|
||
for item_id in USER_SIDE_ITEM_IDS]
|
||
|
||
|
||
class UserSideItemsTestCase(unittest.TestCase):
|
||
def test_bc_list_is_the_v7_audit_set(self):
|
||
self.assertEqual(USER_SIDE_ITEM_IDS,
|
||
["B1", "B2", "B3", "B4", "B5", "B6", "B7", "B8",
|
||
"B9", "C1", "C2", "C3", "C4"])
|
||
self.assertEqual(len(_all_bc_pending()), 13)
|
||
|
||
def test_bc_all_pending_code_passes_production_not_pass(self):
|
||
ev = _full_evidence()
|
||
ev["user_side_items"] = _all_bc_pending()
|
||
report = build_scorecard(ev)
|
||
# 验收核心:代码表仍可 pass,生产表 not_pass 且原因指向用户侧。
|
||
self.assertEqual(report["overall"], "pass")
|
||
self.assertEqual(report["code_overall"], "pass")
|
||
self.assertTrue(report["code_unchanged_by_user_side"])
|
||
self.assertEqual(report["production_overall"], "not_pass")
|
||
self.assertEqual(
|
||
report["production_reason"],
|
||
"user_side_pending: " + ",".join(USER_SIDE_ITEM_IDS))
|
||
self.assertTrue(report["user_side_forced_production_not_pass"])
|
||
us = report["user_side_items"]
|
||
self.assertEqual(us["pending_ids"], USER_SIDE_ITEM_IDS)
|
||
self.assertEqual(us["done_ids"], [])
|
||
self.assertEqual(us["blocked_ids"], [])
|
||
self.assertTrue(us["all_pending"])
|
||
self.assertFalse(us["user_side_unknown"])
|
||
self.assertEqual(us["invalid_items"], [])
|
||
self.assertEqual(us["count"], 13)
|
||
# blocked_by 汇总:三组阻塞方覆盖全部 13 个 id,不多不少。
|
||
grouped = sorted(i for ids in us["blocked_by_summary"].values()
|
||
for i in ids)
|
||
self.assertEqual(grouped, USER_SIDE_ITEM_IDS)
|
||
self.assertEqual(us["blocked_by_summary"]["user"],
|
||
["B1", "B2", "B4", "B7", "B9"])
|
||
|
||
def test_kwarg_user_side_items_overrides_evidence_key(self):
|
||
report = build_scorecard(_full_evidence(),
|
||
user_side_items=_all_bc_pending())
|
||
self.assertEqual(report["code_overall"], "pass")
|
||
self.assertEqual(report["production_overall"], "not_pass")
|
||
self.assertEqual(report["user_side_items"]["count"], 13)
|
||
|
||
def test_passthrough_preserves_extra_fields_verbatim(self):
|
||
items = [{"id": "B1", "status": "pending", "blocked_by": "user",
|
||
"note": "branch protection not yet enabled"},
|
||
{"id": "B3", "status": "done", "blocked_by": None}]
|
||
report = build_scorecard(_full_evidence(), user_side_items=items)
|
||
self.assertEqual(report["user_side_items"]["items"], items)
|
||
self.assertEqual(report["user_side_items"]["pending_ids"], ["B1"])
|
||
self.assertEqual(report["user_side_items"]["done_ids"], ["B3"])
|
||
|
||
def test_empty_list_is_unknown_not_done(self):
|
||
report = build_scorecard(_full_evidence(), user_side_items=[])
|
||
self.assertTrue(report["user_side_unknown"])
|
||
self.assertTrue(report["user_side_items"]["user_side_unknown"])
|
||
self.assertEqual(report["code_overall"], "pass")
|
||
self.assertEqual(report["production_overall"], "not_pass")
|
||
self.assertIn("user_side_unknown", report["production_reason"])
|
||
# 否定表述本身含 user_side_done 一词,故断言否定语义而非子串缺席。
|
||
self.assertTrue(report["production_reason"].startswith(
|
||
"user_side_unknown"))
|
||
self.assertIn("not equivalent to user_side_done",
|
||
report["production_reason"])
|
||
self.assertEqual(report["user_side_items"]["items"], [])
|
||
self.assertEqual(report["user_side_items"]["pending_ids"], [])
|
||
|
||
def test_missing_key_is_unknown_not_done(self):
|
||
report = build_scorecard(_full_evidence())
|
||
self.assertTrue(report["user_side_unknown"])
|
||
self.assertEqual(report["production_overall"], "not_pass")
|
||
self.assertIn("user_side_unknown", report["production_reason"])
|
||
self.assertNotIn("done", report["production_reason"].split(":")[0])
|
||
|
||
def test_invalid_items_are_recorded_not_silently_dropped(self):
|
||
items = [
|
||
{"id": "B1", "status": "pending"}, # 缺 blocked_by
|
||
{"status": "pending", "blocked_by": "user"}, # 缺 id
|
||
{"id": "B2", "status": "whatever", "blocked_by": "user"},
|
||
{"id": "B3", "status": "pending", "blocked_by": "user"},
|
||
{"id": "B3", "status": "pending", "blocked_by": "user"},
|
||
"not-a-dict",
|
||
]
|
||
norm = normalize_user_side_items(items)
|
||
self.assertEqual(norm["items"],
|
||
[{"id": "B3", "status": "pending",
|
||
"blocked_by": "user"}])
|
||
reasons = [inv["reason"] for inv in norm["invalid_items"]]
|
||
self.assertEqual(reasons,
|
||
["blocked_by_missing", "id_missing",
|
||
"status_missing_or_unknown", "duplicate_id",
|
||
"item_not_a_dict"])
|
||
self.assertEqual(len(norm["invalid_items"]), 5)
|
||
# 非法项不编造状态:B1/B2 绝不在 items/pending_ids 里。
|
||
self.assertNotIn("B1", norm["pending_ids"])
|
||
self.assertNotIn("B2", norm["pending_ids"])
|
||
report = build_scorecard(_full_evidence(), user_side_items=items)
|
||
self.assertEqual(report["code_overall"], "pass")
|
||
self.assertEqual(report["production_overall"], "not_pass")
|
||
self.assertIn("user_side_invalid", report["production_reason"])
|
||
self.assertIn("user_side_pending: B3", report["production_reason"])
|
||
|
||
def test_non_list_user_side_is_invalid_not_crash(self):
|
||
norm = normalize_user_side_items("B1")
|
||
self.assertEqual(norm["items"], [])
|
||
self.assertEqual(norm["invalid_items"][0]["reason"],
|
||
"user_side_items_not_a_list")
|
||
self.assertFalse(norm["user_side_unknown"])
|
||
report = build_scorecard(_full_evidence(), user_side_items="B1")
|
||
self.assertEqual(report["production_overall"], "not_pass")
|
||
self.assertIn("user_side_invalid", report["production_reason"])
|
||
|
||
def test_mixed_pending_and_done_keeps_code_pass(self):
|
||
items = [{"id": "B1", "status": "done", "blocked_by": None},
|
||
{"id": "B2", "status": "pending", "blocked_by": "user"},
|
||
{"id": "C1", "status": "blocked",
|
||
"blocked_by": "contract_decision"}]
|
||
report = build_scorecard(_full_evidence(), user_side_items=items)
|
||
self.assertEqual(report["code_overall"], "pass")
|
||
self.assertEqual(report["production_overall"], "not_pass")
|
||
# 原因只列未完成项,不把已 done 的 B1 写进 pending。
|
||
self.assertEqual(report["production_reason"],
|
||
"user_side_pending: B2,C1")
|
||
self.assertEqual(report["user_side_items"]["blocked_ids"], ["C1"])
|
||
self.assertFalse(report["user_side_items"]["all_pending"])
|
||
self.assertFalse(report["user_side_items"]["all_done"])
|
||
|
||
def test_all_done_production_follows_its_own_overall(self):
|
||
items = [{"id": item_id, "status": "done", "blocked_by": None}
|
||
for item_id in USER_SIDE_ITEM_IDS]
|
||
report = build_scorecard(_full_evidence(), user_side_items=items)
|
||
self.assertTrue(report["user_side_items"]["all_done"])
|
||
self.assertEqual(report["code_overall"], "pass")
|
||
self.assertEqual(report["production_overall"], "pass")
|
||
self.assertIsNone(report["production_reason"])
|
||
self.assertFalse(report["user_side_forced_production_not_pass"])
|
||
|
||
def test_standalone_rule_never_changes_code_overall(self):
|
||
rule = apply_user_side_three_table_rule(
|
||
"pass", "pass", _all_bc_pending())
|
||
self.assertEqual(rule["code_overall"], "pass")
|
||
self.assertEqual(rule["production_overall"], "not_pass")
|
||
self.assertEqual(rule["production_reason"],
|
||
"user_side_pending: " + ",".join(USER_SIDE_ITEM_IDS))
|
||
# 代码表本身不达标时,用户侧全 done 也不把它抬高。
|
||
rule2 = apply_user_side_three_table_rule(
|
||
"not_pass", "not_pass",
|
||
[{"id": "B1", "status": "done", "blocked_by": None}])
|
||
self.assertEqual(rule2["code_overall"], "not_pass")
|
||
self.assertEqual(rule2["production_overall"], "not_pass")
|
||
self.assertFalse(rule2["user_side_forced_production_not_pass"])
|
||
# dict 输入同样只取 overall,不改代码表结论。
|
||
rule3 = apply_user_side_three_table_rule(
|
||
{"overall": "pass"}, {"overall": "pass"}, [])
|
||
self.assertEqual(rule3["code_overall"], "pass")
|
||
self.assertTrue(rule3["user_side_unknown"])
|
||
|
||
def test_thresholds_still_unapproved_no_decision_fabricated(self):
|
||
for entry in DEFAULT_THRESHOLDS.values():
|
||
self.assertEqual(entry["source"], "test_default_unapproved")
|
||
report = build_scorecard(_full_evidence(),
|
||
user_side_items=_all_bc_pending())
|
||
self.assertEqual(report["thresholds_source"],
|
||
"test_default_unapproved")
|
||
# 7 维 overall 本身仍只由 7 维决定:用户侧项不进 dimensions。
|
||
self.assertEqual(len(report["dimensions"]), 7)
|
||
self.assertEqual([d["id"] for d in report["dimensions"]],
|
||
["delivery_quality", "gate_audit", "arch_contract",
|
||
"test_coverage", "security_ledger", "cost_quota",
|
||
"ops_scale"])
|
||
|
||
def test_input_items_not_mutated(self):
|
||
items = _all_bc_pending()
|
||
snapshot = [dict(it) for it in items]
|
||
build_scorecard(_full_evidence(), user_side_items=items)
|
||
self.assertEqual(items, snapshot)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|