test_completion.py

python · 31 lines · Exact bytes

Version & provenance · theory-cm7

Recorded SHA-256 8938cacc8d7cae3ec41a39e497443279c39e065ce068adc09e4e5909ad63159bRendered SHA-256 8938cacc8d7cae3ec41a39e497443279c39e065ce068adc09e4e5909ad63159b

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,copy
2from pathlib import Path
3import torch
4from resume_state import validate_source,restore_optimizer_scheduler,select_loss
5from finalization import sha,digest
6class Ledger:
7 def __init__(self):self.data=type('D',(),{'schedule':range(1713765)})()
8 def verify(self,e):return e['cursor']
9class Plan:
10 def cursor(self,n):return n*128
11class Tests(unittest.TestCase):
12 def test_exact_restore_and_no_warmup_reset(self):
13  torch.manual_seed(42);a=torch.nn.Linear(2,1);opt=torch.optim.AdamW(a.parameters(),lr=.01);sch=torch.optim.lr_scheduler.LambdaLR(opt,lambda n:1/(n+1))
14  for _ in range(4):opt.zero_grad();a(torch.rand(3,2)).sum().backward();opt.step();sch.step()
15  state=copy.deepcopy({'optimizer_state_dict':opt.state_dict(),'scheduler_state_dict':sch.state_dict()});weights=copy.deepcopy(a.state_dict());rng=torch.get_rng_state()
16  def step(m,o,s):o.zero_grad();m(torch.rand(3,2)).sum().backward();o.step();s.step()
17  step(a,opt,sch);b=torch.nn.Linear(2,1);b.load_state_dict(weights);ob=torch.optim.AdamW(b.parameters(),lr=.01);sb=torch.optim.lr_scheduler.LambdaLR(ob,lambda n:1/(n+1));restore_optimizer_scheduler(ob,sb,state);torch.set_rng_state(rng);step(b,ob,sb)
18  self.assertTrue(all(torch.equal(a.state_dict()[k],v) for k,v in b.state_dict().items()));self.assertEqual(sch.state_dict(),sb.state_dict())
19 def test_generic_source_counters_and_prior_best(self):
20  with tempfile.TemporaryDirectory() as d:
21   p=Path(d);identity={'test':'fixture'};(p/'config.json').write_text(json.dumps(identity));(p/'clock.json').write_text('{}')
22   cfg={'original_config':str(p/'config.json'),'original_config_sha256':sha(p/'config.json'),'source_deadline':str(p/'clock.json'),'source_cursor':512000,'source_updates':1000,'source_microstep':4000,'prior_selected':{'path':'earlier-best','sha256':'fixture'}}
23   s={'identity':identity,'identity_sha256':digest(identity),'format':'cm7-theory-recovery-v1','terminal':True,'retention_failures':[],'deadline':{},'exposures':{'cursor':512000},'updates':1000,'microstep':4000,'rank_rng':[{},{}],'rank_group_totals':[torch.tensor([[256000.,0,0]])]*2,'scheduler_state_dict':{'last_epoch':1000},'optimizer_state_dict':{'state':{0:{'step':torch.tensor(20000.),'exp_avg':torch.ones(1),'exp_avg_sq':torch.ones(1)}}},'best_checkpoint':cfg['prior_selected']}
24   self.assertEqual(validate_source(s,cfg,{'new_clock':'explicitly authorized'},Ledger(),Plan()),512000)
25   for k,v in [('terminal',False),('microstep',4001),('updates',999),('best_checkpoint',{}),('retention_failures',['exact'])]:
26    with self.assertRaises(ValueError):validate_source(dict(s,**{k:v}),cfg,{},Ledger(),Plan())
27 def test_loss_only(self):
28  self.assertTrue(select_loss({'nll':.5,'retention':{'finite':True,'exact_preserving':0}},.6))
29  with self.assertRaises(FloatingPointError):select_loss({'nll':float('nan'),'retention':{'finite':True}},.6)
30if __name__=='__main__':unittest.main()
31