File size: 1,671 Bytes
7a87926 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | from ylff.services.audit.audit_runner import run_audit
from ylff.services.audit.models import ExternalReferenceMeasurement, OperatingRegime
def _m(capture_id: str, *, err: float, sigma: float):
return ExternalReferenceMeasurement(
capture_id=capture_id,
regime=OperatingRegime.INDOOR_CONSTRAINED,
measurement_type="tag_to_tag",
d_pred=10.0 + float(err),
d_star=10.0,
sigma_d=float(sigma),
)
def test_run_audit_splits_by_capture_id_scene_disjoint():
# Two scenes, multiple measurements each. Ensure split happens at scene granularity.
ms = []
for _ in range(10):
ms.append(_m("scene_a", err=0.1, sigma=0.2))
ms.append(_m("scene_b", err=0.1, sigma=0.2))
res = run_audit(ms, calibrate=True, calibration_split_fraction=0.5)
split = res.summary.get("split", {})
assert split.get("mode") in ("scene_stratified_by_regime", "row_fallback")
if split.get("mode") == "scene_stratified_by_regime":
assert split["num_scenes_total"] == 2
assert split["num_scenes_cal"] + split["num_scenes_audit"] == 2
def test_dataset_level_gate_2b_enforced_only_when_calibrated():
# If calibration set ends up empty (e.g., only one scene and rounding), Gate 2b is skipped.
ms = [_m("scene_only", err=0.2, sigma=0.05) for _ in range(20)]
res = run_audit(ms, calibrate=True, calibration_split_fraction=0.5)
gates = {g.name: g for g in res.gates}
assert "gate_2b_dataset_level_coverage" in gates
# Skipped OR enforced is fine; what we require is the gate exists and is reported.
assert isinstance(gates["gate_2b_dataset_level_coverage"].details, dict)
|