From be81d658b6d547bb75c16e88bd3ad13ea8d6983a Mon Sep 17 00:00:00 2001 From: host Date: Mon, 2 Nov 2026 15:31:02 +0000 Subject: [PATCH] code-to10 v7 mech1: scorecard rubric versioning, append-only rubric + snapshot tables (cherry picked from commit 368475243367685bf9db9335ec929958de031ca6) --- cloud/audit_scorecard.py | 46 ++- cloud/scorecard_rubric.py | 282 +++++++++++++++++ cloud/tests/test_scorecard_mech_rubric.py | 362 ++++++++++++++++++++++ 3 files changed, 684 insertions(+), 6 deletions(-) create mode 100644 cloud/scorecard_rubric.py create mode 100644 cloud/tests/test_scorecard_mech_rubric.py diff --git a/cloud/audit_scorecard.py b/cloud/audit_scorecard.py index e9e3f35..b34be1e 100644 --- a/cloud/audit_scorecard.py +++ b/cloud/audit_scorecard.py @@ -44,7 +44,10 @@ - 文档与技能并入架构与契约一致性 （技能文本、附录溯源、移植规则最终都是契约一致性的一部分）。 -输入格式：build_scorecard(evidence, thresholds=None) +输入格式：build_scorecard(evidence, thresholds=None, rubric_version=None) + rubric_version 只是卡片上的版本锚点（版本化持久存储见 + cloud/scorecard_rubric.py：标尺改动写新版本行、旧行只读， + 本函数不查库、不改旧分）。 evidence: dict，键为上面 7 个维 id，值为该维的证据 dict。 每个维的证据字段见各 _score_* 函数的 docstring。 调用方必须带来源、时间、检查名；本模块不自动从测试文件、 @@ -94,7 +97,7 @@ UNTESTED_KINDS = {"fixture", "viewport_substitute", "code_review_only", "fixture_simulated", "dry_run", "estimated"} -def _thresholds_source(thresholds): +def thresholds_source(thresholds): """阈值来源：只有调用方显式传入 user_approved 才算已拍板。""" if not thresholds: return "test_default_unapproved" @@ -109,6 +112,29 @@ def _thresholds_source(thresholds): return "test_default_unapproved" +# 旧私有名的兼容别名（v2cov 评分卡测试仍从这里导入它）。 +_thresholds_source = thresholds_source + + +def _effective_thresholds(thresholds): + """实际用于计分的阈值表：默认值 + 调用方显式覆盖，逐项归一化。 + + 未被覆盖的键仍取 DEFAULT_THRESHOLDS（与 _t 的逐键回落一致）， + 因此卡片上报告的 thresholds 就是真正算分的表，不再是“恒为 + 默认值”的旧写法。数值本身仍一律标来源，未拍板的不冒充生产值。 + """ + effective = {k: dict(v) for k, v in DEFAULT_THRESHOLDS.items()} + for key, entry in (thresholds or {}).items(): + if isinstance(entry, dict): + merged = {"value": entry["value"]} + merged["source"] = entry.get("source", "test_default_unapproved") + effective[key] = merged + else: + effective[key] = {"value": entry, + "source": "test_default_unapproved"} + return effective + + def _t(thresholds, key): if thresholds and key in thresholds: entry = thresholds[key] @@ -437,8 +463,15 @@ _SCORERS = { } -def build_scorecard(evidence, thresholds=None): - """算出 7 维评分卡。evidence 缺键按空证据处理（该维记 None）。""" +def build_scorecard(evidence, thresholds=None, rubric_version=None): + """算出 7 维评分卡。evidence 缺键按空证据处理（该维记 None）。 + + rubric_version：调用方显式传入的标尺版本号（见 + cloud/scorecard_rubric.py）。本函数不查库、不定版本： + 未传入时卡片记 rubric_version=None，表示“未锚定到版本化 + 标尺”，分数仍按 thresholds（或默认值）算出，但不具备 + 跨标尺可比性。计分规则、7 维定义、达标规则一字未改。 + """ evidence = evidence or {} th = thresholds or {} dimensions = [] @@ -462,8 +495,9 @@ def build_scorecard(evidence, thresholds=None): "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()}, + "thresholds_source": thresholds_source(th), + "thresholds": _effective_thresholds(th), + "rubric_version": rubric_version, } diff --git a/cloud/scorecard_rubric.py b/cloud/scorecard_rubric.py new file mode 100644 index 0000000..801cf34 --- /dev/null +++ b/cloud/scorecard_rubric.py @@ -0,0 +1,282 @@ +"""评分卡标尺版本化：仅追加的持久存储（v7 机制第 1 项）。 + +问题：cloud/audit_scorecard.py 的 DEFAULT_THRESHOLDS 只存在于 +代码里，标尺一改，旧标尺下的分数就再也算不出来、也对不上 +——审计员每次重审都在新标尺下重算旧分，基线漂移就是这么来的。 + +解法（复用 audit_loop_driver 的键思路，不复用它的表）： +- audit_loop_driver 以 (module_id, contract_version) 为键， + 合同版本一变，计数与预算全部重置、旧版本状态原地不动。 + 这里同理以 rubric_version 为键：标尺改动 = 写一个新版本行， + 旧版本行原地不动；同一证据在旧版本下重算，分数逐字相同。 +- 不复用 audit_loop_state 表：那张表存的是循环状态 + （pushback 轮次、失败计数），标尺不是循环状态，混表会让 + 两类数据的生命周期互相牵连。故新开两张表（见下）。 + +两张表（SQLite / Postgres 兼容写法，沿用 audit_loop_wiring 的 +conn + ``?`` 占位符 + CREATE TABLE IF NOT EXISTS 模式；类型只用 +TEXT / REAL，不用任何一方专属类型，不用自增列）： +- scorecard_rubric_versions：标尺版本行，主键 rubric_version。 + 列：rubric_version、thresholds_snapshot（归一化阈值表的 + canonical JSON）、baseline_head（该版本锚定的基线提交， + 调用方显式传入，本模块不跑 git、不猜 HEAD）、 + thresholds_source（user_approved 仅当全部条目都已拍板， + 其余一律 test_default_unapproved）、created_at。 +- scorecard_snapshots：已算出的卡片行，主键 + (rubric_version, evidence_sha256)。列除主键外还有 + baseline_head 与 thresholds_snapshot（从版本行反规范化而来， + 让单行快照自带审计三要素，不必联表也能核对）、 + scorecard_json（build_scorecard 的完整输出）、created_at。 + 选型理由：只存版本行 + 现算也能复现旧分（scorer 是纯函数）， + 但快照行让“旧行分数不变”成为可直接读回的事实而非“重算 + 碰巧一致”的推论；快照不存证据原文，只存证据的 SHA-256， + 避免证据体在库里再留一份副本。 + +仅追加（append-only），这是本模块的核心承诺： +- 对外 API 只有 create / get / list / latest / score / record / + get_snapshot，没有任何改旧行、删旧行的方法；重复建同一 + 版本号直接报 ValueError（主键冲突转译，不做 upsert）。 +- 同一 (rubric_version, evidence) 重复 record 时返回已存行的 + 卡片原样，不覆盖、不合并。 +- 本模块防的是“走正常 API 的误写”（与 ledger 的触发器同款 + 诚实边界）：能直接拿连接跑 SQL 的人当然能改库，库内机制 + 拦不住超级用户；可信锚点仍须落在库外（如导出的行哈希）。 + +事务边界：沿用 audit_loop_wiring 的做法——本模块的写函数 +在写后 commit；调用方若想把多写包进同一事务，应直接用 +底层的 INSERT 语句而非本模块的写函数。读函数不 commit。 + +诚实边界： +- 阈值本身仍不由本模块拍板：thresholds_source 的判定与 + audit_scorecard 一致，测试默认值永远标 + test_default_unapproved；user_approved 只能由调用方显式 + 传入全部已拍板条目。 +- 未在真实 Postgres 上验证：SQL 只用两端通用的写法， + 并以 SQLite（文件库 + :memory:）与假连接（只录语句） + 测试；PG 接通后须以真实 PG 复测一次才能宣称两端可用。 +- 本模块不改 #238 auto_apply（保持 False）、不改 #247 + （纠正为 pass 本身不解封）、不改 v1.1 契约与 7 维定义。 +""" + +from __future__ import annotations + +import hashlib +import json +import time + +from cloud.audit_scorecard import build_scorecard, thresholds_source + +_RUBRIC_TABLE_SQL = ( + "CREATE TABLE IF NOT EXISTS scorecard_rubric_versions(" + "rubric_version TEXT PRIMARY KEY, " + "thresholds_snapshot TEXT NOT NULL, " + "baseline_head TEXT NOT NULL, " + "thresholds_source TEXT NOT NULL, " + "created_at REAL NOT NULL)" +) + +_SNAPSHOT_TABLE_SQL = ( + "CREATE TABLE IF NOT EXISTS scorecard_snapshots(" + "rubric_version TEXT NOT NULL, " + "baseline_head TEXT NOT NULL, " + "evidence_sha256 TEXT NOT NULL, " + "thresholds_snapshot TEXT NOT NULL, " + "scorecard_json TEXT NOT NULL, " + "created_at REAL NOT NULL, " + "PRIMARY KEY(rubric_version, evidence_sha256))" +) + + +def ensure_rubric_tables(conn): + """建两张表（幂等）。写法与 audit_loop_wiring 保持一致。""" + conn.execute(_RUBRIC_TABLE_SQL) + conn.execute(_SNAPSHOT_TABLE_SQL) + try: + conn.commit() + except Exception: + pass + + +def _canonical_json(value): + return json.dumps(value, sort_keys=True, separators=(",", ":")) + + +def _evidence_sha256(evidence): + return hashlib.sha256( + _canonical_json(evidence).encode("utf-8")).hexdigest() + + +def _normalize_thresholds(thresholds): + """阈值表归一化：每项 -> {"value": 数值, "source": 来源}。 + + 纯数值条目按 audit_scorecard 的旧规则视作未拍板； + dict 条目缺 source 时同样视作未拍板，绝不默认已拍板。 + """ + if not isinstance(thresholds, dict) or not thresholds: + raise ValueError("thresholds must be a non-empty dict") + normalized = {} + for key, entry in thresholds.items(): + if isinstance(entry, dict): + if "value" not in entry: + raise ValueError( + "threshold %r must carry a value" % (key,)) + value = entry["value"] + source = entry.get("source", "test_default_unapproved") + else: + value = entry + source = "test_default_unapproved" + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError( + "threshold %r value must be a number" % (key,)) + if not isinstance(source, str) or not source: + raise ValueError( + "threshold %r source must be a non-empty string" % (key,)) + normalized[key] = {"value": value, "source": source} + return normalized + + +def _row_to_version(row): + return { + "rubric_version": row[0], + "thresholds_snapshot": row[1], + "thresholds": json.loads(row[1]), + "baseline_head": row[2], + "thresholds_source": row[3], + "created_at": row[4], + } + + +def create_rubric_version(conn, rubric_version, thresholds, baseline_head, + created_at=None): + """写一个新标尺版本行。版本号已存在时报 ValueError，旧行不动。 + + thresholds：audit_scorecard 阈值表形态（dict 条目或纯数值）。 + baseline_head：调用方显式传入的基线提交标识，不可为空。 + """ + if not isinstance(rubric_version, str) or not rubric_version: + raise ValueError("rubric_version must be a non-empty string") + if not isinstance(baseline_head, str) or not baseline_head: + raise ValueError("baseline_head must be a non-empty string") + normalized = _normalize_thresholds(thresholds) + snapshot = _canonical_json(normalized) + source = thresholds_source(normalized) + stamp = time.time() if created_at is None else created_at + ensure_rubric_tables(conn) + try: + conn.execute( + "INSERT INTO scorecard_rubric_versions" + "(rubric_version, thresholds_snapshot, baseline_head," + " thresholds_source, created_at) VALUES (?, ?, ?, ?, ?)", + (rubric_version, snapshot, baseline_head, source, stamp)) + except Exception as exc: + # 只有主键冲突才转译为“版本已存在”；其它错误原样抛出， + # 绝不能把真实写失败伪装成版本冲突、更不能转成 upsert。 + text = str(exc).lower() + if "unique" in text or "duplicate" in text: + raise ValueError( + "rubric_version %r already exists; create a new version" + " instead of modifying it" % (rubric_version,)) from exc + raise + conn.commit() + return { + "rubric_version": rubric_version, + "thresholds_snapshot": snapshot, + "thresholds": normalized, + "baseline_head": baseline_head, + "thresholds_source": source, + "created_at": stamp, + } + + +def get_rubric_version(conn, rubric_version): + """读一个版本行；不存在返回 None（不抛错、不造默认值）。""" + ensure_rubric_tables(conn) + row = conn.execute( + "SELECT rubric_version, thresholds_snapshot, baseline_head," + " thresholds_source, created_at FROM scorecard_rubric_versions" + " WHERE rubric_version = ?", + (rubric_version,)).fetchone() + if row is None: + return None + return _row_to_version(row) + + +def list_rubric_versions(conn): + """全部版本行，按 (created_at, rubric_version) 升序（写序）。""" + ensure_rubric_tables(conn) + rows = conn.execute( + "SELECT rubric_version, thresholds_snapshot, baseline_head," + " thresholds_source, created_at FROM scorecard_rubric_versions" + " ORDER BY created_at ASC, rubric_version ASC").fetchall() + return [_row_to_version(row) for row in rows] + + +def latest_rubric_version(conn): + """最新版本行（created_at 最大，时间并列时版本号最大）；无行返回 None。""" + ensure_rubric_tables(conn) + row = conn.execute( + "SELECT rubric_version, thresholds_snapshot, baseline_head," + " thresholds_source, created_at FROM scorecard_rubric_versions" + " ORDER BY created_at DESC, rubric_version DESC LIMIT 1").fetchone() + if row is None: + return None + return _row_to_version(row) + + +def scorecard_for_rubric(conn, rubric_version, evidence): + """用指定版本行的快照阈值现算一张卡（只读，不写快照表）。 + + 未知版本号报 ValueError，不回落到默认值——回落会让旧分 + 在不知情下变成新标尺的分，正是版本化要杜绝的漂移。 + """ + version = get_rubric_version(conn, rubric_version) + if version is None: + raise ValueError("unknown rubric_version %r" % (rubric_version,)) + card = build_scorecard(evidence, thresholds=version["thresholds"], + rubric_version=rubric_version) + card["baseline_head"] = version["baseline_head"] + return card + + +def record_scorecard(conn, rubric_version, evidence, created_at=None): + """算卡并写入快照行；同 (版本, 证据) 已有行时返回旧行原样。 + + 返回值恒为卡片 dict（新算的或旧存的），调用方凭它核对。 + """ + version = get_rubric_version(conn, rubric_version) + if version is None: + raise ValueError("unknown rubric_version %r" % (rubric_version,)) + card = build_scorecard(evidence, thresholds=version["thresholds"], + rubric_version=rubric_version) + card["baseline_head"] = version["baseline_head"] + digest = _evidence_sha256(evidence) + ensure_rubric_tables(conn) + row = conn.execute( + "SELECT scorecard_json FROM scorecard_snapshots" + " WHERE rubric_version = ? AND evidence_sha256 = ?", + (rubric_version, digest)).fetchone() + if row is not None: + return json.loads(row[0]) + conn.execute( + "INSERT INTO scorecard_snapshots" + "(rubric_version, baseline_head, evidence_sha256," + " thresholds_snapshot, scorecard_json, created_at)" + " VALUES (?, ?, ?, ?, ?, ?)", + (rubric_version, version["baseline_head"], digest, + version["thresholds_snapshot"], _canonical_json(card), + time.time() if created_at is None else created_at)) + conn.commit() + return card + + +def get_scorecard_snapshot(conn, rubric_version, evidence): + """读一张已存快照卡；无行返回 None（不现算、不回落）。""" + digest = _evidence_sha256(evidence) + ensure_rubric_tables(conn) + row = conn.execute( + "SELECT scorecard_json FROM scorecard_snapshots" + " WHERE rubric_version = ? AND evidence_sha256 = ?", + (rubric_version, digest)).fetchone() + if row is None: + return None + return json.loads(row[0]) diff --git a/cloud/tests/test_scorecard_mech_rubric.py b/cloud/tests/test_scorecard_mech_rubric.py new file mode 100644 index 0000000..e1a2da5 --- /dev/null +++ b/cloud/tests/test_scorecard_mech_rubric.py @@ -0,0 +1,362 @@ +"""v7 机制第 1 项：评分卡标尺版本化定向测试。 + +钉住的规则： +- 标尺改动 = 写新版本行；旧版本行只读，没有改旧行、删旧行的 API。 +- 改标尺后，同一证据在旧版本下重算的分数逐字不变， + 在新版本下独立计分；已存的旧快照行字节不变。 +- 阈值仍不由本模块拍板：默认快照一律 test_default_unapproved， + user_approved 只能由调用方显式传入全部已拍板条目。 +- 以下阈值只用于区分新旧标尺的测试，不代表生产标准。 + +证据设计：coverage_pct=0.87 在默认标尺 +（coverage_nine=0.85、ten=0.90）下得 9.4 分， +在收紧标尺（nine=0.95、ten=0.98）下得 7.4 分， +其余子项全部满分，使 test_coverage 维的分数只由标尺决定。 +""" + +import json +import os +import sqlite3 +import sys +import tempfile +import unittest +from pathlib import Path + +try: + import cloud.scorecard_rubric as rubric + from cloud.audit_scorecard import ( + DEFAULT_THRESHOLDS, + _thresholds_source, + build_scorecard, + thresholds_source, + ) + from cloud.scorecard_rubric import ( + create_rubric_version, + get_rubric_version, + get_scorecard_snapshot, + latest_rubric_version, + list_rubric_versions, + record_scorecard, + scorecard_for_rubric, + ) +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + import cloud.scorecard_rubric as rubric + from cloud.audit_scorecard import ( + DEFAULT_THRESHOLDS, + _thresholds_source, + build_scorecard, + thresholds_source, + ) + from cloud.scorecard_rubric import ( + create_rubric_version, + get_rubric_version, + get_scorecard_snapshot, + latest_rubric_version, + list_rubric_versions, + record_scorecard, + scorecard_for_rubric, + ) + + +def _evidence(): + return { + "test_coverage": { + "prd_coverage_rate": 1.0, + "cross_matrix_coverage_rate": 1.0, + "coverage_pct": 0.87, + "cross_matrix_kind": "real", + "real_pg": True, + "real_service": True, + "agent_six_complete": True, + "lint_items": 0, + "mypy_items": 0, + "bandit_high": 0, + }, + } + + +def _strict_thresholds(): + thresholds = {k: dict(v) for k, v in DEFAULT_THRESHOLDS.items()} + thresholds["coverage_nine"] = {"value": 0.95, + "source": "test_default_unapproved"} + thresholds["coverage_ten"] = {"value": 0.98, + "source": "test_default_unapproved"} + return thresholds + + +def _dim(card, dim_id): + for dim in card["dimensions"]: + if dim["id"] == dim_id: + return dim + raise AssertionError("missing dimension %s" % dim_id) + + +class _FakeCursor: + def __init__(self, row=None): + self._row = row + + def fetchone(self): + return self._row + + def fetchall(self): + return [] if self._row is None else [self._row] + + +class _FakeConn: + """只录语句、不碰真实库的假连接：证明存储层发出的 SQL 种类。 + + get 路径返回预置的版本行；快照 SELECT 返回 None（无旧行）， + 使 record 路径走到 INSERT。commit 只记账，不执行。 + """ + + def __init__(self, version_row): + self.statements = [] + self.version_row = version_row + self.committed = 0 + + def execute(self, sql, params=()): + self.statements.append(sql) + upper = sql.strip().upper() + if upper.startswith("SELECT"): + if "FROM SCORECARD_RUBRIC_VERSIONS" in upper: + return _FakeCursor(self.version_row) + return _FakeCursor(None) + return _FakeCursor(None) + + def commit(self): + self.committed += 1 + + +class RubricVersionTestCase(unittest.TestCase): + def setUp(self): + self.conn = sqlite3.connect(":memory:") + self.conn.row_factory = sqlite3.Row + + def tearDown(self): + self.conn.close() + + def test_create_and_get_round_trip(self): + created = create_rubric_version( + self.conn, "v1", DEFAULT_THRESHOLDS, "b7ac9d7", created_at=1.0) + self.assertEqual(created["rubric_version"], "v1") + self.assertEqual(created["baseline_head"], "b7ac9d7") + self.assertEqual(created["thresholds_source"], + "test_default_unapproved") + self.assertEqual(json.loads(created["thresholds_snapshot"]), + json.loads(json.dumps( + {k: {"value": v["value"], + "source": "test_default_unapproved"} + for k, v in DEFAULT_THRESHOLDS.items()}))) + row = get_rubric_version(self.conn, "v1") + self.assertEqual(row["rubric_version"], "v1") + self.assertEqual(row["baseline_head"], "b7ac9d7") + self.assertEqual(row["thresholds_source"], + "test_default_unapproved") + self.assertEqual(row["thresholds_snapshot"], + created["thresholds_snapshot"]) + self.assertEqual(row["thresholds"]["coverage_nine"]["value"], 0.85) + self.assertIsNone(get_rubric_version(self.conn, "no-such")) + + def test_duplicate_version_raises_and_old_row_unchanged(self): + before = create_rubric_version( + self.conn, "v1", DEFAULT_THRESHOLDS, "b7ac9d7", created_at=1.0) + with self.assertRaises(ValueError): + create_rubric_version( + self.conn, "v1", _strict_thresholds(), "deadbeef", + created_at=2.0) + after = get_rubric_version(self.conn, "v1") + self.assertEqual(after["thresholds_snapshot"], + before["thresholds_snapshot"]) + self.assertEqual(after["baseline_head"], "b7ac9d7") + self.assertEqual(len(list_rubric_versions(self.conn)), 1) + + def test_list_and_latest_ordering(self): + create_rubric_version(self.conn, "v1", DEFAULT_THRESHOLDS, + "b7ac9d7", created_at=1.0) + create_rubric_version(self.conn, "v2", _strict_thresholds(), + "b7ac9d7", created_at=2.0) + versions = list_rubric_versions(self.conn) + self.assertEqual([v["rubric_version"] for v in versions], + ["v1", "v2"]) + self.assertEqual(latest_rubric_version(self.conn)["rubric_version"], + "v2") + empty = sqlite3.connect(":memory:") + try: + self.assertIsNone(latest_rubric_version(empty)) + self.assertEqual(list_rubric_versions(empty), []) + finally: + empty.close() + + def test_unknown_version_never_falls_back_to_defaults(self): + with self.assertRaises(ValueError): + scorecard_for_rubric(self.conn, "no-such", _evidence()) + with self.assertRaises(ValueError): + record_scorecard(self.conn, "no-such", _evidence()) + self.assertIsNone(get_scorecard_snapshot(self.conn, "no-such", + _evidence())) + + def test_rubric_change_old_score_unchanged_new_score_independent(self): + create_rubric_version(self.conn, "v1", DEFAULT_THRESHOLDS, + "b7ac9d7", created_at=1.0) + card_v1_before = scorecard_for_rubric(self.conn, "v1", _evidence()) + self.assertEqual(_dim(card_v1_before, "test_coverage")["score"], 9.4) + self.assertEqual(card_v1_before["rubric_version"], "v1") + self.assertEqual(card_v1_before["baseline_head"], "b7ac9d7") + # 改标尺：只写新版本行，不碰 v1 行。 + create_rubric_version(self.conn, "v2", _strict_thresholds(), + "b7ac9d7", created_at=2.0) + card_v1_after = scorecard_for_rubric(self.conn, "v1", _evidence()) + card_v2 = scorecard_for_rubric(self.conn, "v2", _evidence()) + self.assertEqual(card_v1_after, card_v1_before) + self.assertEqual(_dim(card_v2, "test_coverage")["score"], 7.4) + self.assertNotEqual( + _dim(card_v2, "test_coverage")["score"], + _dim(card_v1_after, "test_coverage")["score"]) + row_v1 = get_rubric_version(self.conn, "v1") + self.assertEqual(row_v1["thresholds"]["coverage_nine"]["value"], + 0.85) + + def test_record_and_get_snapshot(self): + create_rubric_version(self.conn, "v1", DEFAULT_THRESHOLDS, + "b7ac9d7", created_at=1.0) + recorded = record_scorecard(self.conn, "v1", _evidence()) + self.assertEqual(_dim(recorded, "test_coverage")["score"], 9.4) + stored = get_scorecard_snapshot(self.conn, "v1", _evidence()) + self.assertEqual(stored, recorded) + # 同证据同版本重复 record：返回旧行原样，不新增行。 + again = record_scorecard(self.conn, "v1", _evidence()) + self.assertEqual(again, recorded) + count = self.conn.execute( + "SELECT COUNT(*) FROM scorecard_snapshots").fetchone()[0] + self.assertEqual(count, 1) + + def test_record_under_new_rubric_leaves_old_snapshot_byte_identical(self): + create_rubric_version(self.conn, "v1", DEFAULT_THRESHOLDS, + "b7ac9d7", created_at=1.0) + record_scorecard(self.conn, "v1", _evidence()) + raw_before = self.conn.execute( + "SELECT scorecard_json FROM scorecard_snapshots" + " WHERE rubric_version = 'v1'").fetchone()[0] + create_rubric_version(self.conn, "v2", _strict_thresholds(), + "b7ac9d7", created_at=2.0) + record_scorecard(self.conn, "v2", _evidence()) + raw_after = self.conn.execute( + "SELECT scorecard_json FROM scorecard_snapshots" + " WHERE rubric_version = 'v1'").fetchone()[0] + self.assertEqual(raw_after, raw_before) + stored_v2 = get_scorecard_snapshot(self.conn, "v2", _evidence()) + self.assertEqual(_dim(stored_v2, "test_coverage")["score"], 7.4) + + def test_no_mutation_api_exists(self): + names = [n for n in dir(rubric) if not n.startswith("__")] + self.assertFalse([n for n in names + if n.startswith(("update_", "delete_", + "remove_", "set_"))]) + for missing in ("update_rubric_version", "delete_rubric_version", + "update_scorecard_snapshot", + "delete_scorecard_snapshot"): + self.assertFalse(hasattr(rubric, missing)) + + def test_fake_connection_sql_is_only_create_insert_select(self): + snapshot = json.dumps( + {k: {"value": v["value"], "source": "test_default_unapproved"} + for k, v in DEFAULT_THRESHOLDS.items()}, sort_keys=True) + version_row = ("v1", snapshot, "b7ac9d7", + "test_default_unapproved", 1.0) + fake = _FakeConn(version_row) + card = scorecard_for_rubric(fake, "v1", _evidence()) + self.assertEqual(_dim(card, "test_coverage")["score"], 9.4) + recorded = record_scorecard(fake, "v1", _evidence()) + self.assertEqual(recorded["rubric_version"], "v1") + kinds = set() + for sql in fake.statements: + head = sql.strip().upper().split()[0] + kinds.add(head) + self.assertNotIn(head, ("UPDATE", "DELETE", "DROP", "ALTER")) + self.assertIn("CREATE", kinds) + self.assertIn("INSERT", kinds) + self.assertIn("SELECT", kinds) + self.assertEqual(kinds, {"CREATE", "INSERT", "SELECT"}) + + def test_validation_rejects_bad_inputs(self): + with self.assertRaises(ValueError): + create_rubric_version(self.conn, "", DEFAULT_THRESHOLDS, + "b7ac9d7") + with self.assertRaises(ValueError): + create_rubric_version(self.conn, "v1", {}, "b7ac9d7") + with self.assertRaises(ValueError): + create_rubric_version(self.conn, "v1", {"coverage_nine": "high"}, + "b7ac9d7") + with self.assertRaises(ValueError): + create_rubric_version(self.conn, "v1", + {"coverage_nine": {"source": "x"}}, + "b7ac9d7") + with self.assertRaises(ValueError): + create_rubric_version(self.conn, "v1", DEFAULT_THRESHOLDS, "") + self.assertIsNone(get_rubric_version(self.conn, "v1")) + + def test_user_approved_snapshot_keeps_source(self): + approved = {k: {"value": v["value"], "source": "user_approved"} + for k, v in DEFAULT_THRESHOLDS.items()} + create_rubric_version(self.conn, "v-approved", approved, "b7ac9d7", + created_at=1.0) + row = get_rubric_version(self.conn, "v-approved") + self.assertEqual(row["thresholds_source"], "user_approved") + card = scorecard_for_rubric(self.conn, "v-approved", _evidence()) + self.assertEqual(card["thresholds_source"], "user_approved") + # 混入一条未拍板即整体不算已拍板（与 audit_scorecard 一致）。 + mixed = dict(approved) + mixed["coverage_nine"] = {"value": 0.85, + "source": "test_default_unapproved"} + create_rubric_version(self.conn, "v-mixed", mixed, "b7ac9d7", + created_at=2.0) + self.assertEqual( + get_rubric_version(self.conn, "v-mixed")["thresholds_source"], + "test_default_unapproved") + + def test_build_scorecard_rubric_version_and_effective_thresholds(self): + card = build_scorecard(_evidence()) + self.assertIsNone(card["rubric_version"]) + self.assertEqual(set(card["thresholds"].keys()), + set(DEFAULT_THRESHOLDS.keys())) + self.assertEqual(card["thresholds"]["coverage_nine"]["value"], 0.85) + card2 = build_scorecard(_evidence(), _strict_thresholds(), + rubric_version="v2") + self.assertEqual(card2["rubric_version"], "v2") + self.assertEqual(card2["thresholds"]["coverage_nine"]["value"], + 0.95) + # 未被覆盖的键仍取默认值（与 _t 的逐键回落一致）。 + self.assertEqual(card2["thresholds"]["coverage_floor"]["value"], + 0.75) + self.assertEqual(_dim(card2, "test_coverage")["score"], 7.4) + + def test_thresholds_source_public_alias_matches_private(self): + self.assertIs(thresholds_source, _thresholds_source) + self.assertEqual(thresholds_source(None), "test_default_unapproved") + self.assertEqual(thresholds_source({}), "test_default_unapproved") + + def test_file_db_persists_across_connections(self): + tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) + tmp.close() + try: + conn1 = sqlite3.connect(tmp.name) + create_rubric_version(conn1, "v1", DEFAULT_THRESHOLDS, + "b7ac9d7", created_at=1.0) + record_scorecard(conn1, "v1", _evidence()) + conn1.close() + conn2 = sqlite3.connect(tmp.name) + try: + row = get_rubric_version(conn2, "v1") + self.assertEqual(row["baseline_head"], "b7ac9d7") + stored = get_scorecard_snapshot(conn2, "v1", _evidence()) + self.assertEqual( + _dim(stored, "test_coverage")["score"], 9.4) + finally: + conn2.close() + finally: + os.unlink(tmp.name) + + +if __name__ == "__main__": + unittest.main()