From 8fedd7fba07b531ee98ae2e64280e838e0a4ddd1 Mon Sep 17 00:00:00 2001 From: host Date: Mon, 2 Nov 2026 15:30:13 +0000 Subject: [PATCH] v7 mech-5: scorecard baseline_head machine field, stale check, frozen store (append-only) (cherry picked from commit e3e1f517e916eaac597b82d04bac5307a3f8b126) --- cloud/audit_scorecard.py | 83 +++++- cloud/scorecard_baseline_store.py | 175 +++++++++++++ cloud/tests/test_scorecard_mech_baseline.py | 274 ++++++++++++++++++++ 3 files changed, 531 insertions(+), 1 deletion(-) create mode 100644 cloud/scorecard_baseline_store.py create mode 100644 cloud/tests/test_scorecard_mech_baseline.py diff --git a/cloud/audit_scorecard.py b/cloud/audit_scorecard.py index b34be1e..1299151 100644 --- a/cloud/audit_scorecard.py +++ b/cloud/audit_scorecard.py @@ -27,6 +27,21 @@ #247 纠正为 pass 本身不解封； #244 CONDITIONAL 先映射为 fail 再记（本模块只接受已映射的结论， 若调用方传入 verdict="conditional"，记为未实测/无效证据，不算通过）。 +- 基线漂移门禁（v7 机制第 5 项，本模块新增）： + build_scorecard(..., baseline_head=None) 的 baseline_head 是机器字段， + 只能由调用方显式传入当前 main 头；本模块绝不在生产代码里探测 + 版本控制状态、绝不读版本控制目录、绝不编造一个头。 + 不传入（或传入空串）时 baseline_head 为 None，并在 missing_items + 里记 baseline_head_missing，不把缺失读成“与当前一致”。 + check_baseline_stale(scorecard, current_main_head) 是唯一的漂移判定： + current_main_head != baseline_head 时 stale=True，且该旧分整体结论 + 不得再作 pass 依据（返回里以 usable_as_pass_basis=False 明示， + 调用方凭此字段判断）。stale 是附加字段：本项绝不改写 + build_scorecard 的 overall / overall_pass 旧语义，旧值保留以便追溯， + 测试断言 stale 字段而非改写旧 overall。 + 冻结版的服务端存储见 cloud/scorecard_baseline_store.py（只追加、 + 不提供改删旧冻结行的方法）；阈值仍一律标 test_default_unapproved， + 除非调用方显式传入 user_approved。 7 维定义（不要擅自增维、减维、改名换义）： 1. delivery_quality 交付质量 @@ -463,7 +478,8 @@ _SCORERS = { } -def build_scorecard(evidence, thresholds=None, rubric_version=None): +def build_scorecard(evidence, thresholds=None, rubric_version=None, + baseline_head=None): """算出 7 维评分卡。evidence 缺键按空证据处理（该维记 None）。 rubric_version：调用方显式传入的标尺版本号（见 @@ -471,9 +487,18 @@ def build_scorecard(evidence, thresholds=None, rubric_version=None): 未传入时卡片记 rubric_version=None，表示“未锚定到版本化 标尺”，分数仍按 thresholds（或默认值）算出，但不具备 跨标尺可比性。计分规则、7 维定义、达标规则一字未改。 + + baseline_head：调用方显式传入的当前 main 头（机器字段，不是 + docstring 里的文字说明）。不传入或传入空串时归一化为 None， + 并在 missing_items 末尾记 baseline_head_missing；本函数绝不 + 编造、绝不探测版本控制状态。baseline 缺失本身不改写 overall + 语义（overall 仍按 7 维算），漂移与否由 check_baseline_stale + 单独判定，调用方不得只看 overall 就把旧分当现分用。 """ evidence = evidence or {} th = thresholds or {} + if baseline_head is not None and baseline_head == "": + baseline_head = None dimensions = [] missing_all, untested_all = [], [] for dim_id, _name in DIMENSIONS: @@ -481,6 +506,8 @@ def build_scorecard(evidence, thresholds=None, rubric_version=None): 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"]) + if baseline_head is None: + missing_all.append("baseline_head_missing") scores = [d["score"] for d in dimensions if d["score"] is not None] average = _round1(sum(scores) / len(scores)) if scores else None # 总体规则钉死：平均分绝不能作为达标依据。 @@ -493,6 +520,7 @@ def build_scorecard(evidence, thresholds=None, rubric_version=None): "overall": "pass" if overall_pass else "not_pass", "overall_pass": overall_pass, "pass_score": PASS_SCORE, + "baseline_head": baseline_head, "missing_items": missing_all, "untested_items": untested_all, "thresholds_source": thresholds_source(th), @@ -501,6 +529,56 @@ def build_scorecard(evidence, thresholds=None, rubric_version=None): } +def check_baseline_stale(scorecard, current_main_head): + """判定一张旧评分卡是否因 main 推进而失效（基线漂移门禁）。 + + 参数全部由调用方传入：scorecard 是 build_scorecard 的输出 dict， + current_main_head 是调用方当前确认的 main 头。本函数不在生产 + 代码里探测版本控制状态，也不读版本控制目录。 + + 返回 dict（调用方凭 stale / usable_as_pass_basis 判断，绝不能 + 只看旧卡的 overall / overall_pass）： + - stale=True 时：旧分不得再作 pass 依据 + （usable_as_pass_basis=False、overall_usable_as_pass_basis=False、 + overall_must_not_be_used_as_pass_basis=True），即使旧卡的 + overall 本身仍是 "pass"——旧值保留只为追溯，不代表现分有效。 + - stale 的三种情形：baseline_head 缺失（无法证明新鲜，失败关闭， + 不编造为一致）；current_main_head 缺失（无法核对，失败关闭）； + 两者均有值但不相等（main 已推进，旧分过期）。 + - 仅当两者均有值且相等时 stale=False，旧分仍可按原 overall 读。 + """ + scorecard = scorecard or {} + baseline = scorecard.get("baseline_head") + if baseline is not None and baseline == "": + baseline = None + if current_main_head is not None and current_main_head == "": + current_main_head = None + if baseline is None: + reason = ("baseline_head_missing: scorecard carries no frozen " + "baseline_head, freshness cannot be proven") + stale = True + elif current_main_head is None: + reason = ("current_main_head_missing: caller supplied no current " + "main head, drift cannot be checked") + stale = True + elif current_main_head != baseline: + reason = ("baseline_head_drifted: baseline_head=%s " + "current_main_head=%s" % (baseline, current_main_head)) + stale = True + else: + reason = "baseline_head_matches_current_main_head" + stale = False + return { + "stale": stale, + "baseline_head": baseline, + "current_main_head": current_main_head, + "reason": reason, + "usable_as_pass_basis": not stale, + "overall_usable_as_pass_basis": not stale, + "overall_must_not_be_used_as_pass_basis": stale, + } + + def example_scorecard_as_of_2026_10_31(): """2026-10-31 快照示例卡（历史示例，不是当前系统得分）。 @@ -512,6 +590,9 @@ def example_scorecard_as_of_2026_10_31(): 判断分。阈值仍为测试默认值 （source="test_default_unapproved"），未拍板。 总体结论必然是不达标。 + 本示例无冻结基线：build_scorecard 未传 baseline_head，故 + baseline_head=None 且 missing_items 含 baseline_head_missing， + 绝不编造一个头把它写成现分。 """ evidence = { "delivery_quality": { diff --git a/cloud/scorecard_baseline_store.py b/cloud/scorecard_baseline_store.py new file mode 100644 index 0000000..f2fc55f --- /dev/null +++ b/cloud/scorecard_baseline_store.py @@ -0,0 +1,175 @@ +"""评分卡冻结版的服务端存储（v7 机制第 5 项：基线漂移门禁）。 + +定位：只存“冻结的那一版”，不算分、不判 stale、不改旧分。 +算分在 cloud/audit_scorecard.py 的 build_scorecard， +漂移判定在 check_baseline_stale，本模块只做冻结行的写入与读取。 + +键思路沿用仓库 DB 模式：ledger.py / audit_loop_driver.py 按 +(module_id, contract_version) 计 slot，本表按 +(rubric_version, baseline_head) 计冻结版——同一 rubric 版本在 +同一 main 头上冻结的评分卡快照，是一版；main 头推进后冻结的， +是另一版，旧版仍在表里供追溯，不被覆盖。 + +行字段： +- rubric_version：评分口径版本（调用方传入的字符串，如 "v1"； + 本模块不规定口径内容、不替用户拍板阈值）。 +- baseline_head：冻结时调用方传入的 main 头（字符串，非空）。 +- frozen_at：冻结时间戳（秒，调用方不传时取当前时间）。 +- scorecard_snapshot：build_scorecard 输出的完整 dict，以 JSON + 文本存入，读回时还原为 dict。 + +只追加，不改删： +- 本模块只提供 ensure_schema（建表，幂等）、freeze_scorecard + （INSERT 新行）、get_frozen_scorecard / get_latest_frozen_scorecard / + list_frozen_scorecards（只读）。刻意不提供任何 UPDATE / DELETE + 方法；SQLite 下另有触发器在表层挡住 UPDATE / DELETE + （写法沿用 ledger.py 的只追加触发器）。 +- 不 commit：调用方在自己的事务边界提交（与 ledger.ensure_schema + 同款约定）；独立调用方须自己 commit，否则同连接外看不到新行。 +- 真实 PG 未接通：PG 适配连接跳过 SQLite 触发器建（方言不同， + 与 ledger.py 同款处理），PG 下的只追加护栏与真实 PG 验证均未做， + 本项只用 SQLite / 内存库测试并明说。 +- 本模块绝不在生产代码里探测版本控制状态、绝不读版本控制目录： + rubric_version / baseline_head / frozen_at 全部由调用方传入。 +""" + +from __future__ import annotations + +import json +import time + +_FROZEN_TABLE_SQL = ( + "CREATE TABLE IF NOT EXISTS scorecard_baseline_freeze(" + "id INTEGER PRIMARY KEY AUTOINCREMENT, " + "rubric_version TEXT NOT NULL, " + "baseline_head TEXT NOT NULL, " + "frozen_at DOUBLE PRECISION NOT NULL, " + "scorecard_snapshot TEXT NOT NULL)" +) + +_FROZEN_INDEX_SQL = ( + "CREATE INDEX IF NOT EXISTS scorecard_baseline_freeze_key " + "ON scorecard_baseline_freeze(rubric_version, baseline_head)" +) + +# SQLite 只追加触发器（PG 不走这段，见模块 docstring）。 +_SQLITE_GUARD_SQL = ( + "CREATE TRIGGER IF NOT EXISTS scorecard_baseline_freeze_no_update " + "BEFORE UPDATE ON scorecard_baseline_freeze BEGIN " + "SELECT RAISE(ABORT, 'scorecard_baseline_freeze is append-only'); END", + "CREATE TRIGGER IF NOT EXISTS scorecard_baseline_freeze_no_delete " + "BEFORE DELETE ON scorecard_baseline_freeze BEGIN " + "SELECT RAISE(ABORT, 'scorecard_baseline_freeze is append-only'); END", +) + + +def _is_pg_conn(conn): + """PgConnection 适配连接（触发器方言不同，须跳过 SQLite 护栏）。""" + return type(conn).__name__ == "PgConnection" + + +def ensure_schema(conn): + """建冻结表 + 索引 + SQLite 只追加触发器（幂等，不 commit）。""" + conn.execute(_FROZEN_TABLE_SQL) + conn.execute(_FROZEN_INDEX_SQL) + if not _is_pg_conn(conn): + for stmt in _SQLITE_GUARD_SQL: + conn.execute(stmt) + + +def _require_non_empty_str(value, name): + if not isinstance(value, str) or value == "": + raise ValueError("%s must be a non-empty string" % name) + return value + + +def freeze_scorecard(conn, rubric_version, baseline_head, + scorecard_snapshot, frozen_at=None): + """写入一个新冻结行（INSERT only），返回读回形态的行 dict。 + + scorecard_snapshot 必须是 build_scorecard 输出的 dict；不是 dict + 时拒绝写入，不做任何格式猜测。调用方负责 commit。 + """ + _require_non_empty_str(rubric_version, "rubric_version") + _require_non_empty_str(baseline_head, "baseline_head") + if not isinstance(scorecard_snapshot, dict): + raise ValueError("scorecard_snapshot must be a dict") + stamp = float(frozen_at) if frozen_at is not None else time.time() + payload = json.dumps(scorecard_snapshot, sort_keys=True, + ensure_ascii=False) + cur = conn.execute( + "INSERT INTO scorecard_baseline_freeze" + "(rubric_version, baseline_head, frozen_at, scorecard_snapshot) " + "VALUES (?, ?, ?, ?)", + (rubric_version, baseline_head, stamp, payload)) + return { + "id": cur.lastrowid, + "rubric_version": rubric_version, + "baseline_head": baseline_head, + "frozen_at": stamp, + "scorecard_snapshot": scorecard_snapshot, + } + + +def _row_to_dict(row): + row_id, rubric_version, baseline_head, frozen_at, snapshot_text = row + return { + "id": row_id, + "rubric_version": rubric_version, + "baseline_head": baseline_head, + "frozen_at": frozen_at, + "scorecard_snapshot": json.loads(snapshot_text), + } + + +def get_frozen_scorecard(conn, rubric_version, baseline_head): + """读某 (rubric_version, baseline_head) 的最新冻结行，无则 None。 + + 同一键有多行时（不同 frozen_at 多次冻结）返回 id 最大的最新行， + 旧行仍在表里，不被覆盖、不被删除。 + """ + _require_non_empty_str(rubric_version, "rubric_version") + _require_non_empty_str(baseline_head, "baseline_head") + row = conn.execute( + "SELECT id, rubric_version, baseline_head, frozen_at, " + "scorecard_snapshot FROM scorecard_baseline_freeze " + "WHERE rubric_version = ? AND baseline_head = ? " + "ORDER BY id DESC LIMIT 1", + (rubric_version, baseline_head)).fetchone() + return _row_to_dict(row) if row is not None else None + + +def get_latest_frozen_scorecard(conn, rubric_version=None): + """读最新冻结行；传 rubric_version 时只在该口径内取最新，无则 None。""" + if rubric_version is None: + row = conn.execute( + "SELECT id, rubric_version, baseline_head, frozen_at, " + "scorecard_snapshot FROM scorecard_baseline_freeze " + "ORDER BY id DESC LIMIT 1").fetchone() + else: + _require_non_empty_str(rubric_version, "rubric_version") + row = conn.execute( + "SELECT id, rubric_version, baseline_head, frozen_at, " + "scorecard_snapshot FROM scorecard_baseline_freeze " + "WHERE rubric_version = ? ORDER BY id DESC LIMIT 1", + (rubric_version,)).fetchone() + return _row_to_dict(row) if row is not None else None + + +def list_frozen_scorecards(conn, rubric_version=None, baseline_head=None): + """按 id 升序列出冻结行；两参数均可选作过滤条件，不过滤时全列。""" + sql = ("SELECT id, rubric_version, baseline_head, frozen_at, " + "scorecard_snapshot FROM scorecard_baseline_freeze") + clauses, params = [], [] + if rubric_version is not None: + _require_non_empty_str(rubric_version, "rubric_version") + clauses.append("rubric_version = ?") + params.append(rubric_version) + if baseline_head is not None: + _require_non_empty_str(baseline_head, "baseline_head") + clauses.append("baseline_head = ?") + params.append(baseline_head) + if clauses: + sql += " WHERE " + " AND ".join(clauses) + sql += " ORDER BY id ASC" + return [_row_to_dict(row) for row in conn.execute(sql, params).fetchall()] diff --git a/cloud/tests/test_scorecard_mech_baseline.py b/cloud/tests/test_scorecard_mech_baseline.py new file mode 100644 index 0000000..33a1c58 --- /dev/null +++ b/cloud/tests/test_scorecard_mech_baseline.py @@ -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()