tests/test_context_observability.py
"""Synthetic geometry only; no real query media, optimizer or holdout reads."""
from dataclasses import replace
import numpy as np
import pytest
import torch
from mcss.dynamic.types import hash_scene_state
from mcss.geometry import generate_rays
from mcss.mechanism_pilot.context_observability import audit_context_observability
from mcss.mechanism_pilot.direct_capacity_contracts import ObservationBatch, StateSealBarrier
from mcss.mechanism_pilot.direct_capacity_optimization import DirectState
from mcss.mechanism_pilot.visibility import common_visibility_masks
from mcss.types import Cameras
def fixture(n_views=3, sealed=True):
k = torch.tensor([[2.0, 0.0, 1.0], [0.0, 2.0, 1.0], [0.0, 0.0, 1.0]], dtype=torch.float64)
pose = torch.eye(4, dtype=torch.float64)
camera = Cameras(k, pose, (3, 3))
_, rays = generate_rays(camera)
depth = 2 / rays[..., 2]
query = ObservationBatch(
"s",
(9,),
torch.zeros(1, 1, 3, 3, 3),
Cameras(k[None, None], pose[None, None], (3, 3)),
depth[None, None, None],
"SEALED_QUERY_EVALUATION",
)
context = ObservationBatch(
"s",
tuple(range(n_views)),
torch.zeros(1, n_views, 3, 3, 3),
Cameras(
k[None, None].repeat(1, n_views, 1, 1),
pose[None, None].repeat(1, n_views, 1, 1),
(3, 3),
),
depth[None, None, None].repeat(1, n_views, 1, 1, 1),
"CONTEXT_ONLY_RGBD",
)
bounds = torch.tensor([[-3.0, -3.0, -3.0], [3.0, 3.0, 3.0]], dtype=torch.float64)
state = DirectState(bounds, 2).state()
barrier = StateSealBarrier(["sealed"])
if sealed:
barrier.seal("sealed", "s", state)
return dict(
barrier=barrier,
scene_id="s",
query=query,
context=context,
bounds=bounds,
anchor_c2w=context.cameras.c2w[0, 0],
allowed_scene_ids=["s"],
holdout_scene_ids=["protected"],
)
def test_seal_and_holdout_guards_precede_query_processing():
args = fixture(sealed=False)
args["query"] = None
with pytest.raises(PermissionError, match="All planned"):
audit_context_observability(**args)
args = fixture()
args["scene_id"] = "protected"
args["query"] = None
with pytest.raises(PermissionError, match="Holdout"):
audit_context_observability(**args)
def test_all_views_visible_full_arrays_and_state_immutable():
args = fixture()
before = hash_scene_state(args["barrier"].state("sealed"))
result = audit_context_observability(**args)
assert np.all(result["obs_class"] == 2)
assert np.all(result["visible_view_count"] == 3)
assert result["world_xyz"].shape == (3, 3, 3)
assert result["projected_uv"].shape == (3, 3, 3, 2)
assert np.allclose(result["world_xyz"][1, 1], [0, 0, 2])
assert np.all(result["bounds_inside"])
assert np.allclose(result["max_triangulation_angle"], 0, atol=1e-5)
assert np.all(result["triangulation_angle_bin"] == 0)
assert before == hash_scene_state(args["barrier"].state("sealed"))
def test_visible_occluded_conflict_mutually_exclusive_and_old_visibility_exact():
args = fixture()
depth = args["context"].depth.clone()
depth[:, 1] *= 0.5
depth[:, 2] *= 1.5
args["context"] = replace(args["context"], depth=depth)
result = audit_context_observability(**args)
assert np.all(result["obs_class"] == 1)
assert np.all(result["visible_support"][0])
assert np.all(result["occluded_by_context_surface"][1])
assert np.all(result["depth_conflict_front"][2])
assert np.isnan(result["max_triangulation_angle"]).all()
assert np.all(result["triangulation_angle_bin"] == -1)
q = args["query"]
qc = Cameras(q.cameras.intrinsics[0, 0], q.cameras.c2w[0, 0], (3, 3))
pairs = [
(
Cameras(
args["context"].cameras.intrinsics[0, j], args["context"].cameras.c2w[0, j], (3, 3)
),
depth[0, j, 0],
)
for j in range(3)
]
old = common_visibility_masks(qc, q.depth[0, 0, 0], pairs, pairs)
assert np.array_equal(old["depth_consistent_common"].numpy(), result["visible_support"].any(0))
def test_invalid_depth_keeps_pixel_and_missing_context_depth_never_visible():
args = fixture(1)
qdepth = args["query"].depth.clone()
qdepth[0, 0, 0, 0, 0] = float("nan")
args["query"] = replace(args["query"], depth=qdepth)
cdepth = args["context"].depth.clone()
cdepth[0, 0, 0, 1, 1] = 0
args["context"] = replace(args["context"], depth=cdepth)
result = audit_context_observability(**args)
assert result["obs_class"][0, 0] == -1
assert np.array_equal(result["pixel_xy"][0, 0], [0, 0])
assert np.isnan(result["world_xyz"][0, 0]).all()
assert result["obs_class"][1, 1] == 0
assert not result["context_depth_valid"][0, 1, 1]
assert result["query_valid"].sum() == 8
def test_triangulation_visible_pairs_and_anchor_bounds_coordinates():
args = fixture(2)
poses = args["context"].cameras.c2w.clone()
poses[0, 0, 0, 3], poses[0, 1, 0, 3] = -0.1, 0.1
args["context"] = replace(
args["context"], cameras=Cameras(args["context"].cameras.intrinsics, poses, (3, 3))
)
args["anchor_c2w"] = poses[0, 0]
args["bounds"] = torch.tensor([[0.05, -0.1, 1.9], [0.15, 0.1, 2.1]], dtype=torch.float64)
result = audit_context_observability(**args)
assert result["visible_view_count"][1, 1] == 2
assert np.isclose(result["max_triangulation_angle"][1, 1], np.degrees(2 * np.arctan(0.1 / 2)))
assert result["triangulation_angle_bin"][1, 1] == 1
assert result["max_triangulation_angle_bin"][1, 1] == 1
assert result["median_triangulation_angle_bin"][1, 1] == 1
assert np.isclose(result["camera_baseline"][1, 1], 0.2)
assert result["bounds_inside"][1, 1]
assert np.isclose(result["nearest_context_distance"][1, 1], np.sqrt(4.01))
def test_reject_oracle_supervision_and_wrong_anchor():
args = fixture()
args["query"] = replace(
args["query"], supervision="DIAGNOSTIC_QUERY_SUPERVISED_ORACLE_NOT_CONTEXT_ONLY"
)
with pytest.raises(PermissionError, match="sealed query"):
audit_context_observability(**args)
args = fixture()
args["anchor_c2w"] = args["anchor_c2w"].clone()
args["anchor_c2w"][0, 3] = 1
with pytest.raises(PermissionError, match="anchor"):
audit_context_observability(**args)
def cli_module():
import importlib.util
from pathlib import Path
spec = importlib.util.spec_from_file_location(
"observability_cli", Path(__file__).parents[1] / "scripts/audit_context_observability.py"
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def test_region_sufficient_statistics_preserve_empty_and_overall():
cli = cli_module()
arrays = audit_context_observability(**fixture())
row = dict(scene_id="s", role="A", variant="S0", method="direct", query_id=9)
depth = arrays["query_gt_depth"] * 1.2
rows, error = cli.region_rows(row, depth, arrays)
by_name = {r["region"]: r for r in rows}
assert by_name["OVERALL"]["pixel_count"] == 9
assert np.isclose(by_name["OVERALL"]["absrel_mean"], 0.2)
assert by_name["OBS0"]["absrel_mean"] is None
assert by_name["OBS0"]["absrel_sum"] == 0
assert by_name["OBS0"]["total_valid"] == 9
assert len(rows) == 11
summary = cli.observability_summary("s", "A", 9, arrays, error)
assert summary["spearman_visible_count_absrel"] is None
assert summary["region_counts"]["OBS2PLUS"] == 9
def test_region_correlation_all_valid_pixels_and_nonfinite_predictions_rejected():
cli = cli_module()
arrays = audit_context_observability(**fixture())
arrays["visible_view_count"] = np.arange(9).reshape(3, 3)
errors = np.arange(9).reshape(3, 3).astype(float)
result = cli.observability_summary("s", "A", 9, arrays, errors)
assert np.isclose(result["spearman_visible_count_absrel"], 1)
with pytest.raises(ValueError, match="Finite depth"):
cli.region_rows({}, np.full((3, 3), np.nan), arrays)
def test_previous_prediction_must_exist_in_old_seal_and_match_hash(tmp_path):
import hashlib
cli = cli_module()
path = tmp_path / "prediction.npz"
path.write_bytes(b"fixture")
digest = hashlib.sha256(b"fixture").hexdigest()
assert cli.verify_previous(path, {str(path): {"sha256": digest}}) == digest
with pytest.raises(PermissionError, match="seal"):
cli.verify_previous(path, {})
path.write_bytes(b"changed")
with pytest.raises(PermissionError, match="changed"):
cli.verify_previous(path, {str(path): {"sha256": digest}})
def test_formal_barrier_delegates_gate_before_any_checkpoint_load(tmp_path, monkeypatch):
cli = cli_module()
calls = []
def blocked(root, phase):
assert phase == "formal"
calls.append(phase)
raise PermissionError("All136 must complete")
monkeypatch.setattr(cli, "ensure_phase", blocked)
monkeypatch.setattr(
torch, "load", lambda *a, **kw: pytest.fail("No checkpoint load before gate")
)
with pytest.raises(PermissionError, match="All136"):
cli.sealed_formal_barrier(tmp_path, {}, {})
assert calls == ["formal"]