Digital Drosophila Archive
143 lines
7.4 KiB
Python
143 lines
7.4 KiB
Python
"""Synthetic CPU-only budget contract tests, not measured performance evidence."""
|
|
from copy import deepcopy
|
|
import unittest
|
|
|
|
import execution_budget as B
|
|
|
|
|
|
CONTEXT = 'a'*64
|
|
|
|
|
|
class BudgetTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.auth = {}
|
|
self.reserves = {role:dict(wall_seconds=1.,output_bytes=2,basis='explicit fixture allowance')
|
|
for role in B.RESERVE_ROLES}
|
|
self.reserves['cycle_multiplier'] = 1.25
|
|
|
|
def bind(self, name):
|
|
self.auth[name] = 'b'*64
|
|
return dict(receipt_id=name,sha256='b'*64)
|
|
|
|
def cycle(self, name, kind='training', mode='light', steps=1200):
|
|
events = []
|
|
roles = ['brain','original_E','cpu_validator'] if kind == 'training' else ['brain']
|
|
for i,role in enumerate(roles):
|
|
events.append(dict(event_id=name+'-'+role,role=role,receipt_kind='parent-process',
|
|
receipt=self.bind(name+'-'+role),spawn_s=1.+i*4,exit_s=3.+i*4,validated_s=4.+i*4,
|
|
exit_code=0,core_seconds=.5,runtime_seconds=1.,child_wall_seconds=2.))
|
|
return dict(schema=B.SCHEMA,cycle_id=name,kind=kind,diagnostic_mode=mode,steps=steps,
|
|
neurons=B.N,edges=B.E,dt_ms=.1,context_sha256=CONTEXT,clock_id='parent-clock-1',
|
|
started_s=0.,validated_s=len(roles)*4.+1,authority=self.bind(name+'-cycle'),events=events,
|
|
outputs=dict(inventory=self.bind(name+'-inventory'),scope=B.OUTPUT_SCOPE,complete=True,total_bytes=1000),complete=True)
|
|
|
|
def estimate(self, rows, **kwargs):
|
|
return B.estimate_normal_suite(rows,authenticated_receipts=self.auth,target_context_sha256=CONTEXT,
|
|
reserves=self.reserves,**kwargs)
|
|
|
|
def test_full_training_probe_cycles_use_only_end_to_end_and_complete_bytes(self):
|
|
train,probe = self.cycle('train'),self.cycle('probe','probe')
|
|
report = self.estimate([train,probe])
|
|
self.assertEqual(report['status'],'ESTIMATED-NOT-AUTHORIZED')
|
|
self.assertEqual(report['inventory']['brain_calls'],33920)
|
|
self.assertEqual(report['inventory']['biological_seconds'],4070.4)
|
|
self.assertEqual(report['estimated_wall_seconds'],int(1.25*(30000*13+3920*5)+5))
|
|
self.assertEqual(report['estimated_output_bytes'],int(1.25*33920*1000+10))
|
|
self.assertEqual(report['measured_cycles'][0]['timing'],dict(core_seconds=1.5,runtime_seconds=3.,
|
|
child_wall_seconds=6.,end_to_end_seconds=13.))
|
|
for field in ('execution_authorized','scientific_learning_claimed','external_identity_verified_here'):
|
|
self.assertFalse(report[field])
|
|
|
|
def test_missing_either_cycle_returns_symbolic_not_pseudo_precise_total(self):
|
|
for rows in ([],[self.cycle('only-train')],[self.cycle('only-probe','probe')]):
|
|
report = self.estimate(rows)
|
|
self.assertEqual(report['status'],'INSUFFICIENT-MEASUREMENTS')
|
|
self.assertIsNone(report['estimated_wall_seconds'])
|
|
self.assertIsNone(report['estimated_output_bytes'])
|
|
self.assertTrue(report['missing'])
|
|
self.assertIn('30000',report['symbolic_wall'])
|
|
|
|
def test_reference_idle_short_wrong_context_are_reported_but_not_training_samples(self):
|
|
rows = [self.cycle('ref',mode='reference'),self.cycle('idle','state_only_idle',steps=10000),
|
|
self.cycle('short',steps=1000),self.cycle('other')]
|
|
rows[-1]['context_sha256'] = 'c'*64
|
|
report = self.estimate(rows+[self.cycle('probe','probe')])
|
|
self.assertEqual(report['status'],'INSUFFICIENT-MEASUREMENTS')
|
|
self.assertEqual(report['eligible_cycle_ids']['training'],[])
|
|
self.assertEqual(len(report['measured_cycles']),5)
|
|
|
|
def test_actual_parent_event_order_receipts_and_complete_output_required(self):
|
|
mutations = [lambda r:r['events'].pop(),lambda r:r['events'][0].update(receipt_kind='AP-worker-slot'),
|
|
lambda r:r['events'][1].update(spawn_s=0.),lambda r:r['events'][0].update(exit_code=1),
|
|
lambda r:r['events'][0]['receipt'].update(sha256='f'*64),
|
|
lambda r:r['outputs'].update(complete=False),lambda r:r['outputs'].update(scope='state-only'),
|
|
lambda r:r['outputs'].update(total_bytes=True),lambda r:r.update(neurons=16),
|
|
lambda r:r['events'][0].update(child_wall_seconds=1.),lambda r:r.update(complete=1),
|
|
lambda r:r['events'][0].update(runtime_seconds=3.),lambda r:r.update(validated_s=float('nan'))]
|
|
for i,mutate in enumerate(mutations):
|
|
row = self.cycle('mutate'+str(i))
|
|
mutate(row)
|
|
with self.assertRaises(ValueError):
|
|
self.estimate([row])
|
|
with self.assertRaises(ValueError):
|
|
self.estimate([dict(wall_seconds=10,slot='D1')])
|
|
|
|
def test_reserves_have_no_implicit_defaults_and_caps_stop_without_inventory_change(self):
|
|
rows = [self.cycle('train'),self.cycle('probe','probe')]
|
|
report = self.estimate(rows,limits=dict(wall_seconds=1,output_bytes=1))
|
|
self.assertEqual(report['status'],'STOP-BUDGET-EXCEEDED')
|
|
self.assertFalse(report['caps_pass'])
|
|
self.assertEqual(report['inventory'],B.INVENTORY)
|
|
original = deepcopy(self.reserves)
|
|
for role in B.RESERVE_ROLES | {'cycle_multiplier'}:
|
|
self.reserves = deepcopy(original)
|
|
del self.reserves[role]
|
|
with self.assertRaises(ValueError):
|
|
self.estimate(rows)
|
|
self.reserves = deepcopy(original)
|
|
self.reserves['cycle_multiplier'] = .99
|
|
with self.assertRaises(ValueError):
|
|
self.estimate(rows)
|
|
|
|
def test_maxima_not_fitted_or_pooled_with_reference_and_no_mutation(self):
|
|
slow = self.cycle('slow')
|
|
slow['validated_s'] = 25.
|
|
slow['outputs']['total_bytes'] = 2000
|
|
ref = self.cycle('ref',mode='reference')
|
|
ref['validated_s'] = 10000.
|
|
rows = [self.cycle('fast'),slow,ref,self.cycle('probe','probe')]
|
|
saved = deepcopy(rows)
|
|
report = self.estimate(rows)
|
|
self.assertEqual(report['estimated_wall_seconds'],int(1.25*(30000*25+3920*5)+5))
|
|
self.assertEqual(report['estimated_output_bytes'],int(1.25*(30000*2000+3920*1000)+10))
|
|
self.assertEqual(rows,saved)
|
|
|
|
def test_missing_core_runtime_are_unknown_not_required_for_end_to_end(self):
|
|
train,probe = self.cycle('train'),self.cycle('probe','probe')
|
|
for event in train['events']:
|
|
event.update(core_seconds=None,runtime_seconds=None)
|
|
report = self.estimate([train,probe])
|
|
self.assertEqual(report['status'],'ESTIMATED-NOT-AUTHORIZED')
|
|
self.assertIsNone(report['measured_cycles'][0]['timing']['core_seconds'])
|
|
self.assertIsNone(report['measured_cycles'][0]['timing']['runtime_seconds'])
|
|
self.assertEqual(report['measured_cycles'][0]['timing']['end_to_end_seconds'],13.)
|
|
|
|
def test_duplicate_events_and_probe_updates_rejected(self):
|
|
train = self.cycle('train')
|
|
duplicate = deepcopy(train)
|
|
duplicate['cycle_id'] = 'forged-second-cycle'
|
|
with self.assertRaises(ValueError):
|
|
self.estimate([train,duplicate])
|
|
for event in duplicate['events']:
|
|
event['event_id'] += '-renamed'
|
|
with self.assertRaises(ValueError):
|
|
self.estimate([train,duplicate])
|
|
probe = self.cycle('probe','probe')
|
|
probe['events'].append(deepcopy(train['events'][1]))
|
|
with self.assertRaises(ValueError):
|
|
self.estimate([probe])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|