#!/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()