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 2026-11-03 02:49:27 +00:00
1 parent be81d658b6
commit 8fedd7fba0
3 files changed
+531 -1

No files matched your search

+82 -1
View File
@@ -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": {
+175
View File
@@ -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()]
+274
View File
@@ -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()