"""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()