#!/usr/bin/env python3 """Tests retain every fixture under test-results; no cleanup is performed.""" import sys sys.dont_write_bytecode = True import contextlib import io import json import os from pathlib import Path import subprocess import unittest from unittest.mock import patch import rcms as app ROOT = app.HERE / 'test-results' / app.stamp() ROOT.mkdir(parents=True) def grib(path, payload=b''): data = b'GRIB' + b'\0\0\0\2' + (20+len(payload)).to_bytes(8,'big') + payload + b'7777' path.write_bytes(data) class Workflow(unittest.TestCase): def setUp(self): self.root = ROOT / self._testMethodName self.root.mkdir() def test_kernel_guard_retains_file_and_directory(self): file=self.root/'keep'; app.put(file,'retained') empty=self.root/'empty'; empty.mkdir() script="import os; os.unlink(%r); os.rmdir(%r)" % (str(file),str(empty)) subprocess.run(app.guarded(sys.executable)+['-c',script],check=True) self.assertEqual(file.read_text(),'retained') self.assertTrue(empty.is_dir()) def test_exclusive_writes(self): p = self.root / 'keep' app.put(p, 'original') with self.assertRaises(FileExistsError): app.put(p, 'replacement') self.assertEqual(p.read_text(), 'original') def test_copy_refuses_destination(self): p,q = self.root/'source', self.root/'dest' app.put(p,'input'); app.put(q,'output') with self.assertRaises(FileExistsError): app.copy_new(p,q) self.assertEqual(q.read_text(),'output') def test_grib_truncation(self): p = self.root/'valid'; grib(p, b'123') app.grib_complete(p) q = self.root/'partial'; q.write_bytes(p.read_bytes()[:-2]) with self.assertRaisesRegex(RuntimeError, 'Truncated'): app.grib_complete(q) z = self.root/'bad_footer'; z.write_bytes(p.read_bytes()[:-4] + b'0000') with self.assertRaisesRegex(RuntimeError, 'end marker'): app.grib_complete(z) def test_inventory_gap_and_end(self): folder=self.root/'20260821'; folder.mkdir() for family in ['flxf','pgbf']: for time in ['2026082100','2026082112']: grib(folder/(family+time+'.01.2026082100.grb2')) with self.assertRaisesRegex(RuntimeError,'Missing flxf at 2026082106'): app.inventory({'input_root':str(self.root)},'2026082100') for family in ['flxf','pgbf']: grib(folder/(family+'2026082106.01.2026082100.grb2')) end,entries=app.inventory({'input_root':str(self.root)},'2026082100','2026082112') self.assertEqual((end,len(entries)),('2026082112',6)) detected, automatic=app.inventory({'input_root':str(self.root)},'2026082100') self.assertEqual((detected,len(automatic)),('2026082112',6)) with self.assertRaisesRegex(RuntimeError,'Missing'): app.inventory({'input_root':str(self.root)},'2026082100','2026082118') grib(folder/'flxf2026082118.01.2026082100.grb2') with self.assertRaisesRegex(RuntimeError,'Missing pgbf at 2026082118'): app.inventory({'input_root':str(self.root)},'2026082100') def test_input_changed(self): p=self.root/'input'; app.put(p,'abcd'); st=p.stat() man={'inputs':[{'path':str(p),'size':st.st_size,'mtime_ns':st.st_mtime_ns}]} app.inputs_unchanged(man) man['inputs'][0]['size'] += 1 with self.assertRaisesRegex(RuntimeError,'Input changed'): app.inputs_unchanged(man) def test_runner_retains_failure_log(self): r=app.Runner(self.root) with self.assertRaisesRegex(RuntimeError,'Command failed'): r.run([sys.executable,'-c','print("failed here"); raise SystemExit(7)']) self.assertIn('failed here',(self.root/'commands/00001.log').read_text()) def test_namelist_relocation_preserves_physics(self): original=app.template('regcm.nml') text=app.render(original,'2026082100','2027030100') text=app.set_nml(text,'dirout',"'/isolated/raw'") self.assertIn("dirout = '/isolated/raw',",text) self.assertIn('mdate2 = 2027030100',text) self.assertEqual(text.split('&physicsparam')[1],original.split('&physicsparam')[1]) with self.assertRaisesRegex(RuntimeError,'occur once'): app.set_nml(text,'missing',"'x'") def manifest(self): (self.root/'wrf').mkdir() m={'cycle':'2026082100','end':'2026082106','inputs':[],'config':{'utilities':{}},'models':['wrf']} app.put(self.root/'manifest.json',m) return self.root/'manifest.json' def test_post_blocked_without_integration_success(self): m=self.manifest() with patch.dict(os.environ,{'SLURM_JOB_ID':'123'}), patch.object(app,'preflight'), patch.object(app,'preserve_deletions'): with self.assertRaisesRegex(RuntimeError,'success record missing'): app.worker(m,'wrf','post','test') self.assertTrue((self.root/'wrf/post-test/FAILED.json').exists()) self.assertFalse((self.root/'wrf/post-test/SUCCESS.json').exists()) def test_failed_integration_retains_raw(self): m=self.manifest() def fail(man,work,r): (work/'raw').mkdir(); app.put(work/'raw/partial.nc','partial') raise RuntimeError('model failure') with patch.dict(os.environ,{'SLURM_JOB_ID':'123'}), patch.object(app,'preflight'), patch.object(app,'preserve_deletions'), patch.object(app,'wrf_integrate',side_effect=fail): with self.assertRaisesRegex(RuntimeError,'model failure'): app.worker(m,'wrf','integrate','test') self.assertEqual((self.root/'wrf/integration/raw/partial.nc').read_text(),'partial') self.assertTrue((self.root/'wrf/integration/FAILED.json').exists()) self.assertFalse((self.root/'wrf/integration/SUCCESS.json').exists()) def test_failed_post_retains_integration_and_prior_attempt(self): m=self.manifest(); integration=self.root/'wrf/integration'; integration.mkdir() (integration/'raw').mkdir(); app.put(integration/'raw/keep','raw-data') app.put(integration/'SUCCESS.json',{'raw':str(integration/'raw')}) def fail(*args): raise RuntimeError('post failure') with patch.dict(os.environ,{'SLURM_JOB_ID':'124'}), patch.object(app,'preflight'), patch.object(app,'preserve_deletions'), patch.object(app,'wrf_post',side_effect=fail): for token in ['a','b']: with self.assertRaisesRegex(RuntimeError,'post failure'): app.worker(m,'wrf','post',token) self.assertEqual((integration/'raw/keep').read_text(),'raw-data') for token in ['a','b']: self.assertTrue((self.root/('wrf/post-'+token)/'FAILED.json').exists()) def test_existing_attempt_not_reused(self): m=self.manifest(); (self.root/'wrf/integration').mkdir() app.put(self.root/'wrf/integration/keep','retained') with patch.dict(os.environ,{'SLURM_JOB_ID':'123'}): with self.assertRaises(FileExistsError): app.worker(m,'wrf','integrate','test') self.assertEqual((self.root/'wrf/integration/keep').read_text(),'retained') def test_slurm_dependency_and_logs(self): cfg=json.loads((app.HERE/'config.json').read_text()) man={'config':cfg,'python':sys.executable} calls=[] def fake(cmd,**kwargs): calls.append(cmd); return subprocess.CompletedProcess(cmd,0,'9876\n','') with patch.object(app.subprocess,'run',side_effect=fake): app.submit_job(self.root,man,'wrf','post','test','1234') self.assertIn('--dependency=afterok:1234',calls[0]) self.assertIn('--kill-on-invalid-dep=yes',calls[0]) self.assertTrue(any('%j.out' in x for x in calls[0])) self.assertEqual(json.loads((self.root/'wrf-post-test.job.json').read_text())['job_id'],'9876') script=self.root/'wrf-post-test.sh' subprocess.run(['bash','-n',str(script)],check=True) self.assertNotIn('run_post-process.sh',script.read_text()) def test_failed_submission_keeps_evidence(self): cfg=json.loads((app.HERE/'config.json').read_text()) with patch.object(app.subprocess,'run',return_value=subprocess.CompletedProcess([],1,'','queue unavailable')): with self.assertRaisesRegex(RuntimeError,'Submission failed'): app.submit_job(self.root,{'config':cfg,'python':sys.executable},'regcm','integrate','test') self.assertIn('queue unavailable',(self.root/'regcm-integrate-test.submission.json').read_text()) if __name__ == '__main__': print('Preserved test fixtures:',ROOT) unittest.main(verbosity=2)