#!/usr/bin/env python3 """Cheap summary-schema checks; opt in to rendering only during a serial interval. python -m unittest -v test_plot_coupled_convergence.py MEANFIELD_PLOT_RENDER_TESTS=1 python -m unittest -v test_plot_coupled_convergence.py """ import copy import hashlib import json import os from pathlib import Path import tempfile import unittest import plot_coupled_convergence as plots def case(name): n = 3 if name.startswith("n3") else 2 values = [.5, .25, .25] return { "id": name, "complete": True, "artifact_passed": True, "kind": "newton", "outcome": "not_converged", "failure": "Synthetic failed attempt", "initial_residual": 1., "final_residual": .25, "rtol": 1e-8, "atol": 0., "actual_threshold": 1e-8, "attempted_updates": 3, "accepted_updates": 2, "newton_seconds": 6., "metadata": {"n": str(n), "J": "0", "K": ".1", "rho_c": "1", "density_order": "3", "enthalpy_order": "3", "surface_prescription": "harmonics", "surface_degree": "4", "linear_tol": ".03"}, "run": {"input_state_hashes": {"/synthetic/common_state.txt": "a" * 64} if n == 2 else {}}, "history": [{"iteration": str(i), "accepted": "1" if i < 2 else "0", "seconds": str(t), "F": str(f)} for i, (t, f) in enumerate(zip((2., 3., 1.), values))], } def audit(): return { "complete": True, "artifact_passed": True, "auditor_sha256": "b" * 64, "benchmark_checker_sha256": "c" * 64, "scope": "synthetic schema control", "limitations": ["not real FEM"], "campaigns": [{"path": "/synthetic/campaign", "producer_commit": "d" * 40, "mesh_sha256": "e" * 64, "complete": True, "artifact_passed": True, "cases": [case(name) for pair in plots.PAIRS.values() for name in pair]}], } class SchemaChecks(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory() self.directory = Path(self.temporary.name) self.path = self.directory / "audit.json" self.data = audit() self.write() def tearDown(self): self.temporary.cleanup() def write(self): self.path.write_text(json.dumps(self.data)) def test_t0_cumulative_time_threshold_and_failed_attempt_are_preserved(self): pairs, sources = plots.load_pairs([self.path]) self.assertEqual(set(pairs), {"n3", "n2"}) curve = pairs["n3"][0] self.assertEqual(curve.time, [0., 2., 5., 6.]) self.assertEqual(curve.residual, [1., .5, .25, .25]) self.assertEqual(curve.accepted, [True, True, False]) self.assertEqual(curve.threshold, 1e-8) self.assertEqual(sources[0]["sha256"], hashlib.sha256(self.path.read_bytes()).hexdigest()) report = plots.sidecar("n3", pairs["n3"], sources) self.assertEqual(report["curves"][0]["outcome"], "not_converged") self.assertEqual(report["curves"][0]["actual_stopping_threshold"], 1e-8) json.dumps(report, allow_nan=False) def test_incomplete_audit_and_case_are_refused(self): self.data["complete"] = False self.write() with self.assertRaisesRegex(ValueError, "completed"): plots.load_pairs([self.path]) self.data = audit() self.data["campaigns"][0]["cases"][0]["complete"] = False self.write() with self.assertRaisesRegex(ValueError, "incomplete"): plots.load_pairs([self.path]) def test_missing_and_duplicate_requested_curves_are_refused(self): self.data["campaigns"][0]["cases"].pop() self.write() with self.assertRaisesRegex(ValueError, "Missing requested"): plots.load_pairs([self.path]) self.data = audit() self.data["campaigns"][0]["cases"].append(copy.deepcopy(self.data["campaigns"][0]["cases"][0])) self.write() with self.assertRaisesRegex(ValueError, "duplicate"): plots.load_pairs([self.path]) def test_explicit_single_pair_does_not_require_an_unrequested_pair(self): self.data["campaigns"][0]["cases"] = self.data["campaigns"][0]["cases"][:2] self.write() pairs, _ = plots.load_pairs([self.path], "n3") self.assertEqual(list(pairs), ["n3"]) def test_initial_dependent_actual_threshold_is_checked(self): self.data["campaigns"][0]["cases"][0]["actual_threshold"] = 1e-6 self.write() with self.assertRaisesRegex(ValueError, "stopping threshold"): plots.load_pairs([self.path]) def test_restart_source_mismatch_is_refused(self): self.data["campaigns"][0]["cases"][-1]["run"]["input_state_hashes"] = {"/different_state": "f" * 64} self.write() with self.assertRaisesRegex(ValueError, "hash-bound"): plots.load_pairs([self.path]) def test_local_progress_starts_at_zero_and_retains_failed_endpoint(self): pairs, sources = plots.load_pairs([self.path]) self.assertEqual(plots.residual_reduction(pairs["n2"][0]), [0., 50., 75., 75.]) local = plots.sidecar("n2", pairs["n2"], sources)["local_progress_panel"] self.assertEqual(local["shared_initial_residual"], 1.) self.assertTrue(local["zero_included"]) self.assertEqual(local["curves"][1]["residual_reduction_percent"], [0., 50., 75., 75.]) def test_shared_restart_must_also_have_matching_initial_norm(self): candidate = self.data["campaigns"][0]["cases"][-1] candidate["initial_residual"] *= 2 candidate["final_residual"] *= 2 candidate["actual_threshold"] *= 2 for row in candidate["history"]: row["F"] = str(2 * float(row["F"])) self.write() with self.assertRaisesRegex(ValueError, "shared restart initial norm"): plots.load_pairs([self.path]) def test_nonpositive_log_values_are_not_silently_floored(self): self.data["campaigns"][0]["cases"][0]["history"][0]["F"] = "0" self.write() with self.assertRaisesRegex(ValueError, "nonpositive"): plots.load_pairs([self.path]) def test_missing_history_and_falsely_converged_outcome_are_refused(self): self.data["campaigns"][0]["cases"][0]["history"].pop() self.write() with self.assertRaisesRegex(ValueError, "Missing Newton"): plots.load_pairs([self.path]) self.data = audit() self.data["campaigns"][0]["cases"][0]["outcome"] = "converged" self.write() with self.assertRaisesRegex(ValueError, "exceeds"): plots.load_pairs([self.path]) @unittest.skipUnless(os.environ.get("MEANFIELD_PLOT_RENDER_TESTS") == "1", "Rendering reserved for serial validation interval") def test_render_has_log_axis_t0_and_actual_threshold(self): pairs, _ = plots.load_pairs([self.path]) for name, curves in pairs.items(): figure, _ = plots.render_pair(name, curves) axis = figure.axes[0] self.assertEqual(axis.get_yscale(), "log") self.assertEqual(axis.lines[0].get_xdata()[0], 0.) self.assertEqual(list(axis.lines[-1].get_ydata()), [1e-8, 1e-8]) if name == "n2": local = figure.axes[1] self.assertEqual(local.get_yscale(), "linear") self.assertEqual(local.get_ylim()[0], 0.) self.assertEqual(list(local.lines[0].get_ydata()), [0., 50., 75., 75.]) for extension in ("png", "pdf"): path = self.directory / f"{name}.{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()