98 lines
5.0 KiB
Python
98 lines
5.0 KiB
Python
"""CPU-only full admission and unchanged observer tests; fixture receipts are synthetic."""
|
|
from copy import deepcopy
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
from tempfile import TemporaryDirectory
|
|
import unittest
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import run_dfg_full as f
|
|
import test_dfg as pilot_tests
|
|
|
|
|
|
def fixture(path):
|
|
cfg, body, parameters, manifest, _ = f.configuration(f.pilot.ROOT)
|
|
import run_refined_dfg as runner
|
|
_, periodic = runner.helpers(f.pilot.ROOT)
|
|
expected, result = pilot_tests.receipt()
|
|
checkout = dict(checkout_revision='c0be841b2e6e1ccba3b8343cecd3cee2fe0ed14f', checkout_dirty=False)
|
|
result.update(checkout=checkout, startup_seconds=3.3353, step_seconds=3.6794, wall_seconds=8.)
|
|
preflight = dict(**expected, checkout=checkout, config=cfg, body=body, parameters=parameters,
|
|
source=manifest, max_steps=40000, cap_tstar=5., observation_width_tstar=.5,
|
|
gates=periodic.GATES, reference_gates=periodic.REFERENCE_GATES)
|
|
ready = dict(**expected, checkout=checkout, case='uniform', mode='pilot', step=4000,
|
|
tstar=.5, runner_sha256=runner.runner_hash(),
|
|
all_runtime_and_observation_paths_exercised=True,
|
|
calibration_batches_seconds=result['calibration_batches_seconds'])
|
|
(path/'uniform').mkdir(exist_ok=True)
|
|
for name, data in [('result.json', result), ('preflight.json', preflight), ('uniform/ready.json', ready)]:
|
|
(path/name).write_text(json.dumps(data))
|
|
return cfg, body, parameters, manifest
|
|
|
|
|
|
def test_admission_rejects_mismatches():
|
|
with TemporaryDirectory() as temp:
|
|
path = Path(temp); args = fixture(path)
|
|
receipt = f.admit_pilot(path, *args)
|
|
assert abs(receipt['step_force_scaled_seconds'] - 220.764) < 1e-10
|
|
cases = [('result.json', 'collision', 'MRT'),
|
|
('result.json', 'step_seconds', float('nan')),
|
|
('result.json', 'healthy', False),
|
|
('uniform/ready.json', 'step', 160),
|
|
('uniform/ready.json', 'runner_sha256', 'wrong'),
|
|
('preflight.json', 'body', {'objects': []}),
|
|
('preflight.json', 'gates', {}),
|
|
('preflight.json', 'source', {})]
|
|
for filename, key, value in cases:
|
|
fixture(path)
|
|
target = path/filename; data=json.loads(target.read_text()); data[key]=value
|
|
target.write_text(json.dumps(data))
|
|
with unittest.TestCase().assertRaises(ValueError): f.admit_pilot(path, *args)
|
|
fixture(path); (path/'uniform/ready.json').unlink()
|
|
with unittest.TestCase().assertRaises(FileNotFoundError): f.admit_pilot(path, *args)
|
|
|
|
|
|
def test_full_requires_health_periodicity_and_reference():
|
|
good = dict(case='uniform', mode='full', steps=2400000, executed_steps=2400000,
|
|
healthy=True, periodic=True, reference_comparison=dict(reference_agreement=True))
|
|
assert f.qualified(good)
|
|
for key, value in [('healthy', False), ('periodic', False), ('steps', 40000),
|
|
('executed_steps', 2399900), ('mode', 'pilot'), ('case', 'refined'),
|
|
('reference_comparison', {}),
|
|
('reference_comparison', dict(reference_agreement=False))]:
|
|
bad=deepcopy(good); bad[key]=value; assert not f.qualified(bad)
|
|
|
|
|
|
def test_preflight_preserves_physics_and_gates():
|
|
with TemporaryDirectory() as temp:
|
|
root=Path(temp); incoming=root/'pilot'; incoming.mkdir(); cfg, body, params, _=fixture(incoming)
|
|
output=root/'full'
|
|
subprocess.run([sys.executable, str(Path(f.__file__)), '--pilot-results', str(incoming),
|
|
'--output', str(output)], check=True)
|
|
data=json.loads((output/'preflight.json').read_text())
|
|
assert data['config']==cfg and data['body']==body and data['parameters']==params
|
|
assert data['max_steps']==2400000 and data['cap_tstar']==300 and data['wall_seconds']==720
|
|
assert not data['accuracy_qualified'] and not data['executed']
|
|
assert data['numerical_source_commit']=='926ca27407c26ca11283282a6e53249925725376'
|
|
assert 'restart' in data['initialization']
|
|
|
|
|
|
def test_old_files_and_late_observer_unchanged():
|
|
root=f.pilot.ROOT
|
|
for name in ['tests/validation/run_refined_dfg.py',
|
|
'tests/validation/experimental_full9/run_dfg.py',
|
|
'tests/validation/experimental_full9/dfg_source_manifest.json',
|
|
'tests/validation/periodic.py']:
|
|
old=subprocess.check_output(['git','-C',str(root),'show','c0be841:'+name])
|
|
assert old==(root/name).read_bytes()
|
|
source=(root/'tests/validation/run_refined_dfg.py').read_text()
|
|
assert 'begin * SPEED / p["diameter"] >= 240' in source
|
|
assert 'cap = 5.0 if a.mode == "pilot" else 300.0' in source
|
|
|
|
|
|
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())
|