tests/test_optimization_bounds_evaluation.py
"""Synthetic fixtures only: never load formal exposed/holdout media."""
import json
from pathlib import Path
import numpy as np
import pytest
import torch
from PIL import Image
from mcss.dynamic.types import hash_scene_state
from mcss.mechanism_pilot.optimization_bounds_evaluation import evaluate_optimization_bounds
from mcss.mechanism_pilot.small_training import sha
from mcss.types import SceneState
def fixture(root, phase="context"):
scenes, chains, specs = [], [], []
bounds = [[-6.0, -4.0, -6.0], [6.0, 4.0, 6.0]]
for i in range(2):
sid = f"scene{i}"
scene = {
"scene_id": sid,
"cohort_role": "CAPACITY_DEV_EXPOSED",
"roles": {"context_a": [0, 1, 2], "context_b": [0, 3, 4], "primary_query": [8, 9]},
"frames": [],
}
for fid in (0, 1, 2, 3, 4, 8, 9):
rgb = root / f"{sid}_{fid}.png"
depth = root / f"{sid}_{fid}.npy"
Image.fromarray(np.full((4, 4, 3), 50 + i * 20, np.uint8)).save(rgb)
np.save(depth, np.full((4, 4), 3.0, np.float32))
pose = torch.eye(4)
pose[0, 3] = i * 2 + fid * 0.01
scene["frames"].append(
{
"frame_id": fid,
"rgb": str(rgb),
"depth": str(depth),
"intrinsics": [[4.0, 0.0, 1.5], [0.0, 4.0, 1.5], [0.0, 0.0, 1.0]],
"c2w": pose.tolist(),
}
)
scenes.append(scene)
roles = ("joint_query",) if phase == "oracle" else ("A", "B")
for role in roles:
chain_key = f"{sid}_{role}"
folder = root / "checkpoints" / chain_key
folder.mkdir(parents=True)
chains.append(
{
"key": chain_key,
"scene_id": sid,
"phase": phase,
"optimization_path": str(folder.relative_to(root)),
}
)
for budget in (1000, 3000):
for selection in ("FIXED_BUDGET", "CONTEXT_SELECTED"):
key = f"{chain_key}_{budget}_{selection}"
dest = folder / key
dest.mkdir()
state = SceneState(
torch.full((1, 1, 16, 16, 16), -2.0 + i + budget / 10000),
torch.full((1, 3, 16, 16, 16), 0.2 + i * 0.1),
torch.full((1, 1, 16, 16, 16), -3.0),
torch.tensor([bounds]),
)
torch.save(state, dest / "state.pt")
(dest / "summary.json").write_text(
json.dumps({"state_hash": hash_scene_state(state)})
)
specs.append(
{
"key": key,
"chain_key": chain_key,
"scene_id": sid,
"phase": phase,
"role": role,
"track": {
"context": "RGBD",
"oracle": "QUERY_ORACLE",
"secondary": "RGB_ONLY",
}[phase],
"grid": 16,
"samples": 64,
"bounds_mode": "CURRENT_BOUNDS",
"bounds": bounds,
"budget": budget,
"selection": selection,
"state_path": str((dest / "state.pt").relative_to(root)),
"optimization_path": str(dest.relative_to(root)),
}
)
(root / "scene_manifest.json").write_text(
json.dumps(
{"scenes": scenes, "image_size": [4, 4], "FINAL_HOLDOUT_PROHIBITED": ["holdout"]}
)
)
(root / "state_plan.json").write_text(json.dumps({"chains": chains, "states": specs}))
sources = {str(Path(__file__).resolve()): sha(Path(__file__))}
(root / "config.json").write_text(json.dumps({"source_sha256": sources}))
for chain in chains:
states = {}
for spec in specs:
if spec["chain_key"] != chain["key"]:
continue
folder = root / spec["optimization_path"]
states[spec["key"]] = {
"state_file_sha256": sha(folder / "state.pt"),
"summary_sha256": sha(folder / "summary.json"),
"state_hash": json.loads((folder / "summary.json").read_text())["state_hash"],
}
(root / chain["optimization_path"] / "completion.json").write_text(
json.dumps(
{
"phase": phase,
"chain_key": chain["key"],
"plan_sha256": sha(root / "state_plan.json"),
"config_sha256": sha(root / "config.json"),
"source_hashes": sources,
"states": states,
}
)
)
(root / f"{phase}_optimization_complete.json").write_text(
json.dumps(
{
"status": "PASS",
"phase": phase,
"states": len(specs),
"input_hashes": {
str(root / n): sha(root / n)
for n in ["scene_manifest.json", "state_plan.json", "config.json"]
},
}
)
)
return specs, chains
def test_phase_complete_required_before_media(tmp_path, monkeypatch):
fixture(tmp_path)
(tmp_path / "context_optimization_complete.json").unlink()
def denied(*args, **kwargs):
raise AssertionError("query reached")
monkeypatch.setattr(
"mcss.mechanism_pilot.optimization_bounds_evaluation.CapacityEvaluator.query", denied
)
with pytest.raises(PermissionError, match="completion marker"):
evaluate_optimization_bounds(tmp_path, device="cpu")
def test_chain_state_or_summary_mutation_blocks_before_media(tmp_path):
specs, _ = fixture(tmp_path)
summary = tmp_path / specs[0]["optimization_path"] / "summary.json"
summary.write_text(summary.read_text() + " ")
with pytest.raises(PermissionError, match="changed after chain completion"):
evaluate_optimization_bounds(tmp_path, device="cpu")
def test_all_budget_states_sealed_and_wrong_donor_matched(tmp_path, monkeypatch):
torch.set_num_threads(1)
specs, _ = fixture(tmp_path)
from mcss.mechanism_pilot.direct_capacity_contracts import StateSealBarrier
count = []
original = StateSealBarrier.seal
def observe(self, *args):
original(self, *args)
count.append(1)
def checked_sha(path):
if Path(path).suffix in (".png", ".npy"):
assert len(count) == len(specs)
return sha(path)
monkeypatch.setattr(StateSealBarrier, "seal", observe)
monkeypatch.setattr("mcss.mechanism_pilot.optimization_bounds_evaluation.sha", checked_sha)
rows = evaluate_optimization_bounds(tmp_path, device="cpu")
assert len(rows) == len(specs) * 2 * 2
expected = {
s["key"]: json.loads((tmp_path / s["optimization_path"] / "summary.json").read_text())[
"state_hash"
]
for s in specs
}
for s in specs:
for fid in (8, 9):
values = [r for r in rows if r["key"] == s["key"] and r["query_id"] == fid]
assert len({r["query_camera_hash"] for r in values}) == 1
direct = next(r for r in values if r["method"] == "direct")
wrong = next(r for r in values if r["method"] == "wrong_scene")
assert direct["used_state_hash"] == expected[s["key"]]
donor = next(
x
for x in specs
if x["scene_id"] == wrong["donor_scene_id"]
and all(
x[k] == s[k] for k in ("budget", "selection", "role", "track", "bounds_mode")
)
)
assert wrong["used_state_hash"] == expected[donor["key"]]
assert (
len(json.loads((tmp_path / "raw/context_context_results.json").read_text()))
== len(specs) * 3
)
def test_oracle_shares_both_queries_and_has_no_wrong_scene(tmp_path):
torch.set_num_threads(1)
specs, _ = fixture(tmp_path, phase="oracle")
rows = evaluate_optimization_bounds(tmp_path, phase="oracle", device="cpu")
assert len(rows) == len(specs) * 2
assert {r["method"] for r in rows} == {"direct"}
for s in specs:
assert len({r["used_state_hash"] for r in rows if r["key"] == s["key"]}) == 1
def test_holdout_guard_rejects_before_source_or_media_access(tmp_path):
fixture(tmp_path)
path = tmp_path / "scene_manifest.json"
manifest = json.loads(path.read_text())
manifest["FINAL_HOLDOUT_PROHIBITED"].append("scene0")
path.write_text(json.dumps(manifest))
with pytest.raises(PermissionError, match="holdout stays closed"):
evaluate_optimization_bounds(tmp_path, device="cpu")
assert not (tmp_path / "raw").exists()
def test_secondary_depth_metrics_are_post_seal_only(tmp_path):
torch.set_num_threads(1)
specs, _ = fixture(tmp_path, phase="secondary")
rows = evaluate_optimization_bounds(tmp_path, phase="secondary", device="cpu")
assert len(rows) == len(specs) * 4
access = json.loads((tmp_path / "raw/secondary_GT_access.json").read_text())
assert all(
event["purpose"]
in (
"HASH_INPUT_AFTER_ALL_STATES_SEALED",
"POST_SEAL_CONTEXT_EVALUATION_NOT_OPTIMIZER",
"SEALED_QUERY_EVALUATION",
)
for event in access
)
assert all(row["track"] == "RGB_ONLY" for row in rows)