host
a433460343
Single sampling rule; track 3 (#244) is authoritative for WHO is mandatory, track 5 (#238) for HOW the rest are drawn: - Mandatory (1.0): golden-task records + rebuild / refreeze / escalation records + third-failure near-complete still-failed cases (mandatory_triggers, full list per case). Track 3's old deterministic every-10th ordinary rule is removed (it drew zero from batches under 10 ordinary records and was predictable); track 5's old narrow mandatory list is expanded with golden / refreeze / escalate. blocked_quota stays non-mandatory. - Random tiers: normal 0.10, critical-path 0.50, seed required and recorded, ceil so a non-empty group is never rounded to zero. 0.10/0.50 are TEST DEFAULTS, not approved production values; production rates remain a user decision. - One pass, no double counting: mandatory records leave the random pools before drawing; a multi-trigger record (e.g. golden + rebuild) is sampled once with all triggers listed, and its correction follows the (original, corrected) verdict pair, never the trigger list. - cloud/audit_pushback.py::select_audit_sample is now only a record-shaped adapter over cloud/audit_sampling.py::select_sample (seed required, retired reason names critical_path-for-kinds and periodic_sample removed); AUDIT_SAMPLE_EVERY_NTH deleted. - Correction: original never mutated, correction appended, corrector != signer enforced in audit_sampling when both names are supplied; gate effect pinned (correction to pass never unblocks by itself -- schedule/cancel-and-rerun, blocked until a new passing re-audit; correction to fail re-blocks at once). Track 3 / track 5 retrigger vocabularies documented as synonyms. - Skill text (tightening 3 item 2 + correction paragraph) rewritten to the unified rule; new test file pins the boundaries (mandatory union, dedup, ceil incl. the 9-record case, seed reproducibility, golden-set construction under both sample vocabularies, adapter order/non-mutation, correction preservation on both tracks).
254 lines
11 KiB
Python
254 lines
11 KiB
Python
"""Tests for the UNIFIED audit-of-audit sampling rule (gap 2).
|
|
|
|
This file pins the alignment of track 3 (#244) and track 5 (#238)
|
|
only; it does not re-test either track's other semantics.
|
|
|
|
Unified rule under test (see ``cloud/audit_sampling.py``):
|
|
|
|
- Mandatory (1.0) = golden-task records + rebuild / refreeze /
|
|
escalation records + third-failure near-complete still-failed
|
|
cases. Scope from #244, expanded with #238's third-failure
|
|
trigger; #244's deterministic every-10th ordinary rule is gone.
|
|
- The rest by seeded random draw: normal 0.10, critical-path
|
|
0.50, ceil so a non-empty group is never rounded to zero.
|
|
The 0.10 / 0.50 values are TEST DEFAULTS, not approved
|
|
production rates.
|
|
- One pass, no double counting: mandatory records leave the
|
|
random pools before drawing; a multi-trigger record is sampled
|
|
once with all its triggers listed; correction follows the
|
|
(original, corrected) verdict pair, not the trigger list.
|
|
- Correction preserves the original: the original record is never
|
|
mutated, the correction is a separate appended record, and the
|
|
corrector must differ from the original signer.
|
|
"""
|
|
|
|
import copy
|
|
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_pushback import ( # noqa: E402
|
|
build_audit_correction, select_audit_sample)
|
|
from cloud.audit_sampling import ( # noqa: E402
|
|
append_audit_record, build_golden_task_ids, handle_sampling_result,
|
|
mandatory_trigger, mandatory_triggers, select_sample)
|
|
|
|
|
|
def _case(case_id, **kw):
|
|
case = {"id": case_id}
|
|
case.update(kw)
|
|
return case
|
|
|
|
|
|
class TestMandatoryUnion(unittest.TestCase):
|
|
def test_all_mandatory_triggers_sampled(self):
|
|
cases = [
|
|
_case("g-push", task_id="gold-1", kind="pushback"),
|
|
_case("r", decision="rebuild"),
|
|
_case("rf", kind="refreeze"),
|
|
_case("esc", kind="escalate"),
|
|
_case("third", decision="fail_pushback", failure_count=3,
|
|
root_cause="near_complete", continuation_failed=True),
|
|
_case("n0", decision="pass"),
|
|
]
|
|
plan = select_sample(cases, seed="s", golden_task_ids={"gold-1"})
|
|
by_id = {e["case_id"]: e["triggers"] for e in plan["mandatory"]}
|
|
self.assertEqual(by_id["g-push"], ["golden_task"])
|
|
self.assertEqual(by_id["r"], ["rebuild"])
|
|
self.assertEqual(by_id["rf"], ["refreeze"])
|
|
self.assertEqual(by_id["esc"], ["escalate"])
|
|
self.assertEqual(by_id["third"],
|
|
["third_failure_near_complete_still_failed"])
|
|
self.assertIn("n0", plan["random_normal"])
|
|
self.assertEqual(len(plan["sampled_ids"]),
|
|
len(set(plan["sampled_ids"])))
|
|
|
|
def test_blocked_quota_is_not_mandatory(self):
|
|
# Track 3's mandatory kind list is rebuild/refreeze/escalate
|
|
# only; a quota-blocked record reaches the random tiers.
|
|
case = _case("q", kind="blocked_quota")
|
|
self.assertEqual(mandatory_triggers(case), [])
|
|
plan = select_sample([case], seed="s")
|
|
self.assertEqual(plan["mandatory"], [])
|
|
self.assertEqual(plan["random_normal"], ["q"])
|
|
|
|
def test_golden_task_ordinary_record_is_mandatory(self):
|
|
# Golden membership is per task, not per record kind: even
|
|
# a plain pass record of a golden task is mandatory.
|
|
case = _case("g-pass", task_id="gold-1", kind="pushback",
|
|
decision="pass")
|
|
plan = select_sample([case], seed="s",
|
|
golden_task_ids={"gold-1"})
|
|
self.assertEqual(plan["mandatory"][0]["triggers"],
|
|
["golden_task"])
|
|
|
|
|
|
class TestNoDoubleCounting(unittest.TestCase):
|
|
def test_golden_rebuild_sampled_once_with_all_triggers(self):
|
|
cases = [_case("g-re", task_id="gold-1", kind="rebuild")]
|
|
cases += [_case("n%d" % i, decision="pass") for i in range(20)]
|
|
plan = select_sample(cases, seed="s", golden_task_ids={"gold-1"})
|
|
self.assertEqual(plan["sampled_ids"].count("g-re"), 1)
|
|
self.assertEqual(plan["mandatory"][0]["triggers"],
|
|
["golden_task", "rebuild"])
|
|
# The mandatory record consumed no random quota: the draw
|
|
# is still ceil(0.10 * 20) == 2 over the ordinary pool.
|
|
self.assertEqual(len(plan["random_normal"]), 2)
|
|
self.assertNotIn("g-re", plan["random_normal"])
|
|
# Legacy single-trigger view reports the first trigger.
|
|
self.assertEqual(
|
|
mandatory_trigger(cases[0], {"gold-1"}), "golden_task")
|
|
|
|
def test_triage_and_later_rebuild_are_two_records_not_double(self):
|
|
# Deduplication is per record: a triage record and the
|
|
# later rebuild record for the same task are two samples.
|
|
cases = [
|
|
_case("triage", decision="fail_pushback", failure_count=3,
|
|
root_cause="near_complete", continuation_failed=True),
|
|
_case("rebuild", decision="rebuild"),
|
|
]
|
|
plan = select_sample(cases, seed="s")
|
|
self.assertEqual(sorted(plan["sampled_ids"]),
|
|
["rebuild", "triage"])
|
|
|
|
|
|
class TestRandomTiersAndCeil(unittest.TestCase):
|
|
def test_ratios_and_seed_reproducibility(self):
|
|
cases = [_case("n%d" % i, decision="pass") for i in range(20)]
|
|
cases += [_case("c%d" % i, decision="pass", critical_path=True)
|
|
for i in range(10)]
|
|
plan = select_sample(cases, seed="seed-1")
|
|
self.assertEqual(len(plan["random_normal"]), 2)
|
|
self.assertEqual(len(plan["random_critical_path"]), 5)
|
|
again = select_sample(cases, seed="seed-1")
|
|
self.assertEqual(plan["sampled_ids"], again["sampled_ids"])
|
|
|
|
def test_small_groups_never_round_to_zero(self):
|
|
self.assertEqual(len(select_sample(
|
|
[_case("only", decision="pass")], seed="s")["random_normal"]),
|
|
1)
|
|
# 9 ordinary records: the old every-10th rule drew ZERO
|
|
# here; ceil(0.10 * 9) == 1 under the unified rule.
|
|
self.assertEqual(len(select_sample(
|
|
[_case("n%d" % i, decision="pass") for i in range(9)],
|
|
seed="s")["random_normal"]), 1)
|
|
self.assertEqual(len(select_sample(
|
|
[_case("n%d" % i, decision="pass") for i in range(19)],
|
|
seed="s")["random_normal"]), 2)
|
|
self.assertEqual(len(select_sample(
|
|
[_case("c0", decision="pass", critical_path=True)],
|
|
seed="s")["random_critical_path"]), 1)
|
|
self.assertEqual(len(select_sample(
|
|
[_case("c%d" % i, decision="pass", critical_path=True)
|
|
for i in range(3)], seed="s")["random_critical_path"]), 2)
|
|
|
|
def test_missing_seed_rejected(self):
|
|
with self.assertRaises(ValueError):
|
|
select_sample([_case("n", decision="pass")], seed="")
|
|
|
|
|
|
class TestGoldenSetConstruction(unittest.TestCase):
|
|
def test_build_golden_ids_applies_both_vocabularies(self):
|
|
tasks = [
|
|
{"id": "gold-1", "human_approved_by": "user"},
|
|
{"id": "sample-2", "human_approved_by": "user"},
|
|
{"id": "gold-2", "sample": True, "human_approved_by": "user"},
|
|
{"id": "gold-3", "is_example": True,
|
|
"human_approved_by": "user"},
|
|
{"id": "gold-4"},
|
|
]
|
|
self.assertEqual(build_golden_task_ids(tasks), ["gold-1"])
|
|
|
|
|
|
class TestAdapter(unittest.TestCase):
|
|
def test_adapter_returns_unified_selection_in_record_order(self):
|
|
records = [
|
|
{"task_id": "gold-1", "kind": "pushback"},
|
|
{"task_id": "gold-1", "kind": "fix"},
|
|
{"task_id": "gold-1", "kind": "rebuild"},
|
|
{"task_id": "t-rf", "kind": "refreeze"},
|
|
{"task_id": "t-esc", "kind": "escalate"},
|
|
]
|
|
records += [{"task_id": "t-%d" % i, "kind": "pushback"}
|
|
for i in range(9)]
|
|
before = copy.deepcopy(records)
|
|
golden = set(build_golden_task_ids(
|
|
[{"id": "gold-1", "human_approved_by": "user"}]))
|
|
selected = select_audit_sample(records, golden, seed="s")
|
|
by_record = {id(e["record"]): e for e in selected}
|
|
# Every golden record is mandatory, whatever its kind; the
|
|
# golden rebuild record carries both triggers, once.
|
|
self.assertEqual(by_record[id(records[0])]["reasons"],
|
|
["golden_task"])
|
|
self.assertEqual(by_record[id(records[1])]["reasons"],
|
|
["golden_task"])
|
|
self.assertEqual(by_record[id(records[2])]["reasons"],
|
|
["golden_task", "rebuild"])
|
|
self.assertEqual(by_record[id(records[3])]["reason"], "refreeze")
|
|
self.assertEqual(by_record[id(records[4])]["reason"], "escalate")
|
|
# 9 ordinary records contribute ceil(0.10 * 9) == 1.
|
|
ordinary = [e for e in selected
|
|
if e["record"]["task_id"].startswith("t-")
|
|
and e["record"]["task_id"] not in ("t-rf", "t-esc")]
|
|
self.assertEqual(len(ordinary), 1)
|
|
self.assertEqual(ordinary[0]["reason"], "random_normal")
|
|
# Selected entries keep original record order.
|
|
order = [records.index(e["record"]) for e in selected]
|
|
self.assertEqual(order, sorted(order))
|
|
# Input records are never mutated by selection.
|
|
self.assertEqual(records, before)
|
|
with self.assertRaises(ValueError):
|
|
select_audit_sample(records, golden)
|
|
|
|
|
|
class TestCorrectionPreservesOriginal(unittest.TestCase):
|
|
def test_track3_correction_appends_without_mutating(self):
|
|
original = {"task_id": "t-1", "round": 0, "verdict": "pass",
|
|
"signer": "auditor-a", "entry_hash": "h" * 64}
|
|
before = dict(original)
|
|
correction, errors = build_audit_correction(
|
|
original, "fail", "auditor-b", "sample re-run found ci red")
|
|
self.assertEqual(errors, [])
|
|
self.assertEqual(original, before)
|
|
self.assertEqual(correction["original_entry_hash"], "h" * 64)
|
|
self.assertEqual(correction["restored_action"],
|
|
"retrigger_pushback_and_reblock")
|
|
log = [original, correction]
|
|
self.assertEqual(log[0]["verdict"], "pass")
|
|
_bad, bad_errors = build_audit_correction(
|
|
original, "fail", "auditor-a", "same signer")
|
|
self.assertTrue(bad_errors)
|
|
|
|
def test_track5_correction_appends_without_mutating(self):
|
|
case = {"id": "a", "decision": "pass"}
|
|
before = dict(case)
|
|
entry = handle_sampling_result(
|
|
case, "incorrect", corrected_verdict="fail_pushback",
|
|
corrector="auditor-b", original_signer="auditor-a")
|
|
# Track 5 vocabulary for the same wrong-pass pair as track
|
|
# 3's retrigger_pushback_and_reblock (synonyms, see the
|
|
# unified rule in cloud/audit_sampling.py).
|
|
self.assertEqual(entry["retrigger"], "reopen_audit_and_pushback")
|
|
self.assertEqual(case, before)
|
|
log = []
|
|
log2 = append_audit_record(log, entry)
|
|
self.assertEqual(log, [])
|
|
self.assertEqual(len(log2), 1)
|
|
with self.assertRaises(ValueError):
|
|
handle_sampling_result(
|
|
case, "incorrect", corrected_verdict="fail_pushback",
|
|
corrector="auditor-a", original_signer="auditor-a")
|
|
|
|
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"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|