squaredcuber/loss-aware-dro-repro-artifacts / bundle /v0 /tests /test_appendix_canary_validator.py
squaredcuber's picture
download
raw
3.04 kB
from __future__ import annotations
import importlib.util
import json
import shutil
from pathlib import Path
LANE = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location("appendix_canary_validator", LANE / "scripts" / "validate_appendix_canary.py")
assert SPEC and SPEC.loader
VALIDATOR = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(VALIDATOR)
CANARY = LANE / ".openresearch" / "artifacts" / "appendix-canary-v4"
MANIFEST = LANE / ".openresearch" / "control" / "appendix-routes-manifest-v2.json"
def test_historical_appendix_canary_detects_live_source_drift():
errors = VALIDATOR.validate(CANARY, MANIFEST)
assert "live source/config/input tree differs from receipt" in errors
assert "plan byte hash mismatch" in errors
def test_empty_forged_trace_fails_closed(tmp_path):
forged = tmp_path / "appendix-canary-v4"
shutil.copytree(CANARY, forged)
(forged / "absolute_regression.trace.jsonl").write_text("", encoding="utf-8")
errors = VALIDATOR.validate(forged, MANIFEST)
assert "absolute_regression: trace must be nonempty" in errors
assert "absolute_regression: trace hash mismatch" in errors
assert "absolute_regression: manifest trace binding mismatch" in errors
def test_route_terminal_and_config_tampering_fail_closed(tmp_path):
forged = tmp_path / "appendix-canary-v4"
shutil.copytree(CANARY, forged)
receipt_path = forged / "receipt.json"
receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
receipt["scientific_inputs"]["config_sha256"] = "0" * 64
route = next(row for row in receipt["routes"] if row["route"] == "squared_regression")
route["transport_estimand"]["order"] = 1
route["optimization"]["terminal_solver_status"] = "forged"
receipt_path.write_text(json.dumps(receipt), encoding="utf-8")
errors = VALIDATOR.validate(forged, MANIFEST)
assert "config byte hash mismatch" in errors
assert "squared_regression: transport order mismatch" in errors
assert "squared_regression: terminal solver status unaccepted" in errors
def test_stale_source_binding_fails_closed(monkeypatch):
monkeypatch.setattr(VALIDATOR, "source_tree_hash", lambda _: "sha256:stale")
errors = VALIDATOR.validate(CANARY, MANIFEST)
assert "live source/config/input tree differs from receipt" in errors
def test_trace_continuity_tampering_fails_closed(tmp_path):
forged = tmp_path / "appendix-canary-v4"
shutil.copytree(CANARY, forged)
trace_path = forged / "empirical_w1_portfolio.trace.jsonl"
rows = [json.loads(line) for line in trace_path.read_text(encoding="utf-8").splitlines()]
rows[1]["L"][0][0] += 0.1
trace_path.write_text("\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8")
errors = VALIDATOR.validate(forged, MANIFEST)
assert "empirical_w1_portfolio: metric trace discontinuity at row 1" in errors
assert "empirical_w1_portfolio: trace hash mismatch" in errors

Xet Storage Details

Size:
3.04 kB
·
Xet hash:
2d28202ac62f0622d9bafa5cb58debb620043753bea43b472f2d276a5af9ce6c

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.