host
f1f8c8bf62
- cloud/audit_sampling.py (pure, stdlib): mandatory sampling (1.0) for rebuild cases and for third-failure near-complete cases whose continuation round still failed; random sampling for the rest (normal 0.10, critical-path 0.50, ceil so small groups are never rounded to zero, seed required and recorded for reproducibility). - Wrong-verdict handling: correct the original conclusion and retrigger the corresponding flow (wrong pushback -> cancel_pushback_and_rerun_audit; wrong rebuild -> halt_rebuild_and_restore; wrong pass -> reopen_audit_and_pushback; unknown original -> escalate_to_human); every result, correct or corrected, is appended to the audit-of-audit record. - Golden-set approval checklist: list reviewed, criteria defined, sample exclusion, approver+timestamp -- all human-confirmed before approval; sample entries (sample=True or sample- ids) are dropped and counted, never counted as golden. - metrics.build_report gains an optional sampling_plan section; without a plan it states sampling is pending, never implying the audit was re-examined.
174 lines
7.0 KiB
Python
174 lines
7.0 KiB
Python
"""Tests for cloud.audit_sampling: audit-of-audit sampling rules.
|
|
|
|
Every rule has a positive and a negative case. Only the standard
|
|
library; no real model, no network, no secrets.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(
|
|
os.path.dirname(os.path.abspath(__file__)))))
|
|
|
|
from cloud.audit_sampling import ( # noqa: E402
|
|
append_audit_record, filter_golden_tasks,
|
|
golden_approval_checklist, handle_sampling_result,
|
|
mandatory_trigger, select_sample)
|
|
|
|
|
|
def _case(case_id, **kw):
|
|
case = {"id": case_id}
|
|
case.update(kw)
|
|
return case
|
|
|
|
|
|
class TestMandatoryTriggers(unittest.TestCase):
|
|
def test_rebuild_is_always_mandatory(self):
|
|
self.assertEqual(
|
|
mandatory_trigger(_case("r", decision="rebuild")), "rebuild")
|
|
|
|
def test_third_failure_near_complete_still_failed_is_mandatory(self):
|
|
case = _case("n", decision="fail_pushback", failure_count=3,
|
|
root_cause="near_complete", continuation_failed=True)
|
|
self.assertEqual(mandatory_trigger(case),
|
|
"third_failure_near_complete_still_failed")
|
|
|
|
def test_near_complete_continuation_success_is_not_mandatory(self):
|
|
case = _case("n", decision="pass", failure_count=3,
|
|
root_cause="near_complete", continuation_failed=False)
|
|
self.assertIsNone(mandatory_trigger(case))
|
|
|
|
def test_second_failure_near_complete_is_not_mandatory(self):
|
|
case = _case("n", decision="fail_pushback", failure_count=2,
|
|
root_cause="near_complete", continuation_failed=True)
|
|
self.assertIsNone(mandatory_trigger(case))
|
|
|
|
def test_structural_third_failure_is_not_this_trigger(self):
|
|
case = _case("n", decision="rebuild", failure_count=3,
|
|
root_cause="structural", continuation_failed=True)
|
|
# It IS mandatory, but via the rebuild trigger, not the
|
|
# near-complete trigger -- the trigger name must be exact.
|
|
self.assertEqual(mandatory_trigger(case), "rebuild")
|
|
|
|
def test_ordinary_case_is_not_mandatory(self):
|
|
self.assertIsNone(mandatory_trigger(
|
|
_case("o", decision="pass", failure_count=1)))
|
|
|
|
|
|
class TestRandomSample(unittest.TestCase):
|
|
def _cases(self):
|
|
normal = [_case("n%d" % i, decision="pass") for i in range(20)]
|
|
critical = [_case("c%d" % i, decision="pass", critical_path=True)
|
|
for i in range(10)]
|
|
return normal + critical
|
|
|
|
def test_ratios_and_reproducibility(self):
|
|
plan = select_sample(self._cases(), seed="seed-1")
|
|
# ceil(0.10 * 20) normal, ceil(0.50 * 10) critical.
|
|
self.assertEqual(len(plan["random_normal"]), 2)
|
|
self.assertEqual(len(plan["random_critical_path"]), 5)
|
|
self.assertEqual(plan["seed"], "seed-1")
|
|
again = select_sample(self._cases(), seed="seed-1")
|
|
self.assertEqual(plan["sampled_ids"], again["sampled_ids"])
|
|
|
|
def test_mandatory_cases_are_included_not_double_counted(self):
|
|
cases = [_case("r", decision="rebuild"),
|
|
_case("n0", decision="pass")]
|
|
plan = select_sample(cases, seed="s")
|
|
self.assertIn("r", plan["sampled_ids"])
|
|
self.assertEqual(plan["mandatory"][0]["trigger"], "rebuild")
|
|
self.assertEqual(len(plan["sampled_ids"]),
|
|
len(set(plan["sampled_ids"])))
|
|
|
|
def test_small_group_is_never_rounded_to_zero(self):
|
|
plan = select_sample([_case("only", decision="pass")], seed="s")
|
|
self.assertEqual(plan["random_normal"], ["only"])
|
|
|
|
def test_missing_seed_is_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
select_sample(self._cases(), seed="")
|
|
|
|
|
|
class TestGoldenApproval(unittest.TestCase):
|
|
def test_sample_entries_never_count_as_golden(self):
|
|
tasks = [{"id": "sample-1"}, {"id": "real-1", "sample": True},
|
|
{"id": "real-2"}]
|
|
filtered = filter_golden_tasks(tasks)
|
|
self.assertEqual([t["id"] for t in filtered["golden"]],
|
|
["real-2"])
|
|
self.assertEqual(filtered["dropped_samples"], 2)
|
|
|
|
def test_unapproved_checklist_is_pending(self):
|
|
checklist = golden_approval_checklist([{"id": "real-2"}])
|
|
self.assertFalse(checklist["approved"])
|
|
self.assertTrue(any(i["status"] == "pending_human"
|
|
for i in checklist["items"]))
|
|
|
|
def test_fully_confirmed_checklist_is_approved(self):
|
|
checklist = golden_approval_checklist(
|
|
[{"id": "real-2"}], approver="human",
|
|
approved_at="2026-10-31", list_reviewed=True,
|
|
criteria_defined=True)
|
|
self.assertTrue(checklist["approved"])
|
|
|
|
def test_sample_only_set_cannot_be_approved(self):
|
|
checklist = golden_approval_checklist(
|
|
[{"id": "sample-1"}], approver="human",
|
|
approved_at="2026-10-31", list_reviewed=True,
|
|
criteria_defined=True)
|
|
self.assertFalse(checklist["approved"])
|
|
self.assertEqual(checklist["golden_tasks"], [])
|
|
|
|
|
|
class TestSamplingResultHandling(unittest.TestCase):
|
|
def test_correct_verdict_only_records(self):
|
|
entry = handle_sampling_result(
|
|
_case("a", decision="pass"), "correct")
|
|
self.assertIsNone(entry["corrected_decision"])
|
|
self.assertIsNone(entry["retrigger"])
|
|
log = append_audit_record([], entry)
|
|
self.assertEqual(len(log), 1)
|
|
self.assertEqual(log[0]["recorded_in"], "audit_of_audit")
|
|
|
|
def test_wrong_pushback_is_cancelled_and_rerun(self):
|
|
entry = handle_sampling_result(
|
|
_case("a", decision="fail_pushback"), "incorrect",
|
|
corrected_verdict="pass")
|
|
self.assertEqual(entry["retrigger"],
|
|
"cancel_pushback_and_rerun_audit")
|
|
self.assertEqual(entry["corrected_decision"], "pass")
|
|
|
|
def test_wrong_rebuild_is_halted_and_restored(self):
|
|
entry = handle_sampling_result(
|
|
_case("a", decision="rebuild"), "incorrect",
|
|
corrected_verdict="fail_pushback")
|
|
self.assertEqual(entry["retrigger"], "halt_rebuild_and_restore")
|
|
|
|
def test_wrong_pass_reopens_and_pushes_back(self):
|
|
entry = handle_sampling_result(
|
|
_case("a", decision="pass"), "incorrect",
|
|
corrected_verdict="fail_pushback")
|
|
self.assertEqual(entry["retrigger"],
|
|
"reopen_audit_and_pushback")
|
|
|
|
def test_unknown_original_decision_escalates(self):
|
|
entry = handle_sampling_result(
|
|
_case("a", decision="mystery"), "incorrect",
|
|
corrected_verdict="pass")
|
|
self.assertEqual(entry["retrigger"], "escalate_to_human")
|
|
|
|
def test_incorrect_without_correction_is_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
handle_sampling_result(_case("a", decision="pass"),
|
|
"incorrect")
|
|
|
|
def test_invalid_verdict_is_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
handle_sampling_result(_case("a", decision="pass"),
|
|
"maybe")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|