v7 mech-5: scorecard baseline_head machine field, stale check, frozen store (append-only)
(cherry picked from commit e3e1f517e916eaac597b82d04bac5307a3f8b126)
This commit is contained in:
host committed
1 parent
76f67ce310
commit
a49b9556f8
3 files changed
+531
-1
No files matched your search
@@ -0,0 +1,274 @@
|
||||
"""v7 机制第 5 项：基线漂移门禁定向测试。
|
||||
|
||||
只钉本项的三件事，不碰其它机制：
|
||||
(a) build_scorecard 输出 dict 有 baseline_head 机器字段，调用方传入，
|
||||
不传入则为 None 并在 missing_items 里注明，绝不编造；
|
||||
(b) check_baseline_stale：main 推进后旧分返回 stale、不再作 pass 依据，
|
||||
main 头一致时不 stale；stale 是附加字段，旧 overall 值保留以便追溯，
|
||||
测试断言 stale 字段而非改写旧 overall；
|
||||
(c) 冻结版存储表：只提供写入新冻结行与读取接口，不提供 UPDATE/DELETE
|
||||
方法，SQLite 触发器在表层挡住改删；真实 PG 未接通，只用 SQLite /
|
||||
内存库测试并明说。
|
||||
生产代码绝不探测版本控制状态：本文件另有源码断言钉住。
|
||||
"""
|
||||
|
||||
import ast
|
||||
import sqlite3
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from cloud.audit_scorecard import (
|
||||
build_scorecard, check_baseline_stale,
|
||||
example_scorecard_as_of_2026_10_31)
|
||||
from cloud.scorecard_baseline_store import (
|
||||
ensure_schema, freeze_scorecard, get_frozen_scorecard,
|
||||
get_latest_frozen_scorecard, list_frozen_scorecards)
|
||||
except ImportError:
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
from cloud.audit_scorecard import (
|
||||
build_scorecard, check_baseline_stale,
|
||||
example_scorecard_as_of_2026_10_31)
|
||||
from cloud.scorecard_baseline_store import (
|
||||
ensure_schema, freeze_scorecard, get_frozen_scorecard,
|
||||
get_latest_frozen_scorecard, list_frozen_scorecards)
|
||||
|
||||
OLD_HEAD = "b7ac9d7d6af842a182964b0aa0524708e7cdc923"
|
||||
NEW_HEAD = "82d0e84deadbeef00000000000000000000000000"
|
||||
|
||||
|
||||
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 BuildBaselineFieldTestCase(unittest.TestCase):
|
||||
def test_baseline_head_passed_through_as_machine_field(self):
|
||||
card = build_scorecard(_full_evidence(), baseline_head=OLD_HEAD)
|
||||
self.assertEqual(card["baseline_head"], OLD_HEAD)
|
||||
self.assertNotIn("baseline_head_missing", card["missing_items"])
|
||||
self.assertEqual(card["overall"], "pass")
|
||||
self.assertTrue(card["overall_pass"])
|
||||
|
||||
def test_baseline_head_missing_is_none_and_noted_not_fabricated(self):
|
||||
card = build_scorecard(_full_evidence())
|
||||
self.assertIsNone(card["baseline_head"])
|
||||
self.assertIn("baseline_head_missing", card["missing_items"])
|
||||
# 缺失不编造：字段本身就是 None，不是空串、不是占位头。
|
||||
self.assertNotEqual(card["baseline_head"], OLD_HEAD)
|
||||
self.assertNotEqual(card["baseline_head"], "")
|
||||
|
||||
def test_baseline_head_empty_string_normalized_to_missing(self):
|
||||
card = build_scorecard(_full_evidence(), baseline_head="")
|
||||
self.assertIsNone(card["baseline_head"])
|
||||
self.assertIn("baseline_head_missing", card["missing_items"])
|
||||
|
||||
def test_thresholds_still_test_default_unapproved(self):
|
||||
card = build_scorecard(_full_evidence(), baseline_head=OLD_HEAD)
|
||||
self.assertEqual(card["thresholds_source"], "test_default_unapproved")
|
||||
|
||||
def test_historical_example_has_no_frozen_baseline(self):
|
||||
card = example_scorecard_as_of_2026_10_31()
|
||||
self.assertIsNone(card["baseline_head"])
|
||||
self.assertIn("baseline_head_missing", card["missing_items"])
|
||||
self.assertTrue(card["not_current_system_score"])
|
||||
|
||||
|
||||
class CheckBaselineStaleTestCase(unittest.TestCase):
|
||||
def test_same_head_is_not_stale_and_usable(self):
|
||||
card = build_scorecard(_full_evidence(), baseline_head=OLD_HEAD)
|
||||
result = check_baseline_stale(card, OLD_HEAD)
|
||||
self.assertFalse(result["stale"])
|
||||
self.assertTrue(result["usable_as_pass_basis"])
|
||||
self.assertTrue(result["overall_usable_as_pass_basis"])
|
||||
self.assertFalse(result["overall_must_not_be_used_as_pass_basis"])
|
||||
self.assertEqual(result["baseline_head"], OLD_HEAD)
|
||||
self.assertEqual(result["current_main_head"], OLD_HEAD)
|
||||
|
||||
def test_advanced_head_is_stale_and_not_pass_basis(self):
|
||||
card = build_scorecard(_full_evidence(), baseline_head=OLD_HEAD)
|
||||
self.assertEqual(card["overall"], "pass")
|
||||
result = check_baseline_stale(card, NEW_HEAD)
|
||||
self.assertTrue(result["stale"])
|
||||
self.assertFalse(result["usable_as_pass_basis"])
|
||||
self.assertFalse(result["overall_usable_as_pass_basis"])
|
||||
self.assertTrue(result["overall_must_not_be_used_as_pass_basis"])
|
||||
self.assertIn(OLD_HEAD, result["reason"])
|
||||
self.assertIn(NEW_HEAD, result["reason"])
|
||||
# 关键：stale 是附加字段，旧 overall 值保留以便追溯，不被改写。
|
||||
self.assertEqual(card["overall"], "pass")
|
||||
self.assertTrue(card["overall_pass"])
|
||||
|
||||
def test_missing_baseline_is_stale_not_fabricated_as_fresh(self):
|
||||
card = build_scorecard(_full_evidence())
|
||||
result = check_baseline_stale(card, OLD_HEAD)
|
||||
self.assertTrue(result["stale"])
|
||||
self.assertFalse(result["usable_as_pass_basis"])
|
||||
self.assertIn("baseline_head_missing", result["reason"])
|
||||
self.assertIsNone(result["baseline_head"])
|
||||
|
||||
def test_missing_current_head_is_stale_fail_closed(self):
|
||||
card = build_scorecard(_full_evidence(), baseline_head=OLD_HEAD)
|
||||
result = check_baseline_stale(card, None)
|
||||
self.assertTrue(result["stale"])
|
||||
self.assertFalse(result["usable_as_pass_basis"])
|
||||
self.assertIn("current_main_head_missing", result["reason"])
|
||||
|
||||
def test_stale_check_does_not_mutate_scorecard(self):
|
||||
card = build_scorecard(_full_evidence(), baseline_head=OLD_HEAD)
|
||||
before = dict(card)
|
||||
check_baseline_stale(card, NEW_HEAD)
|
||||
self.assertEqual(card["baseline_head"], before["baseline_head"])
|
||||
self.assertEqual(card["overall"], before["overall"])
|
||||
self.assertNotIn("stale", card)
|
||||
|
||||
|
||||
class FrozenStoreTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.conn = sqlite3.connect(":memory:")
|
||||
ensure_schema(self.conn)
|
||||
ensure_schema(self.conn) # 幂等：第二次建表不报错。
|
||||
self.conn.commit()
|
||||
|
||||
def tearDown(self):
|
||||
self.conn.close()
|
||||
|
||||
def test_freeze_and_read_back_round_trips_snapshot(self):
|
||||
card = build_scorecard(_full_evidence(), baseline_head=OLD_HEAD)
|
||||
row = freeze_scorecard(self.conn, "rubric_v1", OLD_HEAD, card,
|
||||
frozen_at=1790000000.0)
|
||||
self.conn.commit()
|
||||
self.assertEqual(row["rubric_version"], "rubric_v1")
|
||||
self.assertEqual(row["baseline_head"], OLD_HEAD)
|
||||
self.assertEqual(row["frozen_at"], 1790000000.0)
|
||||
back = get_frozen_scorecard(self.conn, "rubric_v1", OLD_HEAD)
|
||||
self.assertIsNotNone(back)
|
||||
self.assertEqual(back["baseline_head"], OLD_HEAD)
|
||||
self.assertEqual(back["scorecard_snapshot"]["baseline_head"],
|
||||
OLD_HEAD)
|
||||
self.assertEqual(back["scorecard_snapshot"]["overall"], "pass")
|
||||
self.assertEqual(back["scorecard_snapshot"]["dimensions"],
|
||||
card["dimensions"])
|
||||
|
||||
def test_advanced_head_is_a_separate_frozen_row(self):
|
||||
old_card = build_scorecard(_full_evidence(), baseline_head=OLD_HEAD)
|
||||
new_card = build_scorecard(_full_evidence(), baseline_head=NEW_HEAD)
|
||||
freeze_scorecard(self.conn, "rubric_v1", OLD_HEAD, old_card,
|
||||
frozen_at=1790000000.0)
|
||||
freeze_scorecard(self.conn, "rubric_v1", NEW_HEAD, new_card,
|
||||
frozen_at=1790000001.0)
|
||||
self.conn.commit()
|
||||
rows = list_frozen_scorecards(self.conn, rubric_version="rubric_v1")
|
||||
self.assertEqual(len(rows), 2)
|
||||
self.assertEqual(rows[0]["baseline_head"], OLD_HEAD)
|
||||
self.assertEqual(rows[1]["baseline_head"], NEW_HEAD)
|
||||
# 旧版仍可按原键读回，且新头查询对旧分判 stale。
|
||||
old_back = get_frozen_scorecard(self.conn, "rubric_v1", OLD_HEAD)
|
||||
stale = check_baseline_stale(old_back["scorecard_snapshot"],
|
||||
NEW_HEAD)
|
||||
self.assertTrue(stale["stale"])
|
||||
self.assertFalse(stale["usable_as_pass_basis"])
|
||||
latest = get_latest_frozen_scorecard(self.conn,
|
||||
rubric_version="rubric_v1")
|
||||
self.assertEqual(latest["baseline_head"], NEW_HEAD)
|
||||
|
||||
def test_read_missing_freeze_returns_none(self):
|
||||
self.assertIsNone(get_frozen_scorecard(self.conn, "rubric_v1",
|
||||
OLD_HEAD))
|
||||
self.assertIsNone(get_latest_frozen_scorecard(self.conn))
|
||||
self.assertEqual(list_frozen_scorecards(self.conn), [])
|
||||
|
||||
def test_freeze_rejects_bad_inputs(self):
|
||||
card = build_scorecard(_full_evidence(), baseline_head=OLD_HEAD)
|
||||
with self.assertRaises(ValueError):
|
||||
freeze_scorecard(self.conn, "", OLD_HEAD, card)
|
||||
with self.assertRaises(ValueError):
|
||||
freeze_scorecard(self.conn, "rubric_v1", "", card)
|
||||
with self.assertRaises(ValueError):
|
||||
freeze_scorecard(self.conn, "rubric_v1", OLD_HEAD,
|
||||
"not-a-dict")
|
||||
|
||||
def test_no_update_or_delete_methods_and_triggers_block_mutation(self):
|
||||
import cloud.scorecard_baseline_store as store
|
||||
|
||||
func_names = [n for n, _ in vars(store).items()
|
||||
if callable(_) and not n.startswith("_")]
|
||||
self.assertNotIn("update_frozen_scorecard", func_names)
|
||||
self.assertNotIn("delete_frozen_scorecard", func_names)
|
||||
self.assertFalse(any(n.startswith("update_") or
|
||||
n.startswith("delete_") for n in func_names))
|
||||
card = build_scorecard(_full_evidence(), baseline_head=OLD_HEAD)
|
||||
freeze_scorecard(self.conn, "rubric_v1", OLD_HEAD, card)
|
||||
self.conn.commit()
|
||||
with self.assertRaises(sqlite3.DatabaseError):
|
||||
self.conn.execute(
|
||||
"UPDATE scorecard_baseline_freeze SET baseline_head = ? "
|
||||
"WHERE baseline_head = ?", (NEW_HEAD, OLD_HEAD))
|
||||
self.conn.rollback()
|
||||
with self.assertRaises(sqlite3.DatabaseError):
|
||||
self.conn.execute(
|
||||
"DELETE FROM scorecard_baseline_freeze "
|
||||
"WHERE baseline_head = ?", (OLD_HEAD,))
|
||||
self.conn.rollback()
|
||||
rows = list_frozen_scorecards(self.conn)
|
||||
self.assertEqual(len(rows), 1)
|
||||
self.assertEqual(rows[0]["baseline_head"], OLD_HEAD)
|
||||
|
||||
|
||||
class NoVersionControlProbeTestCase(unittest.TestCase):
|
||||
def test_production_code_does_not_probe_version_control(self):
|
||||
root = Path(__file__).resolve().parents[1]
|
||||
for name in ("audit_scorecard.py", "scorecard_baseline_store.py"):
|
||||
src = (root / name).read_text(encoding="utf-8")
|
||||
self.assertNotIn("subprocess", src, name)
|
||||
self.assertNotIn("rev-parse", src, name)
|
||||
self.assertNotIn("import os", src, name)
|
||||
tree = ast.parse((root / "scorecard_baseline_store.py"
|
||||
).read_text(encoding="utf-8"))
|
||||
funcs = [n.name for n in ast.walk(tree)
|
||||
if isinstance(n, ast.FunctionDef)]
|
||||
self.assertFalse(any(n.startswith("update_") or
|
||||
n.startswith("delete_") for n in funcs))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user