229 lines
12 KiB
Python
229 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Tiny synthetic schema checks; opt in to rendering during serial validation.
|
|
|
|
python -m unittest discover -s experiments -p test_plot_frozen_linear_rates.py -v
|
|
MEANFIELD_PLOT_RENDER_TESTS=1 python -m unittest discover -s experiments -p test_plot_frozen_linear_rates.py -v
|
|
"""
|
|
import copy
|
|
import csv
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
|
|
import plot_frozen_linear_rates as plots
|
|
|
|
|
|
def role_data(role):
|
|
baseline = role == "baseline"
|
|
variant = "independent" if baseline else "gravity-schur"
|
|
values = [1., .2, .08, .12] if baseline else [1., .2, .04, .07]
|
|
true_relative = .11 if baseline else .06
|
|
row = {"role": role, "variant": variant, "rhs_norm": "2", "relative_tolerance": ".1", "absolute_target": ".2",
|
|
"true_residual_norm": str(2 * true_relative), "true_relative": str(true_relative),
|
|
"true_gate_pass": "0" if baseline else "1", "solver_converged": "1",
|
|
"iterations": "2", "reported_iterations": "2", "maximum_iterations": "10", "restart": "5",
|
|
"reported_initial_norm": "2", "reported_final_norm": str(2 * values[-1]),
|
|
"construction_seconds": ".75", "normalization_binding_seconds": ".125", "krylov_setup_seconds": ".125",
|
|
"krylov_seconds": "2", "true_check_seconds": ".25", "setup_plus_solve_seconds": "3",
|
|
"setup_solve_and_true_check_seconds": "3.25"}
|
|
history = [{"role": role, "variant": variant, "iteration": str(i), "reported_norm": str(2 * value),
|
|
"final": "1" if index == 3 else "0"} for index, (i, value) in enumerate(zip((0, 1, 2, 2), values))]
|
|
return row, history
|
|
|
|
|
|
def write_fixture(directory, name="n2_late_gravity"):
|
|
campaign = directory / name / "campaign"
|
|
folder = campaign / name
|
|
folder.mkdir(parents=True)
|
|
summary, history = [], []
|
|
comparison = {"comparison_valid": True, "outcome": "candidate_only_pass"}
|
|
audit_history, checks, hashes, outputs = {}, [], {}, {}
|
|
for role in plots.ROLES:
|
|
row, samples = role_data(role)
|
|
summary.append(row)
|
|
history.extend(samples)
|
|
comparison[role] = row
|
|
audit_history[role] = {"samples": samples}
|
|
checks.extend({"check": role + suffix, "passed": True} for suffix in
|
|
("_rhs_norm", "_defect_norm", "_relative_defect", "_honest_gate"))
|
|
for stem, rows in (("preconditioner_summary", summary), ("preconditioner_history", history)):
|
|
path = folder / (stem + ".csv")
|
|
with path.open("w", newline="") as stream:
|
|
writer = csv.DictWriter(stream, fieldnames=list(rows[0]))
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
outputs[path.name] = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
hashes[str(path.resolve())] = outputs[path.name]
|
|
case = {"id": name, "complete": True, "artifact_passed": True, "kind": "benchmark", "outcome": "candidate_only_pass",
|
|
"metadata": {"n": "2", "J": "0", "density_order": "2"},
|
|
"run": {"finished_utc": "2026-09-19T00:00:00Z", "inputs_unchanged": True, "exit_code": 0,
|
|
"output_hashes": outputs, "input_state_hashes": {"/synthetic/frozen.txt": "a" * 64}},
|
|
"benchmark": {"passed": True, "checks": checks, "comparison": comparison},
|
|
"linear_history": audit_history, "benchmark_audit_scope": {"scope": "synthetic schema; not a real FEM audit"}}
|
|
audit = {"complete": True, "artifact_passed": True, "auditor_sha256": "b" * 64, "benchmark_checker_sha256": "c" * 64,
|
|
"consumed_sha256": hashes, "scope": "synthetic schema control", "limitations": ["No real vectors audited"],
|
|
"campaigns": [{"path": str(campaign), "producer_commit": "d" * 40, "mesh_sha256": "e" * 64,
|
|
"complete": True, "artifact_passed": True, "cases": [case]}]}
|
|
path = directory / name / "audit.json"
|
|
path.write_text(json.dumps(audit))
|
|
return path, audit, folder
|
|
|
|
|
|
class LinearPlotChecks(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.directory = Path(self.temporary.name).resolve()
|
|
self.path, self.audit, self.folder = write_fixture(self.directory)
|
|
|
|
def tearDown(self):
|
|
self.temporary.cleanup()
|
|
|
|
def save(self):
|
|
self.path.write_text(json.dumps(self.audit))
|
|
|
|
def test_final_callback_is_not_the_history_minimum(self):
|
|
row, history = role_data("baseline")
|
|
curve = plots.role_curve(row, history)
|
|
self.assertEqual(curve["iteration"], [0, 1, 2, 2])
|
|
self.assertEqual(curve["reported_relative"], [1., .2, .08, .12])
|
|
self.assertEqual(curve["final_reported_index"], 3)
|
|
self.assertEqual(curve["completed_iterations"], 2)
|
|
self.assertNotEqual(curve["reported_relative"][-1], min(curve["reported_relative"]))
|
|
self.assertEqual(curve["endpoint_true_relative"], .11)
|
|
|
|
def test_true_gate_not_recurrence_flag_controls_failure(self):
|
|
cases, sources, hashes = plots.load_cases([self.path])
|
|
baseline, candidate = cases[0]["curves"]
|
|
self.assertTrue(baseline["solver_recurrence_converged"])
|
|
self.assertFalse(baseline["true_gate_pass"])
|
|
self.assertTrue(candidate["true_gate_pass"])
|
|
self.assertEqual(cases[0]["outcome"], "candidate_only_pass")
|
|
self.assertIn("B FAIL", plots.timing_label(baseline))
|
|
self.assertEqual(baseline["setup_seconds"], 1.)
|
|
self.assertEqual(baseline["setup_solve_true_check_seconds"], 3.25)
|
|
self.assertEqual(len(hashes), 2)
|
|
self.assertEqual(sources[0]["sha256"], hashlib.sha256(self.path.read_bytes()).hexdigest())
|
|
json.dumps(plots.sidecar(cases, sources, hashes), allow_nan=False)
|
|
|
|
def test_exact_target_passes_but_contradictory_gate_fails(self):
|
|
row, history = role_data("candidate")
|
|
row.update(true_residual_norm=".2", true_relative=".1", true_gate_pass="1")
|
|
self.assertTrue(plots.role_curve(row, history)["true_gate_pass"])
|
|
row["true_gate_pass"] = "0"
|
|
with self.assertRaisesRegex(ValueError, "True gate"):
|
|
plots.role_curve(row, history)
|
|
|
|
def test_hash_change_after_audit_is_refused(self):
|
|
path = self.folder / "preconditioner_history.csv"
|
|
path.write_text(path.read_text().replace("0.16", "0.17"))
|
|
with self.assertRaisesRegex(ValueError, "Input hash differs"):
|
|
plots.load_cases([self.path])
|
|
|
|
def test_missing_input_hash_or_embedded_endpoint_mismatch_is_refused(self):
|
|
key = next(iter(self.audit["consumed_sha256"]))
|
|
self.audit["consumed_sha256"][key] = "f" * 64
|
|
self.save()
|
|
with self.assertRaisesRegex(ValueError, "input hash"):
|
|
plots.load_cases([self.path])
|
|
self.audit["consumed_sha256"][key] = hashlib.sha256(Path(key).read_bytes()).hexdigest()
|
|
self.audit["campaigns"][0]["cases"][0]["benchmark"]["comparison"]["candidate"]["true_relative"] = ".09"
|
|
self.save()
|
|
with self.assertRaisesRegex(ValueError, "audited endpoint record"):
|
|
plots.load_cases([self.path])
|
|
|
|
def test_full_audit_required_and_pending_or_failed_cases_refused(self):
|
|
for change in ("complete", "artifact_passed"):
|
|
original = copy.deepcopy(self.audit)
|
|
self.audit[change] = False
|
|
self.save()
|
|
with self.assertRaisesRegex(ValueError, "full linear audit"):
|
|
plots.load_cases([self.path])
|
|
self.audit = original
|
|
self.audit["campaigns"][0]["cases"][0]["kind"] = "execution_failure"
|
|
self.save()
|
|
with self.assertRaisesRegex(ValueError, "complete linear benchmark"):
|
|
plots.load_cases([self.path])
|
|
self.audit.pop("benchmark_checker_sha256")
|
|
self.save()
|
|
with self.assertRaisesRegex(ValueError, "not a full"):
|
|
plots.load_cases([self.path])
|
|
|
|
def test_saved_vector_checks_and_outcome_are_required(self):
|
|
case = self.audit["campaigns"][0]["cases"][0]
|
|
removed = case["benchmark"]["checks"].pop()
|
|
self.save()
|
|
with self.assertRaisesRegex(ValueError, "saved-vector"):
|
|
plots.load_cases([self.path])
|
|
case["benchmark"]["checks"].append(removed)
|
|
case["outcome"] = "both_pass"
|
|
self.save()
|
|
with self.assertRaisesRegex(ValueError, "Contradictory true-gate"):
|
|
plots.load_cases([self.path])
|
|
|
|
def test_final_flags_missing_samples_and_mismatched_final_norm_are_refused(self):
|
|
row, history = role_data("candidate")
|
|
for modified, message in ((history[:-1], "final monitor"),
|
|
([history[0], *history[2:]], "Missing reported")):
|
|
with self.subTest(message=message), self.assertRaisesRegex(ValueError, message):
|
|
plots.role_curve(row, modified)
|
|
history[0]["final"] = "1"
|
|
with self.assertRaisesRegex(ValueError, "final monitor"):
|
|
plots.role_curve(row, history)
|
|
history[0]["final"] = "0"
|
|
row["reported_final_norm"] = ".08"
|
|
with self.assertRaisesRegex(ValueError, "final monitor sample"):
|
|
plots.role_curve(row, history)
|
|
|
|
def test_zero_log_values_are_not_floored_and_time_sum_is_checked(self):
|
|
row, history = role_data("candidate")
|
|
history[1]["reported_norm"] = "0"
|
|
with self.assertRaisesRegex(ValueError, "reported log residual"):
|
|
plots.role_curve(row, history)
|
|
row, history = role_data("candidate")
|
|
row["true_residual_norm"] = "0"
|
|
with self.assertRaisesRegex(ValueError, "true endpoint defect"):
|
|
plots.role_curve(row, history)
|
|
row, history = role_data("candidate")
|
|
row["setup_solve_and_true_check_seconds"] = "9"
|
|
with self.assertRaisesRegex(ValueError, "timing sum"):
|
|
plots.role_curve(row, history)
|
|
|
|
def test_multiple_audits_dynamic_panels_and_no_trial_requirement(self):
|
|
followup, _, _ = write_fixture(self.directory, "n3_late_gravity")
|
|
cases, sources, _ = plots.load_cases([self.path, followup])
|
|
self.assertEqual(len(cases), 2)
|
|
self.assertEqual(len(sources), 2)
|
|
self.assertEqual(plots.panel_shape(9), (3, 3))
|
|
self.assertEqual(plots.panel_shape(10), (4, 3))
|
|
self.assertEqual(plots.panel_shape(1), (1, 1))
|
|
with self.assertRaisesRegex(ValueError, "Duplicate case"):
|
|
plots.load_cases([self.path, self.path])
|
|
|
|
@unittest.skipUnless(os.environ.get("MEANFIELD_PLOT_RENDER_TESTS") == "1", "Rendering reserved for serial validation interval")
|
|
def test_render_ten_cases_has_distinct_endpoints_and_hides_unused_panels(self):
|
|
cases, _, _ = plots.load_cases([self.path])
|
|
cases = [dict(copy.deepcopy(cases[0]), id=f"synthetic_{index}") for index in range(10)]
|
|
figure, _ = plots.render(cases)
|
|
self.assertEqual(len(figure.axes), 12)
|
|
self.assertEqual(sum(axis.get_visible() for axis in figure.axes), 10)
|
|
axis = figure.axes[0]
|
|
self.assertEqual(axis.get_yscale(), "log")
|
|
self.assertEqual(list(axis.lines[0].get_xdata()), [0, 1, 2, 2])
|
|
self.assertEqual(list(axis.lines[-1].get_ydata()), [.1, .1])
|
|
self.assertEqual(list(axis.collections[0].get_offsets()[0]), [2., .11])
|
|
self.assertEqual(list(axis.collections[1].get_offsets()[0]), [2., .06])
|
|
self.assertIn("B FAIL", axis.get_xlabel())
|
|
for extension in ("png", "pdf"):
|
|
path = self.directory / ("rates." + extension)
|
|
figure.savefig(path)
|
|
self.assertGreater(path.stat().st_size, 1000)
|
|
from matplotlib import pyplot as plt
|
|
plt.close(figure)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|