Files
Jyotisha/tests/test_minute_rectification_holdout_intake.py
T
Jesse_Chen 3dc2af0c96
Independent Staging Quality Gate / validate (push) Successful in 10m22s
Independent Staging Quality Gate / publish (push) Successful in 10m17s
docs(rectification): audit two holdout dates and open the v4 intake
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-29 00:54:37 +08:00

113 lines
4.3 KiB
Python

import hashlib
import json
from copy import deepcopy
from pathlib import Path
from scripts.minute_rectification_holdout_intake import DEFAULT_INTAKE, append_case
from scripts.minute_rectification_holdout_validator import DEFAULT_MANIFEST, case_errors
V3_HOLDOUT = Path(__file__).resolve().parents[1] / "references" / "real_case_calibration" / "minute_rectification_holdout_v3.json"
EXPECTED_V4_INTAKE_CASE_IDS = {
"barack_obama_1961_aa_v4_holdout",
"paul_ryan_1970_aa_v4_holdout",
"sean_lennon_1975_aa_v4_holdout",
"kurt_cobain_1967_aa_v4_holdout",
}
def _reviewed_case() -> dict:
case = deepcopy(json.loads(DEFAULT_MANIFEST.read_text(encoding="utf-8"))["cases"][0])
case["case_id"] = "new-reviewed-case"
case["adjudicator"] = "independent-reviewer"
case["independent_human_reviewed"] = True
case["frozen_before_scoring"] = True
case["false_minute_commitments"] = [
{
"offset_minutes": offset,
"commitment_hash": hashlib.sha256(f"control:{offset}".encode()).hexdigest(),
}
for offset in case["false_minute_offsets"]
]
return case
def _queue(path: Path) -> None:
path.write_text(json.dumps({
"schema_version": "minute-rectification-holdout-v4-intake",
"minimum_gate": {
"events_per_case": 3,
"domains_per_case": 2,
"independent_event_sources_per_case": 2,
"negative_minutes_per_case": 4,
"day_precision_events_per_case": 3,
},
"cases": [],
}), encoding="utf-8")
def test_intake_appends_reviewed_case_but_keeps_release_blocked(tmp_path: Path) -> None:
path = tmp_path / "intake.json"
_queue(path)
report = append_case(path, _reviewed_case())
data = json.loads(path.read_text(encoding="utf-8"))
assert report["appended"] is True
assert report["verified_minute_claim_allowed"] is False
assert data["production_tuning_allowed"] is False
assert data["verified_minute_claim_allowed"] is False
assert data["cases"][0]["ingested_at"].endswith("Z")
def test_intake_rejects_missing_review_and_commitments(tmp_path: Path) -> None:
path = tmp_path / "intake.json"
_queue(path)
case = _reviewed_case()
case["independent_human_reviewed"] = False
case["false_minute_commitments"] = []
report = append_case(path, case)
assert report["appended"] is False
assert "independent_review_not_attested" in report["errors"]
assert "false_minute_commitments_do_not_match_offsets" in report["errors"]
def test_default_intake_is_non_production_and_contains_day_precision_v3_cases() -> None:
data = json.loads(DEFAULT_INTAKE.read_text(encoding="utf-8"))
assert {case["case_id"] for case in data["cases"]} == EXPECTED_V4_INTAKE_CASE_IDS
assert len(data["cases"]) == 4
assert data["production_tuning_allowed"] is False
assert data["verified_minute_claim_allowed"] is False
def test_year_precision_v3_cases_are_rejected_for_insufficient_day_precision() -> None:
v3 = json.loads(V3_HOLDOUT.read_text(encoding="utf-8"))
gate = json.loads(DEFAULT_INTAKE.read_text(encoding="utf-8"))["minimum_gate"]
rejected: dict[str, list[str]] = {}
for case in v3["cases"]:
payload = deepcopy(case)
payload.update({
"adjudicator": "v3_sealed_set_post_audit_migration",
"independent_human_reviewed": True,
"frozen_before_scoring": True,
"false_minute_commitments": [
{
"offset_minutes": offset,
"commitment_hash": hashlib.sha256(f"control:{offset}".encode()).hexdigest(),
}
for offset in case["false_minute_offsets"]
],
})
errors = case_errors(payload, gate, require_review_safeguards=True)
if errors:
rejected[str(case["case_id"])] = errors
assert "insufficient_day_precision_events" in rejected["john_robbins_1947_aa_v4_holdout"]
assert "insufficient_day_precision_events" in rejected["iwao_takamoto_1925_aa_v4_holdout"]
assert "insufficient_day_precision_events" in rejected["angelina_jolie_1975_aa_v4_holdout"]
for case_id in EXPECTED_V4_INTAKE_CASE_IDS:
assert case_id not in rejected
assert all("insufficient_day_precision_events" in errors for errors in rejected.values())