diff --git a/tests/validation/multigrid/spacetime_reference.py b/tests/validation/multigrid/spacetime_reference.py index 879aba6..2d3801e 100644 --- a/tests/validation/multigrid/spacetime_reference.py +++ b/tests/validation/multigrid/spacetime_reference.py @@ -229,6 +229,15 @@ def advance(spec, plan, state, mode="spacetime", mutation=None): negative_boundary_count=negative_boundary, nonfinite_boundary_count=nonfinite_boundary) +def write_receipt(path, result): + """Normalize NumPy scalars at the JSON boundary; reject other unknown types.""" + def scalar(value): + if isinstance(value, np.generic): + return value.item() + raise TypeError(f"Object of type {type(value).__name__} is not JSON serializable") + path.write_text(json.dumps(result, indent=2, allow_nan=False, default=scalar)+"\n") + + def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--n64", action="store_true", help="explicitly run the frozen 1024-interval experiment") @@ -267,7 +276,7 @@ def main(): limits=dict(seconds=args.max_seconds,rss_mib=args.max_rss_mib),runs=[],completed=False) def save(): result["resources"] = dict(seconds=time.monotonic()-started,peak_rss_mib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss/1024) - args.output.write_text(json.dumps(result,indent=2,allow_nan=False)+"\n") + write_receipt(args.output, result) save() try: for mode in ("constant_ivp", "spacetime"): diff --git a/tests/validation/multigrid/test_spacetime_reference.py b/tests/validation/multigrid/test_spacetime_reference.py index dd3387b..c18d181 100644 --- a/tests/validation/multigrid/test_spacetime_reference.py +++ b/tests/validation/multigrid/test_spacetime_reference.py @@ -1,5 +1,8 @@ """Tiny CPU algebra/event tests; no full trajectory or GPU execution.""" from dataclasses import replace +import json +from pathlib import Path +import tempfile import unittest import numpy as np import spacetime_reference as st @@ -117,5 +120,36 @@ class SpaceTime(unittest.TestCase): self.assertEqual(d['nonfinite_boundary_count'],0) +class Receipts(unittest.TestCase): + def test_completed_and_failed_screen_receipts(self): + for passed in (True, False): + with self.subTest(passed=passed), tempfile.TemporaryDirectory() as directory: + # Same NumPy comparison type produced by the final metric screen. + checks = {'velocity_l2': np.float64(.5 if passed else 1.) <= .8, + 'mass_momentum': True} + self.assertIsInstance(checks['velocity_l2'], np.bool_) + result = dict(completed=True, checks=checks, + experimental_pass=all(checks.values()), + runs=[dict(rows=[dict(step=np.int64(1024), + mass_relative=np.float32(0.))])]) + path = Path(directory)/'receipt.json' + st.write_receipt(path, result) + saved = json.loads(path.read_text()) + self.assertIs(saved['completed'], True) + self.assertIs(saved['experimental_pass'], passed) + self.assertIs(saved['checks']['velocity_l2'], passed) + self.assertEqual(saved['runs'][0]['rows'][0], + dict(step=1024, mass_relative=0.)) + + def test_receipts_still_reject_nonfinite_and_unknown_values(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory)/'receipt.json' + with self.assertRaises(ValueError): + st.write_receipt(path, {'invalid': np.float32('nan')}) + with self.assertRaises(TypeError): + st.write_receipt(path, {'invalid': object()}) + self.assertFalse(path.exists()) + + if __name__=='__main__': unittest.main(verbosity=2)