feat: 7-dim project audit scorecard (all >=9 to pass, average never a pass criterion)
- cloud/audit_scorecard.py: pure stdlib scorer, structured evidence in, 7 dimension scores (0-10, 1 decimal), basis, missing/untested items, overall pass only when all 7 >= 9; unknown -> None, never 0/10/estimated; fixture/viewport/code-review evidence -> untested, never pass; thresholds all test_default_unapproved, user_approved only on explicit input. - server.py: GET /audit-scorecard read-only, same auth/member scope as /metrics, no DB writes, no ledger writes, no pushback/reaudit triggers; on main base it honestly reports unknown/not-enforced, overall not_pass. - docs/audit-scorecard.md: spec, formulas, threshold table, unknown handling, overall rule, 10-to-7 merge mapping, relations to #238 metrics-report, #244 sampling, #260/#261 loop, and a system-computed example (current state, overall not_pass) clearly separated from the previous judgement estimates. - tests: 12 new items pin all-pass, one-below-nine fails, average 9.1 with one 4 still fails, unknown None, fixture untested, conditional not a pass, thresholds unapproved, endpoint 401/200 read-only. - suites: cloud 1442 OK (skipped 4), top-level 105 OK, deploy 39 OK.
This commit is contained in:
1 parent
b36f4adc58
commit
e5d570c233
4 files changed
+941
No files matched your search
@@ -0,0 +1,516 @@
|
||||
"""7 维项目级审计评分卡(纯函数、标准库 only)。
|
||||
|
||||
定位:这是一个评分卡,不是门禁、不是审计、不是自动循环。
|
||||
它不重写 metrics、arch_conformance、ledger、audit_loop 的逻辑,
|
||||
只接受调用方显式传入的结构化证据,算出 7 维各自的分数。
|
||||
|
||||
计分规则(钉死,不可改为平均分达标):
|
||||
- 每维分数 0-10,保留一位小数;>=9 通过,<9 不通过。
|
||||
- 总体结论:7 维全部 >=9 才算达标(overall="pass")。
|
||||
任一维 <9,总体就不达标(overall="not_pass")。
|
||||
- 平均分只作参考信息(average_score),绝不能作为达标依据。
|
||||
代码里把“平均分达标”写成明确的不通过路径并加测试钉住:
|
||||
例如 6 维 10 分、1 维 4 分,平均 9.1 分,总体仍是不达标。
|
||||
- 未知处理:任何一维的数据源缺失、未知、未验证,
|
||||
该维分数记为 None(未知),不记 0、不记 10、不估算,
|
||||
总体直接不达标,并列出缺失项(missing_items)。
|
||||
“没数据”既不算通过,也不算 0 分去拉低平均分后又用平均分糊弄。
|
||||
- 未实测处理:调用方传入“夹具模拟”“视口替代”“未编译端代码评审”
|
||||
这类证据时,该项记为“未实测”(untested_items),而不是“通过”。
|
||||
若某维的关键证据全部是未实测,该维记为 None。
|
||||
- 阈值:DEFAULT_THRESHOLDS 全部标注 source="test_default_unapproved"
|
||||
(测试默认值、未拍板)。生产阈值须用户拍板后,由调用方显式
|
||||
传入 thresholds(含 source="user_approved")才能替换默认值。
|
||||
本模块绝不把测试初值写成生产值。
|
||||
- 与已定语义的关系(本模块不改它们):
|
||||
#238 evaluate_feedback 的 auto_apply 保持 False;
|
||||
#247 纠正为 pass 本身不解封;
|
||||
#244 CONDITIONAL 先映射为 fail 再记(本模块只接受已映射的结论,
|
||||
若调用方传入 verdict="conditional",记为未实测/无效证据,不算通过)。
|
||||
|
||||
7 维定义(不要擅自增维、减维、改名换义):
|
||||
1. delivery_quality 交付质量
|
||||
2. gate_audit 门禁与审计有效性
|
||||
3. arch_contract 架构与契约一致性
|
||||
4. test_coverage 测试与验证覆盖
|
||||
5. security_ledger 安全与账本可信
|
||||
6. cost_quota 成本与配额可观测
|
||||
7. ops_scale 生产运维与规模
|
||||
|
||||
10 维到 7 维的合并映射(写进文档,也写在这里):
|
||||
- 流程完整性并入门禁与审计有效性(G0-G8 最终靠门禁和重审证明);
|
||||
- 自动化程度并入门禁与审计有效性 + 生产运维与规模
|
||||
(自动化的证据就是自动循环是否真实跑通、真实 App 是否安装);
|
||||
- 文档与技能并入架构与契约一致性
|
||||
(技能文本、附录溯源、移植规则最终都是契约一致性的一部分)。
|
||||
|
||||
输入格式:build_scorecard(evidence, thresholds=None)
|
||||
evidence: dict,键为上面 7 个维 id,值为该维的证据 dict。
|
||||
每个维的证据字段见各 _score_* 函数的 docstring。
|
||||
调用方必须带来源、时间、检查名;本模块不自动从测试文件、
|
||||
摘要文档、PR 描述里拼凑分数。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
PASS_SCORE = 9.0
|
||||
|
||||
DIMENSIONS = [
|
||||
("delivery_quality", "交付质量"),
|
||||
("gate_audit", "门禁与审计有效性"),
|
||||
("arch_contract", "架构与契约一致性"),
|
||||
("test_coverage", "测试与验证覆盖"),
|
||||
("security_ledger", "安全与账本可信"),
|
||||
("cost_quota", "成本与配额可观测"),
|
||||
("ops_scale", "生产运维与规模"),
|
||||
]
|
||||
|
||||
DIMENSION_IDS = [d[0] for d in DIMENSIONS]
|
||||
|
||||
# 阈值表:每一项都带 source,默认值一律是测试初值、未拍板。
|
||||
# “9 分线”是达到 9 分的下限,“10 分线”是达到 10 分的下限,
|
||||
# “floor”是低于该值后分数进入 0-5 区间的参考点(不是达标线)。
|
||||
# 这些数值本身不是生产标准;生产阈值须用户拍板后显式传入。
|
||||
DEFAULT_THRESHOLDS = {
|
||||
"defect_rate_nine": {"value": 0.05, "source": "test_default_unapproved"},
|
||||
"defect_rate_ten": {"value": 0.02, "source": "test_default_unapproved"},
|
||||
"defect_rate_floor": {"value": 0.20, "source": "test_default_unapproved"},
|
||||
"first_pass_nine": {"value": 0.90, "source": "test_default_unapproved"},
|
||||
"first_pass_ten": {"value": 0.95, "source": "test_default_unapproved"},
|
||||
"first_pass_floor": {"value": 0.50, "source": "test_default_unapproved"},
|
||||
"rework_rate_nine": {"value": 0.10, "source": "test_default_unapproved"},
|
||||
"rework_rate_ten": {"value": 0.05, "source": "test_default_unapproved"},
|
||||
"rework_rate_floor": {"value": 0.50, "source": "test_default_unapproved"},
|
||||
"overturn_rate_floor": {"value": 0.30, "source": "test_default_unapproved"},
|
||||
"coverage_nine": {"value": 0.85, "source": "test_default_unapproved"},
|
||||
"coverage_ten": {"value": 0.90, "source": "test_default_unapproved"},
|
||||
"coverage_floor": {"value": 0.75, "source": "test_default_unapproved"},
|
||||
"sampling_normal": {"value": 0.10, "source": "test_default_unapproved"},
|
||||
"sampling_critical": {"value": 0.50, "source": "test_default_unapproved"},
|
||||
"parallel_scale_target": {"value": 40, "source": "test_default_unapproved"},
|
||||
}
|
||||
|
||||
UNTESTED_KINDS = {"fixture", "viewport_substitute", "code_review_only",
|
||||
"fixture_simulated", "dry_run", "estimated"}
|
||||
|
||||
|
||||
def _thresholds_source(thresholds):
|
||||
"""阈值来源:只有调用方显式传入 user_approved 才算已拍板。"""
|
||||
if not thresholds:
|
||||
return "test_default_unapproved"
|
||||
sources = set()
|
||||
for entry in thresholds.values():
|
||||
if isinstance(entry, dict):
|
||||
sources.add(entry.get("source", "test_default_unapproved"))
|
||||
else:
|
||||
sources.add("test_default_unapproved")
|
||||
if sources == {"user_approved"}:
|
||||
return "user_approved"
|
||||
return "test_default_unapproved"
|
||||
|
||||
|
||||
def _t(thresholds, key):
|
||||
if thresholds and key in thresholds:
|
||||
entry = thresholds[key]
|
||||
if isinstance(entry, dict):
|
||||
return entry["value"]
|
||||
return entry
|
||||
return DEFAULT_THRESHOLDS[key]["value"]
|
||||
|
||||
|
||||
def _round1(value):
|
||||
return round(float(value) + 1e-9, 1)
|
||||
|
||||
|
||||
def _score_higher(value, nine, ten, floor):
|
||||
"""越高越好:floor->5 分,nine->9 分,ten->10 分,分段线性,低于 floor 按比例。"""
|
||||
if value is None:
|
||||
return None
|
||||
v = float(value)
|
||||
if v >= ten:
|
||||
return 10.0
|
||||
if v >= nine:
|
||||
return 9.0 + (v - nine) / (ten - nine) if ten > nine else 10.0
|
||||
if v >= floor:
|
||||
return 5.0 + 4.0 * (v - floor) / (nine - floor) if nine > floor else 5.0
|
||||
if floor > 0:
|
||||
return max(0.0, 5.0 * v / floor)
|
||||
return 0.0
|
||||
|
||||
|
||||
def _score_lower(value, ten, nine, floor):
|
||||
"""越低越好:ten->10 分,nine->9 分,floor->5 分,低于 ten 满分,高于 floor 递减。"""
|
||||
if value is None:
|
||||
return None
|
||||
v = float(value)
|
||||
if v <= ten:
|
||||
return 10.0
|
||||
if v <= nine:
|
||||
return 9.0 + (nine - v) / (nine - ten) if nine > ten else 10.0
|
||||
if v <= floor:
|
||||
return 5.0 + 4.0 * (floor - v) / (floor - nine) if floor > nine else 5.0
|
||||
return max(0.0, 5.0 * floor / v) if v > 0 else 0.0
|
||||
|
||||
|
||||
def _result(dim_id, score, basis, missing, untested):
|
||||
if score is None:
|
||||
status = "unknown"
|
||||
elif score >= PASS_SCORE:
|
||||
status = "pass"
|
||||
else:
|
||||
status = "fail"
|
||||
return {
|
||||
"id": dim_id,
|
||||
"score": _round1(score) if score is not None else None,
|
||||
"status": status,
|
||||
"basis": basis,
|
||||
"missing_items": missing,
|
||||
"untested_items": untested,
|
||||
}
|
||||
|
||||
|
||||
def _score_delivery(ev, th):
|
||||
"""证据:first_pass_rate, defect_rate, rework_rate,
|
||||
first_pass_known, defect_known, rework_known, attempts_present,
|
||||
evidence_kind("real" 以外若为 estimated/fixture 则记未实测)。"""
|
||||
missing, untested, basis = [], [], []
|
||||
for key in ("first_pass_rate", "defect_rate", "rework_rate"):
|
||||
if ev.get(key) is None:
|
||||
missing.append(key)
|
||||
for key in ("first_pass_known", "defect_known", "rework_known"):
|
||||
if not ev.get(key):
|
||||
missing.append(key + "_not_known")
|
||||
if not ev.get("attempts_present"):
|
||||
missing.append("attempts_present_false_or_missing")
|
||||
kind = ev.get("evidence_kind")
|
||||
if kind in UNTESTED_KINDS:
|
||||
untested.append("metrics_evidence_kind=%s" % kind)
|
||||
if missing or untested:
|
||||
return _result("delivery_quality", None, basis, missing, untested)
|
||||
subs = [
|
||||
("first_pass_rate", _score_higher(ev["first_pass_rate"],
|
||||
_t(th, "first_pass_nine"),
|
||||
_t(th, "first_pass_ten"),
|
||||
_t(th, "first_pass_floor"))),
|
||||
("defect_rate", _score_lower(ev["defect_rate"],
|
||||
_t(th, "defect_rate_ten"),
|
||||
_t(th, "defect_rate_nine"),
|
||||
_t(th, "defect_rate_floor"))),
|
||||
("rework_rate", _score_lower(ev["rework_rate"],
|
||||
_t(th, "rework_rate_ten"),
|
||||
_t(th, "rework_rate_nine"),
|
||||
_t(th, "rework_rate_floor"))),
|
||||
]
|
||||
for name, sub in subs:
|
||||
basis.append("%s=%s -> %.1f" % (name, ev[name], sub))
|
||||
return _result("delivery_quality", min(s for _, s in subs), basis, missing, untested)
|
||||
|
||||
|
||||
def _score_gate(ev, th):
|
||||
"""证据:hard_gate_enforced(bool/None)、gate_items_passed、
|
||||
gate_items_total、merge_blocked_verified、loop_end_to_end_verified、
|
||||
sampling_misjudgment_rate(None 表示未抽查/未知)、
|
||||
audit_verdict("conditional" 记无效证据,不算通过)。"""
|
||||
missing, untested, basis = [], [], []
|
||||
if ev.get("hard_gate_enforced") is None:
|
||||
missing.append("hard_gate_enforced_unknown")
|
||||
if ev.get("gate_items_passed") is None or ev.get("gate_items_total") is None:
|
||||
missing.append("gate_items_passed_or_total_missing")
|
||||
if ev.get("audit_verdict") == "conditional":
|
||||
untested.append("audit_verdict=conditional_not_a_pass")
|
||||
if ev.get("evidence_kind") in UNTESTED_KINDS:
|
||||
untested.append("gate_evidence_kind=%s" % ev["evidence_kind"])
|
||||
if missing or untested:
|
||||
return _result("gate_audit", None, basis, missing, untested)
|
||||
if not ev.get("hard_gate_enforced"):
|
||||
basis.append("hard_gate_enforced=False(record 模式):已知未强制,分数封顶 4.0")
|
||||
return _result("gate_audit", 4.0, basis, missing, untested)
|
||||
total = ev["gate_items_total"]
|
||||
passed = ev["gate_items_passed"]
|
||||
rate = passed / total if total else 0.0
|
||||
sub_gate = _score_higher(rate, 0.99, 1.0, 0.80)
|
||||
basis.append("gate_items %s/%s=%.3f -> %.1f" % (passed, total, rate, sub_gate))
|
||||
subs = [sub_gate]
|
||||
if not ev.get("merge_blocked_verified"):
|
||||
basis.append("merge_blocked_verified=False:合并阻塞未真实验证,分数封顶 4.0")
|
||||
return _result("gate_audit", 4.0, basis, missing, untested)
|
||||
if not ev.get("loop_end_to_end_verified"):
|
||||
basis.append("loop_end_to_end_verified=False:自动循环端到端未真实验证,分数封顶 4.0")
|
||||
return _result("gate_audit", 4.0, basis, missing, untested)
|
||||
mis = ev.get("sampling_misjudgment_rate")
|
||||
if mis is None:
|
||||
missing.append("sampling_misjudgment_rate_unknown")
|
||||
return _result("gate_audit", None, basis, missing, untested)
|
||||
sub_mis = _score_lower(mis, 0.0, 0.02, _t(th, "overturn_rate_floor"))
|
||||
basis.append("sampling_misjudgment_rate=%s -> %.1f" % (mis, sub_mis))
|
||||
subs.append(sub_mis)
|
||||
return _result("gate_audit", min(subs), basis, missing, untested)
|
||||
|
||||
|
||||
def _score_arch(ev, th):
|
||||
"""证据:contract_frozen、arch_violations、directory_violations、
|
||||
import_violations、stack_changed_without_approval(bool/None)。"""
|
||||
missing, untested, basis = [], [], []
|
||||
for key in ("arch_violations", "directory_violations", "import_violations"):
|
||||
if ev.get(key) is None:
|
||||
missing.append(key + "_unknown")
|
||||
if ev.get("contract_frozen") is None:
|
||||
missing.append("contract_frozen_unknown")
|
||||
if missing:
|
||||
return _result("arch_contract", None, basis, missing, untested)
|
||||
total_violations = (ev["arch_violations"] + ev["directory_violations"]
|
||||
+ ev["import_violations"])
|
||||
basis.append("violations total=%s (arch=%s dir=%s import=%s)"
|
||||
% (total_violations, ev["arch_violations"],
|
||||
ev["directory_violations"], ev["import_violations"]))
|
||||
if not ev.get("contract_frozen"):
|
||||
basis.append("contract_frozen=False:已知未冻结,分数封顶 4.0")
|
||||
return _result("arch_contract", 4.0, basis, missing, untested)
|
||||
if ev.get("stack_changed_without_approval"):
|
||||
basis.append("stack_changed_without_approval=True:分数封顶 4.0")
|
||||
return _result("arch_contract", 4.0, basis, missing, untested)
|
||||
if total_violations > 0:
|
||||
basis.append("violations>0:分数封顶 5.0")
|
||||
return _result("arch_contract", 5.0, basis, missing, untested)
|
||||
return _result("arch_contract", 10.0, basis, missing, untested)
|
||||
|
||||
|
||||
def _score_test(ev, th):
|
||||
"""证据:prd_coverage_rate、agent_six_complete、cross_matrix_coverage_rate、
|
||||
coverage_pct、lint_items、mypy_items、bandit_high、
|
||||
cross_matrix_kind("real" 以外记未实测)、real_pg(bool)、
|
||||
real_service(bool)。"""
|
||||
missing, untested, basis = [], [], []
|
||||
for key in ("prd_coverage_rate", "cross_matrix_coverage_rate", "coverage_pct"):
|
||||
if ev.get(key) is None:
|
||||
missing.append(key + "_missing")
|
||||
kind = ev.get("cross_matrix_kind")
|
||||
if kind in UNTESTED_KINDS or kind in ("viewport_substitute", "code_review_only"):
|
||||
untested.append("cross_matrix_kind=%s" % kind)
|
||||
if kind is not None and kind != "real":
|
||||
untested.append("cross_matrix_not_real")
|
||||
if not ev.get("real_pg"):
|
||||
untested.append("real_pg_not_verified")
|
||||
if not ev.get("real_service"):
|
||||
untested.append("real_service_not_verified")
|
||||
if missing or untested:
|
||||
return _result("test_coverage", None, basis, missing, untested)
|
||||
if not ev.get("agent_six_complete"):
|
||||
basis.append("agent_six_complete=False:agent 专属六项缺项,分数封顶 4.0")
|
||||
return _result("test_coverage", 4.0, basis, missing, untested)
|
||||
subs = [
|
||||
("prd_coverage_rate", _score_higher(ev["prd_coverage_rate"], 0.99, 1.0, 0.80)),
|
||||
("cross_matrix_coverage_rate", _score_higher(ev["cross_matrix_coverage_rate"], 0.99, 1.0, 0.80)),
|
||||
("coverage_pct", _score_higher(ev["coverage_pct"],
|
||||
_t(th, "coverage_nine"),
|
||||
_t(th, "coverage_ten"),
|
||||
_t(th, "coverage_floor"))),
|
||||
]
|
||||
for name, sub in subs:
|
||||
basis.append("%s=%s -> %.1f" % (name, ev[name], sub))
|
||||
# 静态检查:数值缺失记缺失,不估算;有值时,>0 即封顶(9 分线要求接近清零)。
|
||||
for key in ("lint_items", "mypy_items", "bandit_high"):
|
||||
if ev.get(key) is None:
|
||||
missing.append(key + "_missing")
|
||||
if missing:
|
||||
return _result("test_coverage", None, basis, missing, untested)
|
||||
static_total = ev["lint_items"] + ev["mypy_items"] + ev["bandit_high"]
|
||||
basis.append("static total lint=%s mypy=%s bandit_high=%s"
|
||||
% (ev["lint_items"], ev["mypy_items"], ev["bandit_high"]))
|
||||
if static_total > 0:
|
||||
basis.append("static_total>0:分数封顶 8.0(静态检查未接近清零)")
|
||||
return _result("test_coverage", min(8.0, min(s for _, s in subs)), basis, missing, untested)
|
||||
return _result("test_coverage", min(s for _, s in subs), basis, missing, untested)
|
||||
|
||||
|
||||
def _score_security(ev, th):
|
||||
"""证据:ledger_verify_ok(bool/None)、ledger_coverage_rate、
|
||||
shell_whitelist_pass_rate、branch_protection
|
||||
("protected"/"unprotected"/"unknown"/None)、anchor_check_recent。"""
|
||||
missing, untested, basis = [], [], []
|
||||
if ev.get("ledger_verify_ok") is None:
|
||||
missing.append("ledger_verify_ok_unknown")
|
||||
if ev.get("ledger_coverage_rate") is None:
|
||||
missing.append("ledger_coverage_rate_missing")
|
||||
bp = ev.get("branch_protection")
|
||||
if bp is None or bp == "unknown":
|
||||
missing.append("branch_protection_unknown")
|
||||
if missing:
|
||||
return _result("security_ledger", None, basis, missing, untested)
|
||||
if bp == "unprotected":
|
||||
basis.append("branch_protection=unprotected:已知未保护,分数封顶 4.0")
|
||||
return _result("security_ledger", 4.0, basis, missing, untested)
|
||||
if not ev.get("ledger_verify_ok"):
|
||||
basis.append("ledger_verify_ok=False:账本校验未通过,分数封顶 4.0")
|
||||
return _result("security_ledger", 4.0, basis, missing, untested)
|
||||
subs = [_score_higher(ev["ledger_coverage_rate"], 0.99, 1.0, 0.80)]
|
||||
basis.append("ledger_coverage_rate=%s -> %.1f" % (ev["ledger_coverage_rate"], subs[0]))
|
||||
rate = ev.get("shell_whitelist_pass_rate")
|
||||
if rate is not None:
|
||||
sub = _score_higher(rate, 0.99, 1.0, 0.80)
|
||||
subs.append(sub)
|
||||
basis.append("shell_whitelist_pass_rate=%s -> %.1f" % (rate, sub))
|
||||
if not ev.get("anchor_check_recent"):
|
||||
basis.append("anchor_check_recent=False:锚点定期校验未落地,分数封顶 8.0")
|
||||
return _result("security_ledger", min(8.0, min(subs)), basis, missing, untested)
|
||||
return _result("security_ledger", min(subs), basis, missing, untested)
|
||||
|
||||
|
||||
def _score_cost(ev, th):
|
||||
"""证据:cost_records_complete(bool)、unpriced_as_zero(bool,
|
||||
True 即违规)、per_creator_present、quota_cutoff_verified、
|
||||
thresholds_approved(bool,仅影响阈值来源标注,不封顶分数)。"""
|
||||
missing, untested, basis = [], [], []
|
||||
if not ev.get("cost_records_complete"):
|
||||
missing.append("cost_records_complete_false_or_missing"
|
||||
"(turns/llm_requests/usd_micros 缺失)")
|
||||
if ev.get("unpriced_as_zero") is None:
|
||||
missing.append("unpriced_as_zero_unknown")
|
||||
if missing:
|
||||
return _result("cost_quota", None, basis, missing, untested)
|
||||
if ev.get("unpriced_as_zero"):
|
||||
basis.append("unpriced_as_zero=True:无价记成 0/免费,分数封顶 0.0")
|
||||
return _result("cost_quota", 0.0, basis, missing, untested)
|
||||
if not ev.get("quota_cutoff_verified"):
|
||||
basis.append("quota_cutoff_verified=False:配额硬切断未真实验证,分数封顶 4.0")
|
||||
return _result("cost_quota", 4.0, basis, missing, untested)
|
||||
if not ev.get("per_creator_present"):
|
||||
basis.append("per_creator_present=False:按创建者成本未落地,分数封顶 8.0")
|
||||
return _result("cost_quota", 8.0, basis, missing, untested)
|
||||
basis.append("cost_records_complete=True, quota_cutoff_verified=True, "
|
||||
"per_creator_present=True, unpriced_as_zero=False")
|
||||
return _result("cost_quota", 10.0, basis, missing, untested)
|
||||
|
||||
|
||||
def _score_ops(ev, th):
|
||||
"""证据:ci_green_rate(None 表示 CI 未绿/未知)、deploy_success_rate、
|
||||
rollback_drill_done、ha_status
|
||||
("second_host"/"single_host_accepted"/"single_host_unaccepted"/None)、
|
||||
parallel_scale_proven(int/None)、parallel_scale_accepted_alternative(bool)。"""
|
||||
missing, untested, basis = [], [], []
|
||||
if ev.get("ci_green_rate") is None:
|
||||
missing.append("ci_green_rate_unknown(CI 未绿或未统计)")
|
||||
if ev.get("rollback_drill_done") is None:
|
||||
missing.append("rollback_drill_done_unknown")
|
||||
ha = ev.get("ha_status")
|
||||
if ha is None:
|
||||
missing.append("ha_status_unknown")
|
||||
if missing:
|
||||
return _result("ops_scale", None, basis, missing, untested)
|
||||
if not ev.get("rollback_drill_done"):
|
||||
basis.append("rollback_drill_done=False:回滚演练未真实跑过,分数封顶 4.0")
|
||||
return _result("ops_scale", 4.0, basis, missing, untested)
|
||||
subs = [_score_higher(ev["ci_green_rate"], 0.99, 1.0, 0.80)]
|
||||
basis.append("ci_green_rate=%s -> %.1f" % (ev["ci_green_rate"], subs[0]))
|
||||
if ev.get("deploy_success_rate") is not None:
|
||||
sub = _score_higher(ev["deploy_success_rate"], 0.99, 1.0, 0.80)
|
||||
subs.append(sub)
|
||||
basis.append("deploy_success_rate=%s -> %.1f" % (ev["deploy_success_rate"], sub))
|
||||
if ha == "single_host_unaccepted":
|
||||
basis.append("ha_status=single_host_unaccepted:分数封顶 5.0")
|
||||
return _result("ops_scale", 5.0, basis, missing, untested)
|
||||
proven = ev.get("parallel_scale_proven")
|
||||
target = _t(th, "parallel_scale_target")
|
||||
if proven is None and not ev.get("parallel_scale_accepted_alternative"):
|
||||
missing.append("parallel_scale_proven_unknown")
|
||||
return _result("ops_scale", None, basis, missing, untested)
|
||||
if proven is not None:
|
||||
sub = _score_higher(min(proven, target) / target if target else 0,
|
||||
0.99, 1.0, 0.5)
|
||||
# 规模是离散证据:40 人满分,18 人约 4.5 分,上面线性公式在
|
||||
# 0.5-1.0 区间给出 5-10 分,与“18 人约 4.5”不一致,故直接用比例*10。
|
||||
sub = _round1(10.0 * min(proven, target) / target) if target else 0.0
|
||||
subs.append(sub)
|
||||
basis.append("parallel_scale_proven=%s target=%s -> %.1f"
|
||||
% (proven, target, sub))
|
||||
return _result("ops_scale", min(subs), basis, missing, untested)
|
||||
|
||||
|
||||
_SCORERS = {
|
||||
"delivery_quality": _score_delivery,
|
||||
"gate_audit": _score_gate,
|
||||
"arch_contract": _score_arch,
|
||||
"test_coverage": _score_test,
|
||||
"security_ledger": _score_security,
|
||||
"cost_quota": _score_cost,
|
||||
"ops_scale": _score_ops,
|
||||
}
|
||||
|
||||
|
||||
def build_scorecard(evidence, thresholds=None):
|
||||
"""算出 7 维评分卡。evidence 缺键按空证据处理(该维记 None)。"""
|
||||
evidence = evidence or {}
|
||||
th = thresholds or {}
|
||||
dimensions = []
|
||||
missing_all, untested_all = [], []
|
||||
for dim_id, _name in DIMENSIONS:
|
||||
result = _SCORERS[dim_id](evidence.get(dim_id) or {}, th)
|
||||
dimensions.append(result)
|
||||
missing_all.extend("%s: %s" % (dim_id, m) for m in result["missing_items"])
|
||||
untested_all.extend("%s: %s" % (dim_id, u) for u in result["untested_items"])
|
||||
scores = [d["score"] for d in dimensions if d["score"] is not None]
|
||||
average = _round1(sum(scores) / len(scores)) if scores else None
|
||||
# 总体规则钉死:平均分绝不能作为达标依据。
|
||||
overall_pass = all(d["score"] is not None and d["score"] >= PASS_SCORE
|
||||
for d in dimensions)
|
||||
return {
|
||||
"dimensions": dimensions,
|
||||
"average_score": average,
|
||||
"average_is_not_pass_criterion": True,
|
||||
"overall": "pass" if overall_pass else "not_pass",
|
||||
"overall_pass": overall_pass,
|
||||
"pass_score": PASS_SCORE,
|
||||
"missing_items": missing_all,
|
||||
"untested_items": untested_all,
|
||||
"thresholds_source": _thresholds_source(th),
|
||||
"thresholds": {k: v for k, v in DEFAULT_THRESHOLDS.items()},
|
||||
}
|
||||
|
||||
|
||||
def example_scorecard_current_state():
|
||||
"""用当前真实状态证据跑出来的系统评分卡(不是上一轮判断分)。
|
||||
|
||||
证据口径:生产仍跑 v0.2、main 仍 aebc623、CI 账单模式、
|
||||
#238/#240/#244/#241/#247/#249/#253/#255/#256/#258/#259/#260/#261
|
||||
全部只到 PR OPEN。凡是只到 PR、未合并部署、未真实验证的,
|
||||
在这里都记为未知或已知未强制,不估算、不写成判断分。
|
||||
总体结论必然是不达标。
|
||||
"""
|
||||
evidence = {
|
||||
"delivery_quality": {
|
||||
"first_pass_rate": None, "defect_rate": None, "rework_rate": None,
|
||||
"first_pass_known": False, "defect_known": False,
|
||||
"rework_known": False, "attempts_present": False,
|
||||
},
|
||||
"gate_audit": {
|
||||
"hard_gate_enforced": False, "gate_items_passed": 0,
|
||||
"gate_items_total": 5, "merge_blocked_verified": False,
|
||||
"loop_end_to_end_verified": False,
|
||||
"sampling_misjudgment_rate": None,
|
||||
},
|
||||
"arch_contract": {
|
||||
"contract_frozen": None, "arch_violations": None,
|
||||
"directory_violations": None, "import_violations": None,
|
||||
},
|
||||
"test_coverage": {
|
||||
"prd_coverage_rate": None, "cross_matrix_coverage_rate": None,
|
||||
"coverage_pct": 0.7787, "cross_matrix_kind": "viewport_substitute",
|
||||
"real_pg": False, "real_service": False,
|
||||
"agent_six_complete": True,
|
||||
"lint_items": 109, "mypy_items": 0, "bandit_high": 4,
|
||||
},
|
||||
"security_ledger": {
|
||||
"ledger_verify_ok": None, "ledger_coverage_rate": None,
|
||||
"branch_protection": "unknown",
|
||||
},
|
||||
"cost_quota": {
|
||||
"cost_records_complete": False, "unpriced_as_zero": False,
|
||||
"per_creator_present": False, "quota_cutoff_verified": False,
|
||||
},
|
||||
"ops_scale": {
|
||||
"ci_green_rate": None, "rollback_drill_done": False,
|
||||
"ha_status": "single_host_unaccepted",
|
||||
"parallel_scale_proven": 18,
|
||||
},
|
||||
}
|
||||
return build_scorecard(evidence)
|
||||
@@ -3133,6 +3133,8 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
|
||||
self._metrics(user_id)
|
||||
elif path == "/metrics-report" and method == "GET":
|
||||
self._metrics_report(user_id)
|
||||
elif path == "/audit-scorecard" and method == "GET":
|
||||
self._audit_scorecard(user_id)
|
||||
elif path == "/models" and method == "GET":
|
||||
self._models_overview(user_id)
|
||||
elif path == "/relay/models" and method == "GET":
|
||||
@@ -10713,6 +10715,32 @@ pid 须为整数、address 须为字符串,否则 400(注册表里的
|
||||
conn.close()
|
||||
metrics = _compute_metrics(records, costs)
|
||||
self._send_json(200, _build_metrics_report(metrics))
|
||||
def _audit_scorecard(self, user_id):
|
||||
"""只读 7 维审计评分卡报告端点。
|
||||
|
||||
权限:与 /metrics 同款,任意项目成员即可读(不要求 owner),
|
||||
但本端点不统计任何项目数据、不写库、不写账本、不改链头、
|
||||
不触发推回或重审——它只是把评分卡模块的当前状态报出来。
|
||||
主基底(main)上没有存放结构化评分证据的表(metrics-report、
|
||||
台账、账本、门禁模块都在未合并的 PR 分支上),所以本端点
|
||||
在主基底上返回的只能是“全部未知/已知未强制”的诚实状态,
|
||||
总体必然是不达标;调用方要拿到真实分数,必须显式把结构化
|
||||
证据传给 cloud/audit_scorecard.py 的 build_scorecard(),
|
||||
端点本身绝不从测试文件、摘要文档、PR 描述里拼凑分数。
|
||||
"""
|
||||
try:
|
||||
from cloud.audit_scorecard import example_scorecard_current_state
|
||||
report = example_scorecard_current_state()
|
||||
except ImportError: # pragma: no cover - 导入失败时仍要诚实返回
|
||||
report = {
|
||||
"dimensions": [], "average_score": None,
|
||||
"average_is_not_pass_criterion": True,
|
||||
"overall": "not_pass", "overall_pass": False,
|
||||
"missing_items": ["audit_scorecard_module_unavailable"],
|
||||
"untested_items": [],
|
||||
"thresholds_source": "test_default_unapproved",
|
||||
}
|
||||
self._send_json(200, report)
|
||||
|
||||
# -- OAuth(pre-auth,靠 state + secret,不走 X-API-Key)-----------
|
||||
def _oauth_provider(self, path):
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
"""7 维审计评分卡定向测试。
|
||||
|
||||
钉住的规则(任何一处改动都必须先改文档再改测试):
|
||||
- 7 维全部 >=9 才达标;任一维 <9 总体不达标。
|
||||
- 平均分绝不能作为达标依据:6 维 10 分、1 维 4 分,平均 9.1 分,
|
||||
总体仍是不达标。
|
||||
- 未知维记 None 且总体不达标;未知不记 0、不记 10、不估算。
|
||||
- 夹具/视口替代/代码评审证据记未实测,不算通过。
|
||||
- 阈值默认值标注 test_default_unapproved;只有调用方显式传入
|
||||
user_approved 才算已拍板。
|
||||
- 端点只读:GET /audit-scorecard 不写库、不写账本、不改链头。
|
||||
"""
|
||||
|
||||
import http.client
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from cloud.audit_scorecard import (
|
||||
DEFAULT_THRESHOLDS, build_scorecard, example_scorecard_current_state)
|
||||
from cloud.auth import AuthStore
|
||||
from cloud.server import create_server
|
||||
except ImportError:
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
from cloud.audit_scorecard import (
|
||||
DEFAULT_THRESHOLDS, build_scorecard, example_scorecard_current_state)
|
||||
from cloud.auth import AuthStore
|
||||
from cloud.server import create_server
|
||||
|
||||
|
||||
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,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
class ScorecardTestCase(unittest.TestCase):
|
||||
def test_all_nine_passes(self):
|
||||
report = build_scorecard(_full_evidence())
|
||||
self.assertEqual(report["overall"], "pass")
|
||||
self.assertTrue(report["overall_pass"])
|
||||
self.assertEqual(len(report["dimensions"]), 7)
|
||||
for dim in report["dimensions"]:
|
||||
self.assertIsNotNone(dim["score"])
|
||||
self.assertGreaterEqual(dim["score"], 9.0)
|
||||
self.assertEqual(dim["status"], "pass")
|
||||
|
||||
def test_one_dimension_below_nine_fails_overall(self):
|
||||
ev = _full_evidence()
|
||||
ev["gate_audit"]["hard_gate_enforced"] = False
|
||||
report = build_scorecard(ev)
|
||||
self.assertEqual(report["overall"], "not_pass")
|
||||
self.assertFalse(report["overall_pass"])
|
||||
gate = [d for d in report["dimensions"] if d["id"] == "gate_audit"][0]
|
||||
self.assertEqual(gate["score"], 4.0)
|
||||
|
||||
def test_average_91_with_one_four_still_fails(self):
|
||||
# 6 维 10 分、1 维 4 分:平均 9.1 分,总体仍是不达标。
|
||||
# 构造方式:只让门禁维已知未强制(4.0),其余维满分证据。
|
||||
ev = _full_evidence()
|
||||
ev["gate_audit"]["hard_gate_enforced"] = False
|
||||
report = build_scorecard(ev)
|
||||
self.assertEqual(report["average_score"], 9.1)
|
||||
self.assertTrue(report["average_is_not_pass_criterion"])
|
||||
self.assertEqual(report["overall"], "not_pass")
|
||||
self.assertFalse(report["overall_pass"])
|
||||
|
||||
def test_unknown_dimension_is_none_and_fails_overall(self):
|
||||
report = build_scorecard({})
|
||||
self.assertEqual(report["overall"], "not_pass")
|
||||
for dim in report["dimensions"]:
|
||||
self.assertIsNone(dim["score"])
|
||||
self.assertEqual(dim["status"], "unknown")
|
||||
self.assertTrue(report["missing_items"])
|
||||
self.assertIsNone(report["average_score"])
|
||||
|
||||
def test_unknown_is_not_zero_and_not_ten(self):
|
||||
ev = _full_evidence()
|
||||
ev["delivery_quality"] = {}
|
||||
report = build_scorecard(ev)
|
||||
dim = [d for d in report["dimensions"]
|
||||
if d["id"] == "delivery_quality"][0]
|
||||
self.assertIsNone(dim["score"])
|
||||
self.assertNotEqual(dim["score"], 0)
|
||||
self.assertNotEqual(dim["score"], 10)
|
||||
|
||||
def test_fixture_evidence_is_untested_not_pass(self):
|
||||
ev = _full_evidence()
|
||||
ev["test_coverage"]["cross_matrix_kind"] = "viewport_substitute"
|
||||
report = build_scorecard(ev)
|
||||
dim = [d for d in report["dimensions"]
|
||||
if d["id"] == "test_coverage"][0]
|
||||
self.assertIsNone(dim["score"])
|
||||
self.assertEqual(dim["status"], "unknown")
|
||||
self.assertTrue(dim["untested_items"])
|
||||
self.assertEqual(report["overall"], "not_pass")
|
||||
|
||||
def test_conditional_verdict_is_not_pass(self):
|
||||
ev = _full_evidence()
|
||||
ev["gate_audit"]["audit_verdict"] = "conditional"
|
||||
report = build_scorecard(ev)
|
||||
dim = [d for d in report["dimensions"] if d["id"] == "gate_audit"][0]
|
||||
self.assertIsNone(dim["score"])
|
||||
self.assertTrue(dim["untested_items"])
|
||||
|
||||
def test_threshold_defaults_are_unapproved(self):
|
||||
for entry in DEFAULT_THRESHOLDS.values():
|
||||
self.assertEqual(entry["source"], "test_default_unapproved")
|
||||
report = build_scorecard(_full_evidence())
|
||||
self.assertEqual(report["thresholds_source"],
|
||||
"test_default_unapproved")
|
||||
|
||||
def test_user_approved_thresholds_change_source_only(self):
|
||||
th = {k: {"value": v["value"], "source": "user_approved"}
|
||||
for k, v in DEFAULT_THRESHOLDS.items()}
|
||||
report = build_scorecard(_full_evidence(), thresholds=th)
|
||||
self.assertEqual(report["thresholds_source"], "user_approved")
|
||||
self.assertEqual(report["overall"], "pass")
|
||||
|
||||
def test_example_current_state_is_not_pass_and_not_estimated(self):
|
||||
# 示例是系统用当前真实状态证据算出来的,不是上一轮判断分。
|
||||
report = example_scorecard_current_state()
|
||||
self.assertEqual(report["overall"], "not_pass")
|
||||
self.assertFalse(report["overall_pass"])
|
||||
# 至少有一维是未知(None),且总体不达标不靠平均分。
|
||||
self.assertTrue(any(d["score"] is None
|
||||
for d in report["dimensions"]))
|
||||
|
||||
|
||||
class ScorecardEndpointTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = tempfile.TemporaryDirectory()
|
||||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||||
self.auth = AuthStore(self.db_path)
|
||||
self.owner_id, self.owner_key = self.auth.create_user("owner")
|
||||
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
||||
self.port = self.server.server_address[1]
|
||||
self.thread = threading.Thread(
|
||||
target=self.server.serve_forever, daemon=True)
|
||||
self.thread.start()
|
||||
|
||||
def tearDown(self):
|
||||
self.server.shutdown()
|
||||
self.server.server_close()
|
||||
self.thread.join(timeout=5)
|
||||
try:
|
||||
self.auth.conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.tmp.cleanup()
|
||||
|
||||
def _req(self, method, path, key=None):
|
||||
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
|
||||
hdrs = {}
|
||||
if key is not None:
|
||||
hdrs["X-API-Key"] = key
|
||||
try:
|
||||
conn.request(method, path, headers=hdrs)
|
||||
resp = conn.getresponse()
|
||||
data = resp.read()
|
||||
finally:
|
||||
conn.close()
|
||||
try:
|
||||
parsed = json.loads(data.decode("utf-8")) if data else None
|
||||
except ValueError:
|
||||
parsed = None
|
||||
return resp.status, parsed
|
||||
|
||||
def test_endpoint_requires_auth(self):
|
||||
status, _ = self._req("GET", "/audit-scorecard")
|
||||
self.assertEqual(status, 401)
|
||||
|
||||
def test_endpoint_is_readonly_and_not_pass_on_main_base(self):
|
||||
status, report = self._req("GET", "/audit-scorecard",
|
||||
key=self.owner_key)
|
||||
self.assertEqual(status, 200)
|
||||
self.assertEqual(len(report["dimensions"]), 7)
|
||||
self.assertEqual(report["overall"], "not_pass")
|
||||
self.assertTrue(report["average_is_not_pass_criterion"])
|
||||
self.assertEqual(report["thresholds_source"],
|
||||
"test_default_unapproved")
|
||||
# 只读:端点调用前后,库里不产生评分卡表(本端点不写库)。
|
||||
import sqlite3
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
try:
|
||||
tables = {r[0] for r in conn.execute(
|
||||
"SELECT name FROM sqlite_master WHERE type='table'")}
|
||||
finally:
|
||||
conn.close()
|
||||
self.assertNotIn("audit_scorecard", tables)
|
||||
Reference in new issue
Block a user