test_owner.py

python · 22 lines · Exact bytes

Version & provenance · theory-cm7

Recorded SHA-256 0ed29b725d822cddd6acb36d03d677adf5f6a71f4d7a91788fe4b4ce23b5ee84Rendered SHA-256 0ed29b725d822cddd6acb36d03d677adf5f6a71f4d7a91788fe4b4ce23b5ee84

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,json
2from pathlib import Path
3from unittest.mock import patch
4import owner
5class Tests(unittest.TestCase):
6 def test_single_admission_preserves_clock(self):
7  with tempfile.TemporaryDirectory() as td:
8   shared=Path(td);root=shared/'completion-v1';cfg={'output':str(root),'original_root':str(shared),'run_id':'cpu-fixture','command':['/bin/true'],'source_cursor':512000,'source_deadline':'prior-clock-preserved'};cp=shared/'config.json';ap=shared/'approval.json';cp.write_text(json.dumps(cfg));ap.write_text('{}')
9   with patch.object(owner,'validate',return_value=(shared,root)),patch.object(owner,'supervise',return_value={'returncode':0,'deadline_expired':False}),patch.object(owner,'finalize',return_value={'status':'certified'}):
10    self.assertEqual(owner.run(cp,ap)['returncode'],0);clock=(root/'deadline.json').read_bytes()
11    with self.assertRaises(FileExistsError):owner.run(cp,ap)
12    self.assertEqual((root/'deadline.json').read_bytes(),clock)
13   self.assertEqual(json.loads(clock)['budget_seconds'],2400)
14 def test_reject_wrong_budget_and_approval(self):
15  for cfg in [{'draft_only':True},{'budget_seconds':3600,'checkpoint_margin_seconds':120},{'budget_seconds':2400,'checkpoint_margin_seconds':120,'source_checkpoint_sha256':'fixture'}]:
16   with self.assertRaises(ValueError):owner.validate(cfg,{})
17 def test_failure_not_certified(self):
18  with tempfile.TemporaryDirectory() as td:
19   shared=Path(td);root=shared/'completion-v1';cfg={'output':str(root),'original_root':str(shared),'run_id':'cpu-failure','command':['/bin/true'],'source_cursor':512000,'source_deadline':'prior'};cp=shared/'cfg';ap=shared/'approval';cp.write_text(json.dumps(cfg));ap.write_text('{}')
20   with patch.object(owner,'validate',return_value=(shared,root)),patch.object(owner,'supervise',return_value={'returncode':0,'deadline_expired':False}),patch.object(owner,'finalize',return_value={'status':'failed'}):self.assertEqual(owner.run(cp,ap)['returncode'],125)
21if __name__=='__main__':unittest.main()
22