test_deadline.py

python · 29 lines · Exact bytes

Version & provenance · curriculum-cm7

Recorded SHA-256 9130864257cf207a5e86fe1a2881ebdfe40d6e199aff02685faf88285838c3d0Rendered SHA-256 9130864257cf207a5e86fe1a2881ebdfe40d6e199aff02685faf88285838c3d0

Apache-2.0 original project code; dependencies retain their own licenses. License notes. Line links are stable within this snapshot. CM-7 completed implementation source; original and public rendered hashes recorded. Host-specific paths use /work placeholders; inspect reproduction instructions before running. Separate Rust helper retains its GPL license.

1import unittest,tempfile,time,json,sys
2from pathlib import Path
3from unittest.mock import patch
4import deadline as d
5import owner
6class Tests(unittest.TestCase):
7    def test_eight_hour_and_no_reset(self):
8        with tempfile.TemporaryDirectory() as t:
9            p=Path(t)/'clock';v=d.create_or_resume(p,{'run':'same'},budget=28800,margin=180)
10            self.assertEqual(v,d.create_or_resume(p,{'run':'same'},budget=28800,margin=180))
11            with self.assertRaises(ValueError):d.create_or_resume(p,{'run':'other'},budget=28800,margin=180)
12            with self.assertRaises(ValueError):d.create_or_resume(Path(t)/'bad',{},budget=28801,margin=180)
13    def test_cpu_watchdog(self):
14        with tempfile.TemporaryDirectory() as t:
15            v=d.create_or_resume(Path(t)/'deadline',{},budget=6,margin=.2)
16            result=d.supervise([sys.executable,'-c','import signal,time;signal.signal(signal.SIGTERM,signal.SIG_IGN);time.sleep(20)'],v)
17            self.assertTrue(result['deadline_expired']);self.assertNotEqual(result['returncode'],0)
18    def test_resume_missing_deadline_and_alias_refused(self):
19        with tempfile.TemporaryDirectory() as t:
20            root=Path(t);cp=root/'last.pt';cp.write_bytes(b'fixture')
21            cfg={'budget_seconds':28800,'checkpoint_margin_seconds':180,'output':str(root/'out'),'run_id':'fixture','command':['never'],
22                'sealed_dataset_manifest_sha256':'x','schedule_sha256':'y','runtime_sha256':{'z':'x'}}
23            approval={'approved':True,'config_sha256':owner.digest(cfg),'resume':{'path':str(cp),'sha256':'x'}}
24            c=root/'cfg';a=root/'approval';c.write_text(json.dumps(cfg));a.write_text(json.dumps(approval))
25            with self.assertRaisesRegex(ValueError,'original deadline'):owner.run(c,a)
26            cfg['output']=str(root/'out'/'..'/'out');approval['config_sha256']=owner.digest(cfg)
27            with self.assertRaisesRegex(ValueError,'canonical'):owner.validate(cfg,approval)
28if __name__=='__main__':unittest.main(verbosity=2)
29