| """Synthetic computational fixtures; not scientific performance measurements.""" |
| import unittest |
| from copy import deepcopy |
| import numpy as np |
| from peppa.compiled import Node,Plan,validate_plan,execute,replay,fit_ensemble,aggregate,release,robust_reward,redesign_schedule |
|
|
| class CompiledTests(unittest.TestCase): |
| def setUp(self): |
| self.registry={'double':{'revision':'r1','run':lambda p,s:{'b':p['a']*2},'validate':lambda r:None},'plus':{'revision':'r2','run':lambda p,s:{'c':p['b']+1},'validate':lambda r:None}} |
| self.plan=Plan((Node('02','plus','r2',('b',),('c',)),Node('01','double','r1',('a',),('b',))),('c',),2) |
| def test_dependency_schedule_replay_and_cache(self): |
| cache={};a=execute(self.plan,{'a':2},self.registry,cache) |
| self.registry['double']['run']=lambda *args: self.fail('cache was bypassed') |
| b=execute(self.plan,{'a':2},self.registry,cache) |
| self.assertEqual(a,b);self.assertEqual(replay(a['events']),{'a':2,'b':4,'c':5}) |
| def test_cycle_missing_and_overwrite(self): |
| for nodes in [(Node('01','double','r1',('c',),('b',)),Node('02','plus','r2',('b',),('c',))), (Node('01','double','r1',('a',),('a',)),)]: |
| with self.assertRaises(ValueError):validate_plan(Plan(nodes,('c',),5),{'a':2},self.registry) |
| def test_budget_and_revision_before_execution(self): |
| with self.assertRaises(ValueError):validate_plan(Plan(self.plan.nodes,('c',),1),{'a':2},self.registry) |
| self.registry['double']['revision']='changed' |
| with self.assertRaises(ValueError):execute(self.plan,{'a':2},self.registry) |
| def test_failure_blocks_dependency_and_charges(self): |
| self.registry['double']['run']=lambda p,s:{'bad':3} |
| a=execute(self.plan,{'a':2},self.registry) |
| self.assertFalse(a['complete']);self.assertEqual(a['spent'],1) |
| self.assertEqual(a['artifacts'],{'a':2}) |
| self.assertTrue(any(e['kind']=='blocked' for e in a['events'])) |
| def test_retry_bound(self): |
| self.registry['double']['run']=lambda p,s:{'b':float('nan')} |
| plan=Plan((Node('01','double','r1',('a',),('b',),attempts=3),),('b',),3) |
| a=execute(plan,{'a':1},self.registry) |
| self.assertEqual(a['spent'],3);self.assertFalse(a['complete']) |
| def test_trace_tampering(self): |
| a=execute(self.plan,{'a':2},self.registry);a['events'][1]['result']['b']=999 |
| with self.assertRaises(ValueError):replay(a['events']) |
| def test_duplicate_invariance(self): |
| a=np.array([[1,2],[2,4],[3,6],[4,8.]]) |
| y=np.array([1,2,3,4.]);f=fit_ensemble(a,y,['a','b']) |
| b=np.column_stack([a,a[:,0]]) |
| g=fit_ensemble(b,y,['a','b','a']) |
| self.assertEqual(f,g);np.testing.assert_equal(aggregate(a,['a','b'],f)[0],aggregate(b,['a','b','a'],g)[0]) |
| def test_conflicting_lineage(self): |
| with self.assertRaises(ValueError):fit_ensemble([[1,2],[3,4]],[1,3],['a','a']) |
| def candidate(self): |
| return {'id':'a','sequence':'ACDE','chemistry_verified':True,'scores':{'p':{'value':2,'lower':1,'upper':3,'unit':'u','support':'calibrated','lineage':'m1','domain_ok':True}}} |
| def obligations(self):return [{'id':'p','unit':'u','support':'calibrated','threshold':.5,'direction':'ge','scale':1}] |
| def test_complete_release_and_shortfall(self): |
| r=release([self.candidate()],self.obligations(),12) |
| self.assertEqual(r['selected'],['a']);self.assertEqual(r['unfilled'],11) |
| def test_missing_unit_domain_proxy_chemistry(self): |
| for fault in ['missing','unit','domain','support','chemistry']: |
| c=self.candidate() |
| if fault=='missing':c['scores']={} |
| if fault=='unit':c['scores']['p']['unit']='wrong' |
| if fault=='domain':c['scores']['p']['domain_ok']=False |
| if fault=='support':c['scores']['p']['support']='proxy' |
| if fault=='chemistry':c['chemistry_verified']=False |
| r=release([c],self.obligations());self.assertEqual(r['selected'],[],fault) |
| def test_robust_context_penalty(self): |
| self.assertLess(robust_reward([[2,-2],[2,2]],[.5,.5]),robust_reward([[2,2],[2,2]],[.5,.5])) |
| def test_redesign_ties(self): |
| self.assertEqual(redesign_schedule({'specificity':3,'binding':3,'cmc':1},5),['binding','specificity','cmc','binding','specificity']) |
|
|
| if __name__=='__main__':unittest.main() |
|
|
