File size: 4,341 Bytes
98bde72 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | """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()
|