tests/test_v8_risk_audit.py
import json
from pathlib import Path
import numpy as np
import pytest
from mcss.v8_risk_audit import (
build_anchor_risk_fields,
deterministic_source_derangement,
evaluate_selective_risk,
midrank_percentile,
paired_bootstrap_mean_ci,
read_v8_risk_cache,
write_v8_risk_cache,
)
def _points(depth: np.ndarray) -> np.ndarray:
zeros = np.zeros_like(depth)
return np.stack((zeros, zeros, depth), axis=-1).astype(np.float32)
def test_midrank_percentile_is_stable_and_tie_aware() -> None:
values = np.asarray([4.0, 1.0, 1.0, 8.0], dtype=np.float32)
valid = np.asarray([True, True, True, False])
ranked = midrank_percentile(values, valid)
assert ranked[0] == pytest.approx(5.0 / 6.0)
assert ranked[1] == pytest.approx(1.0 / 3.0)
assert ranked[2] == pytest.approx(1.0 / 3.0)
assert np.isnan(ranked[3])
def test_anchor_risk_fields_keep_emvsnet_as_the_action_point() -> None:
full_depth = np.asarray([[1.0, 1.0, 1.0, 1.0]], dtype=np.float32)
uni_depth = np.asarray([[1.0, 1.10, 1.30, 1.0]], dtype=np.float32)
drops = [
_points(np.asarray([[1.0, 1.0, 1.0, 1.40]], dtype=np.float32)),
_points(np.asarray([[1.0, 1.0, 1.0, 1.45]], dtype=np.float32)),
_points(np.asarray([[1.0, 1.0, 1.0, 1.35]], dtype=np.float32)),
]
valid = np.ones_like(full_depth, dtype=np.bool_)
fields = build_anchor_risk_fields(
emvsnet_points=_points(full_depth),
unidepth_points=_points(uni_depth),
emvsnet_native_risk=np.asarray([[0.1, 0.2, 0.3, 0.4]], dtype=np.float32),
unidepth_native_risk=np.asarray([[0.1, 0.2, 0.3, 0.4]], dtype=np.float32),
emvsnet_valid=valid,
unidepth_valid=valid,
drop_points=drops,
drop_native_risks=[np.full_like(full_depth, value) for value in (0.1, 0.2, 0.3)],
drop_valids=[valid, valid, valid],
cell_width_m=0.25,
)
np.testing.assert_array_equal(fields["comparison_valid"], valid)
np.testing.assert_allclose(fields["action_points_m"], _points(full_depth))
assert fields["disagreement_raw"][0, 2] == pytest.approx(1.2)
assert fields["drop_instability_raw"][0, 3] == pytest.approx(1.6)
assert fields["type_code"][0, 0] == 1 # model-supported
assert fields["type_code"][0, 2] == 3 # conflict
assert fields["type_code"][0, 3] == 2 # uncertain
expected = np.maximum.reduce(
[
fields["emvsnet_native_rank"],
fields["unidepth_native_rank"],
fields["disagreement_risk"],
fields["drop_instability_risk"],
]
)
np.testing.assert_allclose(fields["full_risk"], expected)
def test_drop_invalidity_is_risk_evidence_not_a_reason_to_delete_the_pixel() -> None:
depth = np.asarray([[1.0, 1.0, 1.0, 1.0]], dtype=np.float32)
supplier_valid = np.ones_like(depth, dtype=np.bool_)
first_drop_valid = supplier_valid.copy()
first_drop_valid[0, 1] = False
fields = build_anchor_risk_fields(
emvsnet_points=_points(depth),
unidepth_points=_points(depth),
emvsnet_native_risk=np.asarray([[0.1, 0.2, 0.3, 0.4]], dtype=np.float32),
unidepth_native_risk=np.asarray([[0.1, 0.2, 0.3, 0.4]], dtype=np.float32),
emvsnet_valid=supplier_valid,
unidepth_valid=supplier_valid,
drop_points=[_points(depth), _points(depth), _points(depth)],
drop_native_risks=[np.full_like(depth, value) for value in (0.1, 0.2, 0.3)],
drop_valids=[first_drop_valid, supplier_valid, supplier_valid],
cell_width_m=0.25,
)
assert fields["comparison_valid"][0, 1]
assert not fields["all_drop_valid"][0, 1]
assert fields["drop_instability_risk"][0, 1] == pytest.approx(1.0)
assert fields["full_risk"][0, 1] == pytest.approx(1.0)
assert fields["type_code"][0, 1] == 2 # uncertain
def test_selective_metrics_reward_correct_error_ordering() -> None:
errors = np.asarray([0.05, 0.10, 0.40, 0.80], dtype=np.float64)
valid = np.ones(4, dtype=np.bool_)
good = evaluate_selective_risk(errors, errors, valid, coverages=(0.5, 1.0))
bad = evaluate_selective_risk(errors[::-1], errors, valid, coverages=(0.5, 1.0))
assert good["aurc"] < bad["aurc"]
assert good["ause"] == pytest.approx(0.0)
assert good["spearman"] == pytest.approx(1.0)
assert bad["spearman"] == pytest.approx(-1.0)
assert good["risk_curve"][0] == pytest.approx(0.075)
assert good["aurc"] == pytest.approx(
np.mean([0.05, 0.075, (0.05 + 0.10 + 0.40) / 3.0, 0.3375])
)
assert good["aurc_definition"] == "empirical_mean_over_all_retained_counts"
def test_source_derangement_never_keeps_a_source_in_place() -> None:
first = deterministic_source_derangement("scene", (1, 2, 3, 4))
second = deterministic_source_derangement("scene", (1, 2, 3, 4))
assert first == second
assert sorted(first) == [1, 2, 3]
assert all(index != value for index, value in enumerate(first, start=1))
def test_paired_bootstrap_reports_window_level_difference() -> None:
result = paired_bootstrap_mean_ci(
np.asarray([-0.4, -0.2, -0.3]),
replicates=1000,
seed=7,
)
assert result["sample_count"] == 3
assert result["mean"] == pytest.approx(-0.3)
assert result["ci95"][1] < 0.0
def test_risk_cache_is_label_free_sealed_and_overwrite_refusing(tmp_path: Path) -> None:
fields = {
"comparison_valid": np.ones((2, 3), dtype=np.bool_),
"full_risk": np.arange(6, dtype=np.float32).reshape(2, 3),
"type_code": np.ones((2, 3), dtype=np.uint8),
}
destination = tmp_path / "risk"
write_v8_risk_cache(
destination,
fields,
metadata={"scene_id": "scene", "label_access": False},
)
loaded, metadata = read_v8_risk_cache(destination)
np.testing.assert_array_equal(loaded["full_risk"], fields["full_risk"])
assert metadata["sealed"] is True
assert metadata["label_access"] is False
assert "error" not in json.dumps(metadata).lower()
with pytest.raises(FileExistsError, match="overwrite"):
write_v8_risk_cache(destination, fields, metadata={"label_access": False})