148 lines
6.4 KiB
Python
148 lines
6.4 KiB
Python
#!/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_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])
|
|
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()
|