"""v7 机制第 3 项：差分审计与涨幅停表定向测试。 钉住的规则（改动前先改 cloud/scorecard_diff.py 的 docstring）： - 审计范围 = 变更文件 + 声称修好的项；未变更/基线外文件不进包。 - 每轮记 code/production/product 三张表各自的分数，绝不只记平均分； 停表只看代码表，不看平均分。 - 连续两轮代码表涨幅 <0.5（阈值测试默认值、未拍板）即停表， 返回 escalate、不再自动推回；涨幅 >=0.5 或只有一轮 <0.5 不触发。 - 改标尺即新序列：不同 rubric_version 的轮次不连算。 - 真实 PG 不接通：SQLite/内存测试并明说；PG_DDL 只是模式兼容文本。 - 本模块不改 audit_loop_driver 语义、不改 #238 auto_apply、 不改 #247 纠正为 pass 不解封、不改 v1.1 契约。 """ import os import sqlite3 import sys import tempfile import unittest from pathlib import Path try: from cloud.scorecard_diff import ( PG_DDL, SQLITE_DDL, STAGNATION_THRESHOLD, ScorecardGainLedger, SqliteScorecardGainLedger, append_round, audit_scope, check_stagnation, code_gains, read_rounds, ) except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.scorecard_diff import ( PG_DDL, SQLITE_DDL, STAGNATION_THRESHOLD, ScorecardGainLedger, SqliteScorecardGainLedger, append_round, audit_scope, check_stagnation, code_gains, read_rounds, ) def _round(version, code, production=None, product=None): return {"rubric_version": version, "code": code, "production": production, "product": product} class AuditScopeTest(unittest.TestCase): def test_path_set_changed_in_scope_unchanged_out(self): scope = audit_scope( {"cloud/server.py", "cloud/foo.py"}, {"cloud/server.py", "cloud/bar.py"}, claimed_fixed=[], ) self.assertEqual(scope["in_scope"]["files"], ["cloud/bar.py", "cloud/foo.py"]) self.assertEqual(scope["out_of_scope"]["files"], ["cloud/server.py"]) self.assertNotIn("cloud/server.py", scope["in_scope"]["files"]) def test_dict_snapshot_modified_by_hash_in_scope_same_hash_out(self): scope = audit_scope( {"a.py": "hash1", "b.py": "hash2"}, {"a.py": "hash1", "b.py": "hash9"}, ) self.assertEqual(scope["in_scope"]["files"], ["b.py"]) self.assertEqual(scope["out_of_scope"]["files"], ["a.py"]) def test_added_and_removed_are_changed(self): scope = audit_scope({"old.py": "h1"}, {"new.py": "h2"}) self.assertEqual(scope["in_scope"]["files"], ["new.py", "old.py"]) self.assertEqual(scope["out_of_scope"]["files"], []) def test_baseline_outside_files_excluded_from_package(self): # 基线外文件：未变更、也不在声称修好集内 -> 不进审计包， # 只出现在 out_of_scope，不出现在 in_scope 的任何位置。 scope = audit_scope( {"cloud/server.py": "s1", "cloud/untouched.py": "u1", "cloud/other.py": "o1"}, {"cloud/server.py": "s2", "cloud/untouched.py": "u1", "cloud/other.py": "o1"}, claimed_fixed=["check_changed"], ) package = scope["in_scope"]["files"] + scope["in_scope"]["claimed_fixed"] self.assertNotIn("cloud/untouched.py", package) self.assertNotIn("cloud/other.py", package) self.assertIn("cloud/untouched.py", scope["out_of_scope"]["files"]) self.assertIn("cloud/other.py", scope["out_of_scope"]["files"]) self.assertEqual(scope["in_scope"]["files"], ["cloud/server.py"]) def test_claimed_fixed_in_scope_even_when_file_unchanged(self): scope = audit_scope( {"a.py": "h1"}, {"a.py": "h1"}, claimed_fixed=["check_x", "check_x", "check_a"], ) self.assertEqual(scope["in_scope"]["files"], []) self.assertEqual(scope["in_scope"]["claimed_fixed"], ["check_a", "check_x"]) self.assertEqual(scope["out_of_scope"]["files"], ["a.py"]) def test_files_key_form_and_none_claimed(self): scope = audit_scope({"files": ["a.py"]}, {"files": ["a.py", "b.py"]}) self.assertEqual(scope["in_scope"]["files"], ["b.py"]) self.assertEqual(scope["in_scope"]["claimed_fixed"], []) self.assertEqual(scope["out_of_scope"]["files"], ["a.py"]) def test_empty_baseline_all_current_changed(self): scope = audit_scope(None, {"a.py", "b.py"}) self.assertEqual(scope["in_scope"]["files"], ["a.py", "b.py"]) self.assertEqual(scope["out_of_scope"]["files"], []) class GainLedgerTest(unittest.TestCase): def test_memory_ledger_keeps_three_tables_not_only_average(self): ledger = ScorecardGainLedger() entry = append_round(ledger, "v1", code=8.0, production=7.0, product=9.0) self.assertEqual(entry["code"], 8.0) self.assertEqual(entry["production"], 7.0) self.assertEqual(entry["product"], 9.0) self.assertEqual(entry["average_for_reference_only"], 8.0) self.assertTrue(entry["average_is_not_pass_criterion"]) rounds = read_rounds(ledger) self.assertEqual(len(rounds), 1) self.assertEqual(rounds[0]["code"], 8.0) self.assertEqual(rounds[0]["production"], 7.0) self.assertEqual(rounds[0]["product"], 9.0) def test_average_none_when_any_table_unknown(self): ledger = ScorecardGainLedger() append_round(ledger, "v1", code=8.0, production=None, product=9.0) self.assertIsNone(read_rounds(ledger)[0]["average_for_reference_only"]) self.assertEqual(read_rounds(ledger)[0]["code"], 8.0) def test_read_returns_copies_append_stores_copies(self): ledger = ScorecardGainLedger() entry = {"rubric_version": "v1", "code": 8.0, "production": 8.0, "product": 8.0} append_round(ledger, entry=entry) entry["code"] = 0.0 self.assertEqual(read_rounds(ledger)[0]["code"], 8.0) read_rounds(ledger)[0]["code"] = 0.0 self.assertEqual(read_rounds(ledger)[0]["code"], 8.0) def test_plain_list_ledger(self): ledger = [] append_round(ledger, "v1", code=8.0, production=8.0, product=8.0) self.assertEqual(len(read_rounds(ledger)), 1) self.assertEqual(read_rounds(ledger)[0]["rubric_version"], "v1") def test_invalid_round_rejected(self): ledger = ScorecardGainLedger() with self.assertRaises(ValueError): append_round(ledger, "", code=8.0, production=8.0, product=8.0) with self.assertRaises(ValueError): append_round(ledger, "v1", code=11.0, production=8.0, product=8.0) with self.assertRaises(ValueError): append_round(ledger, "v1", code=True, production=8.0, product=8.0) self.assertEqual(read_rounds(ledger), []) def test_sqlite_memory_roundtrip_and_ddl(self): self.assertIn("scorecard_gain_rounds", SQLITE_DDL) self.assertIn("rubric_version", SQLITE_DDL) self.assertIn("code", SQLITE_DDL) ledger = SqliteScorecardGainLedger() try: append_round(ledger, "v1", code=8.0, production=7.5, product=9.0) append_round(ledger, "v1", code=8.3, production=7.5, product=9.0) rounds = read_rounds(ledger) self.assertEqual([r["code"] for r in rounds], [8.0, 8.3]) self.assertEqual(rounds[0]["production"], 7.5) self.assertEqual(rounds[1]["product"], 9.0) finally: ledger.close() def test_sqlite_file_persists_across_reopen(self): tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False) tmp.close() try: ledger = SqliteScorecardGainLedger(db_path=tmp.name) try: append_round(ledger, "v1", code=8.0, production=8.0, product=8.0) finally: ledger.close() reopened = SqliteScorecardGainLedger(db_path=tmp.name) try: self.assertEqual(len(read_rounds(reopened)), 1) self.assertEqual(read_rounds(reopened)[0]["code"], 8.0) finally: reopened.close() finally: os.unlink(tmp.name) def test_pg_ddl_is_text_only_and_mirrors_columns(self): # PG 不接通：只断言 PG_DDL 文本与 SQLite 列名对应，不连真实 PG。 for column in ("rubric_version", "code", "production", "product", "average_for_reference_only"): self.assertIn(column, PG_DDL) self.assertIn("scorecard_gain_rounds", PG_DDL) conn = sqlite3.connect(":memory:") try: conn.execute(SQLITE_DDL) finally: conn.close() class StagnationTest(unittest.TestCase): def test_threshold_is_test_default_unapproved(self): self.assertEqual(STAGNATION_THRESHOLD["value"], 0.5) self.assertEqual(STAGNATION_THRESHOLD["source"], "test_default_unapproved") def test_two_small_gains_trigger_escalate_no_pushback(self): history = [_round("v1", 8.0), _round("v1", 8.3), _round("v1", 8.5)] self.assertEqual(code_gains(history), [0.3, 0.2]) result = check_stagnation(history) self.assertEqual(result["decision"], "escalate") self.assertEqual(result["reason"], "stagnation") self.assertFalse(result["auto_pushback"]) self.assertFalse(result["escalation_grants_user_acceptance"]) self.assertEqual(result["consecutive_small_gains"], 2) self.assertEqual(result["rubric_version"], "v1") def test_gain_at_or_above_threshold_does_not_trigger(self): # 0.6 大涨幅打断连续计数：最近两次涨幅为 0.6、0.2，只一轮小涨幅。 result = check_stagnation([_round("v1", 8.0), _round("v1", 8.6), _round("v1", 8.8)]) self.assertEqual(result["decision"], "continue") self.assertEqual(result["consecutive_small_gains"], 1) # 恰好 0.5 不算 <0.5：两次 0.5 涨幅也不触发。 result = check_stagnation([_round("v1", 8.0), _round("v1", 8.5), _round("v1", 9.0)]) self.assertEqual(result["decision"], "continue") self.assertEqual(code_gains([_round("v1", 8.0), _round("v1", 8.5), _round("v1", 9.0)]), [0.5, 0.5]) def test_single_small_gain_does_not_trigger(self): result = check_stagnation([_round("v1", 8.0), _round("v1", 8.3)]) self.assertEqual(result["decision"], "continue") self.assertEqual(result["consecutive_small_gains"], 1) self.assertEqual(check_stagnation([_round("v1", 8.0)])["decision"], "continue") self.assertEqual(check_stagnation([])["decision"], "continue") def test_rubric_version_change_starts_new_sequence(self): # 旧标尺 v1 有两次小涨幅，但最新轮已改标尺为 v2：v2 序列内 # 只有一轮 0.2 涨幅，不连算旧序列，不触发。 history = [_round("v1", 8.0), _round("v1", 8.3), _round("v1", 8.5), _round("v2", 8.0), _round("v2", 8.2)] self.assertEqual(code_gains(history), [0.2]) result = check_stagnation(history) self.assertEqual(result["decision"], "continue") self.assertEqual(result["rubric_version"], "v2") # v2 序列内攒够两次小涨幅后才触发。 history.append(_round("v2", 8.4)) self.assertEqual(check_stagnation(history)["decision"], "escalate") def test_unknown_code_score_breaks_streak(self): history = [_round("v1", 8.0), _round("v1", 8.3), _round("v1", None)] self.assertEqual(code_gains(history), [0.3, None]) self.assertEqual(check_stagnation(history)["decision"], "continue") def test_negative_gain_counts_as_small_gain(self): # 代码表下滑也是未达阈值：连续两次下滑触发 escalate，不是 pushback。 result = check_stagnation([_round("v1", 9.0), _round("v1", 8.8), _round("v1", 8.6)]) self.assertEqual(code_gains([_round("v1", 9.0), _round("v1", 8.8), _round("v1", 8.6)]), [-0.2, -0.2]) self.assertEqual(result["decision"], "escalate") self.assertFalse(result["auto_pushback"]) def test_nested_scores_form_and_ledger_input(self): history = [ {"rubric_version": "v1", "scores": {"code": 8.0, "production": 8.0, "product": 8.0}}, {"rubric_version": "v1", "scores": {"code": 8.2, "production": 8.0, "product": 8.0}}, {"rubric_version": "v1", "scores": {"code": 8.4, "production": 8.0, "product": 8.0}}, ] self.assertEqual(check_stagnation(history)["decision"], "escalate") ledger = ScorecardGainLedger() append_round(ledger, "v1", code=8.0, production=8.0, product=8.0) append_round(ledger, "v1", code=8.2, production=8.0, product=8.0) append_round(ledger, "v1", code=8.4, production=8.0, product=8.0) self.assertEqual(check_stagnation(ledger)["decision"], "escalate") def test_average_gain_does_not_drive_stagnation(self): # 三表平均涨幅大、代码表涨幅小：停表只看代码表，仍触发。 history = [ _round("v1", 8.0, production=5.0, product=5.0), _round("v1", 8.2, production=9.0, product=9.0), _round("v1", 8.4, production=9.0, product=9.0), ] self.assertEqual(check_stagnation(history)["decision"], "escalate") if __name__ == "__main__": unittest.main()