"""Operational routing tests; all data is retained in workspace test-results.""" import sys sys.dont_write_bytecode = True import json import os from pathlib import Path import unittest from unittest.mock import patch import rcms import operations as ops ROOT = rcms.HERE/'test-results'/('operations-'+rcms.stamp()) ROOT.mkdir() class Operations(unittest.TestCase): def setUp(self): self.root=ROOT/self._testMethodName; self.root.mkdir() self.cfg={'run_user':ops.username(), 'operational_layout':True, 'runs_root':str(self.root/'runs'), 'current_result_root':str(self.root/'current'), 'archive_versions_root':str(self.root/'versions'), 'legacy_dirs':{m:str(self.root/m) for m in ops.MODEL_NAMES}, 'regcm_dataset':str(self.root/'regcm/Dataset'), 'utilities':{}} self.cycle='2026082100' for m in ops.MODEL_NAMES: (self.root/m/'wdir').mkdir(parents=True) (self.root/'regcm/Dataset/CFS').mkdir(parents=True) def manifest(self, model='wrf'): root=ops.storage(self.cfg)/self.cycle/'attempt'; (root/model).mkdir(parents=True) man={'config':self.cfg,'cycle':self.cycle,'end':'2026082106','models':[model],'inputs':[]} rcms.put(root/'manifest.json',man) return root,man def products(self,root): post=root/'wrf/post-initial'; (post/'products').mkdir(parents=True) for name in ops.required_products('wrf'): rcms.put(post/'products'/name,'verified fixture') rcms.put(post/'SUCCESS.json',{'products':str(post/'products')}) return post def test_other_model_attempt_does_not_block_new_model(self): self.manifest('wrf') self.assertEqual(ops.plan(self.cfg,self.cycle,['regcm'],{})['regcm']['action'],'NEW') def test_snapshot_has_operational_worker_module(self): self.cfg['slurm']={m:{'nodes':1,'tasks_per_node':1} for m in ops.MODEL_NAMES} with patch.object(rcms,'validate_config'), patch.object(rcms,'inventory',return_value=('2026082106',[])): root,man=ops.new_run(self.cfg,self.cycle,['wrf']) self.assertTrue((root/'runtime/operations.py').is_file()) self.assertEqual(man['snapshot_sha256']['operations.py'],ops.sha256(root/'runtime/operations.py')) self.assertEqual(man['account'],ops.username()) def test_wrong_account_creates_nothing(self): self.cfg['run_user']='different-account' with self.assertRaisesRegex(RuntimeError,'current account'): ops.write_preflight(self.cfg) self.assertFalse(ops.storage(self.cfg).exists()) def test_plan_read_only_and_existing_cycle_blocks(self): self.assertEqual(ops.plan(self.cfg,self.cycle,['wrf'],{})['wrf']['action'],'NEW') self.assertFalse(ops.storage(self.cfg).exists()) (self.root/'wrf/wdir'/self.cycle).mkdir() self.assertEqual(ops.plan(self.cfg,self.cycle,['wrf'],{})['wrf']['action'],'BLOCKED') def test_active_job_prevents_duplicate(self): rows={'123':{'job_id':'123','name':'rcms-wrf-'+self.cycle+'-integrate-initial','state':'RUNNING'}} self.assertEqual(ops.plan(self.cfg,self.cycle,['wrf'],rows)['wrf']['action'],'WAIT') def test_lock_excludes_second_controller(self): with ops.cycle_lock(self.cfg,self.cycle): with self.assertRaisesRegex(RuntimeError,'Another controller'): with ops.cycle_lock(self.cfg,self.cycle): pass def test_worker_uses_legacy_wdir_output(self): root,man=self.manifest() def integrate(man,work,runner): output=work/'output'; output.mkdir(); rcms.put(output/'kept','raw') return {'raw':str(output)} with patch.dict(os.environ,{'SLURM_JOB_ID':'1'}), patch.object(rcms,'preserve_deletions'), patch.object(rcms,'preflight'), patch.object(rcms,'wrf_integrate',side_effect=integrate): rcms.worker(root/'manifest.json','wrf','integrate','initial') actual=self.root/'wrf/wdir'/self.cycle self.assertEqual((root/'wrf/integration').resolve(),actual) self.assertEqual((actual/'output/kept').read_text(),'raw') def test_partial_archive_never_overwritten(self): root,man=self.manifest(); post=self.products(root) dest=ops.archive(self.cfg,'wrf',self.cycle); dest.mkdir(parents=True) rcms.put(dest/'slp.nc','historical') with self.assertRaisesRegex(RuntimeError,'historical result preserved'): ops.publish(root,'wrf',post) self.assertEqual((dest/'slp.nc').read_text(),'historical') self.assertFalse((post/'ARCHIVED.json').exists()) def test_publish_expected_layout_and_idempotence(self): root,man=self.manifest(); post=self.products(root) receipt=ops.publish(root,'wrf',post) dest=ops.archive(self.cfg,'wrf',self.cycle) self.assertTrue(dest.is_symlink()) self.assertEqual(len(receipt['files']),19) self.assertEqual(ops.publish(root,'wrf',post),receipt) self.assertEqual(ops.plan(self.cfg,self.cycle,['wrf'],{})['wrf']['action'],'SKIP') def test_post_failure_does_not_publish(self): root,man=self.manifest(); integration=root/'wrf/integration'; integration.mkdir() (integration/'output').mkdir(); rcms.put(integration/'SUCCESS.json',{'raw':str(integration/'output')}) with patch.dict(os.environ,{'SLURM_JOB_ID':'1'}), patch.object(rcms,'preserve_deletions'), patch.object(rcms,'preflight'), patch.object(rcms,'wrf_post',side_effect=RuntimeError('post failed')), patch.object(ops,'publish') as publish: with self.assertRaisesRegex(RuntimeError,'post failed'): rcms.worker(root/'manifest.json','wrf','post','initial') publish.assert_not_called() def test_archive_failure_keeps_post_success(self): root,man=self.manifest(); integration=root/'wrf/integration'; integration.mkdir() (integration/'output').mkdir(); rcms.put(integration/'SUCCESS.json',{'raw':str(integration/'output')}) def post(man,raw,work,runner): p=work/'products'; p.mkdir(); return p with patch.dict(os.environ,{'SLURM_JOB_ID':'1'}), patch.object(rcms,'preserve_deletions'), patch.object(rcms,'preflight'), patch.object(rcms,'wrf_post',side_effect=post), patch.object(ops,'publish',side_effect=RuntimeError('archive failed')): with self.assertRaisesRegex(RuntimeError,'archive failed'): rcms.worker(root/'manifest.json','wrf','post','initial') self.assertTrue((root/'wrf/post-initial/SUCCESS.json').exists()) self.assertTrue((root/'wrf/post-initial/ARCHIVE_FAILED.json').exists()) self.assertFalse((root/'wrf/post-initial/FAILED.json').exists()) if __name__=='__main__': print('Fixtures:',ROOT) unittest.main(verbosity=2)