test: add bounded first-failure uniform SRT diagnostic

tests/validation/uniform_srt_diagnostic/test_diagnostic.py

@@ -0,0 +1,100 @@
+"""CPU-only synthetic diagnostic and source-contract tests; no simulation results."""
+import json
+from pathlib import Path
+import sys
+from tempfile import TemporaryDirectory
+import unittest
+import numpy as np
+sys.path.insert(0,str(Path(__file__).resolve().parent))
+import diagnose_uniform_srt as d
+cfg,body,p,manifest=d.source_config(d.DEFAULT_SOURCE)
+
+
+def fixture():
+    flags=np.ones((12,20),np.uint16)
+    flags[:,0]=2|0x20; flags[:,-1]=2|0x30
+    flags[[0,-1]]=2|0x10; flags[5,6]=2|0x40|0x200
+    g=np.full((9,12,20),1/9,np.float32)
+    macro=dict(rho=np.ones((12,20)),ux=np.zeros((12,20)),uy=np.zeros((12,20)))
+    params=dict(center=[6,5],radius=1,dx=.1,mean_speed=.01,diameter=80)
+    return g,macro,flags,params
+
+
+def test_exact_config_and_source():
+    assert manifest['commit']=='9805383f9c50d867e8555471baf5dbda8333a652'
+    assert p['diameter']==80 and p['mean_speed']==.01 and p['nx']==1761 and p['ny']==330
+    assert cfg['method']['collision']=='SRT' and cfg['method']['ddf_shifting']
+    assert cfg['method']['outlet']['srt_neq_damp']==.5
+    assert np.isclose(3*p['viscosity']+.5,.524)
+    assert body['objects'][0]['center']==[160,160.5]
+
+
+def test_inactive_nonfinite_does_not_fail():
+    g,m,f,p=fixture(); fluid,_=d.categories(f,p)
+    g[:,~fluid]=np.nan
+    for a in m.values(): a[~fluid]=np.nan
+    r=d.diagnose(g,m,np.zeros(2),f,p,1)
+    assert not r['physical_failure']
+    assert r['categories']['body_solid']['nonfinite_population_nodes']==1
+    assert r['categories']['outlet_ghost']['nodes']==10
+    assert r['categories']['inlet_ghost']['nodes']==10
+    json.dumps(r,allow_nan=False)
+
+
+def test_population_macro_force_separate():
+    g,m,f,p=fixture(); g[4,6,17]=np.inf
+    r=d.diagnose(g,m,np.zeros(2),f,p,27)
+    assert r['physical_failure'] and r['physical_population_nonfinite']
+    assert not r['original_health_failure'] and not r['force_nonfinite']
+    assert r['first_physical_examples'][0]['category']=='outlet_near'
+    assert r['first_physical_examples'][0]['nonfinite_q']==[4]
+    assert r['first_physical_examples'][0]['populations'][4]=='+Inf'
+    json.dumps(r,allow_nan=False)
+    g,m,f,p=fixture(); m['ux'][6,10]=np.nan
+    r=d.diagnose(g,m,np.zeros(2),f,p,28)
+    assert r['physical_macro_nonfinite'] and not r['physical_population_nonfinite']
+    assert not r['force_nonfinite'] and r['original_health_failure']
+    g,m,f,p=fixture(); r=d.diagnose(g,m,np.array([np.nan,0]),f,p,29)
+    assert r['force_nonfinite'] and not r['physical_macro_nonfinite']
+    assert not r['first_physical_examples']
+
+
+def test_original_safety_limits():
+    g,m,f,p=fixture(); m['rho'][5,7]=.49
+    r=d.diagnose(g,m,[0,0],f,p,1)
+    assert r['density_limit'] and r['first_physical_examples'][0]['category']=='body_near'
+    g,m,f,p=fixture(); m['ux'][6,10]=.2
+    assert d.diagnose(g,m,[0,0],f,p,1)['mach_limit']
+
+
+def test_disjoint_complete_categories():
+    _,_,f,p=fixture(); fluid,masks=d.categories(f,p)
+    np.testing.assert_array_equal(sum(mask.astype(int) for mask in masks.values()),np.ones(f.shape))
+    physical=sum(masks[k].astype(int) for k in ('body_near','inlet_near','outlet_near','wall_near','bulk'))
+    np.testing.assert_array_equal(physical,fluid.astype(int))
+
+
+def test_bounded_state_evidence():
+    g,m,f,p=fixture(); state=dict(raw=g.copy(),g=g,macro=m,step=1)
+    with TemporaryDirectory() as tmp:
+        dest=Path(tmp); size=d.save_states(dest,state,state,f,{'status':'synthetic_test'})
+        assert size<d.LIMIT_BYTES
+        with np.load(dest/'states.npz') as saved:
+            np.testing.assert_array_equal(saved['previous_raw_shifted'],g)
+            np.testing.assert_array_equal(saved['current_physical'],g)
+        assert sorted(x.name for x in dest.iterdir())==['diagnostics.json','states.npz']
+    actual_bytes=4*(9*330*1761*4)+(330*1761*2)+16
+    assert actual_bytes<d.LIMIT_BYTES-4*1024**2
+
+
+def test_source_guard_and_deadline():
+    with TemporaryDirectory() as tmp:
+        with unittest.TestCase().assertRaises((ValueError,FileNotFoundError)):
+            d.source_config(Path(tmp))
+    with unittest.TestCase().assertRaises(TimeoutError): d.deadline_check(0)
+
+
+if __name__=='__main__':
+    result=unittest.TextTestRunner(verbosity=2).run(unittest.TestSuite(
+        unittest.FunctionTestCase(fn) for name,fn in sorted(globals().items()) if name.startswith('test_')))
+    raise SystemExit(not result.wasSuccessful())