196 lines
7.9 KiB
Python
196 lines
7.9 KiB
Python
"""code-to10 v6 A8：audit_loop_driver 缺口补测。
|
||
|
||
补测基线（v5 头 10023fb 全量覆盖率，audit_loop_driver.py
|
||
496 语句缺 116 行，77%）：本文件只补下列纯函数分支，其余
|
||
缺口行以全量复测为准：
|
||
- 338：LoopState.rebuilds_used 属性。
|
||
- 436、440：_normalize_failures 的非 str、非 dict 项回落
|
||
（str(item) 作 check）。
|
||
- 456-457：_as_result_map 的列表项映射（id/check + passed）。
|
||
- 490-507：_validate_test_changes_local 的 modify_assertion
|
||
双缺口、audit 未登记、add 放行、未知动作、非 dict 项。
|
||
- 528-562：_classify_root_cause_local 的 proof/条款/多轮
|
||
归因与 unknown 回落、非 dict 项跳过。
|
||
|
||
只加测试，不改生产代码；直接测纯函数本身（不 mock 被测
|
||
代码、不加 no-coverage pragma、不 skip）。
|
||
"""
|
||
|
||
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.audit_loop_driver import (
|
||
_as_result_map,
|
||
_classify_root_cause_local,
|
||
_normalize_failures,
|
||
_validate_test_changes_local,
|
||
new_loop_state,
|
||
)
|
||
except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.audit_loop_driver import (
|
||
_as_result_map,
|
||
_classify_root_cause_local,
|
||
_normalize_failures,
|
||
_validate_test_changes_local,
|
||
new_loop_state,
|
||
)
|
||
|
||
|
||
class RebuildsUsedTestCase(unittest.TestCase):
|
||
def test_rebuilds_used_mirrors_rebuild_no(self):
|
||
state = new_loop_state("m1", "v1")
|
||
self.assertEqual(state.rebuilds_used, 0)
|
||
state.rebuild_no = 1
|
||
self.assertEqual(state.rebuilds_used, 1)
|
||
|
||
|
||
class NormalizeFailuresTestCase(unittest.TestCase):
|
||
def test_str_and_dict_pass_through(self):
|
||
self.assertEqual(_normalize_failures(["ci_green"]),
|
||
[{"check": "ci_green"}])
|
||
self.assertEqual(_normalize_failures([{"check": "x", "n": 1}]),
|
||
[{"check": "x", "n": 1}])
|
||
|
||
def test_non_str_non_dict_falls_back_to_str(self):
|
||
# 436/440：数字、None 等非 str 非 dict 项以 str(item) 作 check。
|
||
self.assertEqual(_normalize_failures([123, None]),
|
||
[{"check": "123"}, {"check": "None"}])
|
||
# dict 项被复制而非原样引用：改动结果不污染输入。
|
||
src = {"check": "x"}
|
||
out = _normalize_failures([src])
|
||
out[0]["check"] = "y"
|
||
self.assertEqual(src["check"], "x")
|
||
|
||
|
||
class AsResultMapTestCase(unittest.TestCase):
|
||
def test_dict_passes_through_as_copy(self):
|
||
src = {"t1": True}
|
||
out = _as_result_map(src)
|
||
self.assertEqual(out, {"t1": True})
|
||
out["t1"] = False
|
||
self.assertTrue(src["t1"])
|
||
|
||
def test_list_entries_mapped_by_id_then_check(self):
|
||
# 456-457：列表项按 id 优先、check 兜底映射 passed 布尔值。
|
||
values = [{"id": "t1", "passed": True},
|
||
{"check": "c2", "passed": False},
|
||
{"id": "t3", "passed": 1}]
|
||
self.assertEqual(_as_result_map(values),
|
||
{"t1": True, "c2": False, "t3": True})
|
||
|
||
def test_empty_and_none_map_to_empty(self):
|
||
self.assertEqual(_as_result_map([]), {})
|
||
self.assertEqual(_as_result_map(None), {})
|
||
|
||
|
||
class ValidateTestChangesLocalTestCase(unittest.TestCase):
|
||
def test_add_needs_nothing(self):
|
||
self.assertEqual(_validate_test_changes_local(
|
||
[{"action": "add", "test_id": "t9"}], False, []), [])
|
||
|
||
def test_modify_assertion_without_refreeze_and_audit(self):
|
||
# 双缺口：既无重冻结、又无 separate audit，各记一条。
|
||
violations = _validate_test_changes_local(
|
||
[{"action": "modify_assertion", "test_id": "t1"}], False, [])
|
||
self.assertEqual(len(violations), 2)
|
||
self.assertTrue(any("without contract re-freeze" in v
|
||
for v in violations))
|
||
self.assertTrue(any("without separate audit" in v
|
||
for v in violations))
|
||
|
||
def test_modify_assertion_with_unrecorded_audit(self):
|
||
# 有重冻结但 audit id 未登记：只记未登记一条。
|
||
violations = _validate_test_changes_local(
|
||
[{"action": "modify_assertion", "test_id": "t1",
|
||
"separate_audit_id": "audit-9"}], True, ["audit-1"])
|
||
self.assertEqual(violations,
|
||
["modify_assertion:t1 separate audit not recorded"])
|
||
|
||
def test_modify_assertion_fully_approved_passes(self):
|
||
self.assertEqual(_validate_test_changes_local(
|
||
[{"action": "modify_assertion", "test_id": "t1",
|
||
"separate_audit_id": "audit-9"}], True, ["audit-9"]), [])
|
||
|
||
def test_delete_without_refreeze(self):
|
||
violations = _validate_test_changes_local(
|
||
[{"action": "delete", "test_id": "t1",
|
||
"separate_audit_id": "audit-9"}], False, ["audit-9"])
|
||
self.assertEqual(violations,
|
||
["delete:t1 without contract re-freeze"])
|
||
|
||
def test_unknown_action_and_non_dict_item(self):
|
||
violations = _validate_test_changes_local(
|
||
[{"action": "rename", "test_id": "t1"}, "not-a-dict"], True, [])
|
||
self.assertEqual(len(violations), 2)
|
||
self.assertTrue(any("unknown test change action:rename" in v
|
||
for v in violations))
|
||
self.assertTrue(any("unknown test change action:None" in v
|
||
for v in violations))
|
||
|
||
|
||
class ClassifyRootCauseLocalTestCase(unittest.TestCase):
|
||
def test_non_dict_items_skipped_to_unknown(self):
|
||
self.assertEqual(_classify_root_cause_local(
|
||
[["not-a-dict"], [123]]), "unknown")
|
||
|
||
def test_proof_in_two_rounds_wins(self):
|
||
rounds = [
|
||
[{"root_cause": "contract", "proved_unimplementable": True}],
|
||
[{"root_cause": "contract", "proved_unimplementable": True}],
|
||
]
|
||
self.assertEqual(_classify_root_cause_local(rounds), "contract")
|
||
|
||
def test_single_proof_round_does_not_win(self):
|
||
# 仅一轮 proof：proof 门槛（>=2 轮）未满，回落 unknown。
|
||
rounds = [[{"root_cause": "contract",
|
||
"proved_unimplementable": True}]]
|
||
self.assertEqual(_classify_root_cause_local(rounds), "unknown")
|
||
|
||
def test_same_clause_in_two_rounds_is_contract(self):
|
||
rounds = [
|
||
[{"root_cause": "design",
|
||
"evidence": {"contract_clause": "C1"}}],
|
||
[{"root_cause": "design",
|
||
"evidence": {"contract_clause": "C1"}}],
|
||
]
|
||
self.assertEqual(_classify_root_cause_local(rounds), "contract")
|
||
|
||
def test_clause_only_in_implementation_does_not_count(self):
|
||
# 条款只在 implementation 项上出现时不计入条款轮次。
|
||
rounds = [
|
||
[{"root_cause": "implementation",
|
||
"evidence": {"contract_clause": "C1"}}],
|
||
[{"root_cause": "implementation",
|
||
"evidence": {"contract_clause": "C1"}}],
|
||
]
|
||
self.assertEqual(_classify_root_cause_local(rounds),
|
||
"implementation")
|
||
|
||
def test_two_contract_rounds(self):
|
||
rounds = [[{"root_cause": "contract"}],
|
||
[{"root_cause": "contract"}]]
|
||
self.assertEqual(_classify_root_cause_local(rounds), "contract")
|
||
|
||
def test_two_design_rounds(self):
|
||
rounds = [[{"root_cause": "design"}],
|
||
[{"root_cause": "design"}]]
|
||
self.assertEqual(_classify_root_cause_local(rounds), "design")
|
||
|
||
def test_two_implementation_rounds(self):
|
||
rounds = [[{"root_cause": "implementation"}],
|
||
[{"root_cause": "implementation"}]]
|
||
self.assertEqual(_classify_root_cause_local(rounds),
|
||
"implementation")
|
||
|
||
def test_one_round_each_is_unknown(self):
|
||
rounds = [[{"root_cause": "contract"}],
|
||
[{"root_cause": "design"}]]
|
||
self.assertEqual(_classify_root_cause_local(rounds), "unknown")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|