code-to10 v7 mech1: scorecard rubric versioning, append-only rubric + snapshot tables
(cherry picked from commit 368475243367685bf9db9335ec929958de031ca6)
This commit is contained in:
host committed
1 parent
c1d2a57486
commit
76f67ce310
3 files changed
+684
-6
No files matched your search
@@ -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()
|
||||
Reference in new issue
Block a user