tegwick
919aefbef5
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a09cbc-910e-77a0-b336-3130194350b3
70 lines
3.6 KiB
Python
70 lines
3.6 KiB
Python
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from intelligence_radar.trials import budget, execute, verify
|
|
|
|
|
|
class TrialTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory();self.addCleanup(self.temp.cleanup)
|
|
self.root=Path(self.temp.name)
|
|
self.config={'campaign':'test','budget_usd':'10'}
|
|
self.model={'id':'test/model','max_usd_per_million':{'prompt':'1','completion':'5'}}
|
|
self.case={'id':'one','instruction':'Return JSON','input':{},'expected':{'answer':1}}
|
|
self.env=patch.dict(os.environ,{'OPENROUTER_API_KEY':'test-not-a-secret'})
|
|
self.env.start();self.addCleanup(self.env.stop)
|
|
|
|
def response(self,*args,**kwargs):
|
|
return io.BytesIO(json.dumps({'id':'generation-test','model':'test/model','provider':'test',
|
|
'usage':{'cost':.001},'choices':[{'message':{'content':'{"answer":1}'}}]}).encode())
|
|
|
|
def test_billing_and_exact_once_success(self):
|
|
result=execute(self.root,self.config,self.case,self.model,opener=self.response)
|
|
self.assertTrue(result['verification']['accepted'])
|
|
self.assertEqual(budget(self.root,'test')['spent_usd'],'0.001')
|
|
same=execute(self.root,self.config,self.case,self.model,opener=lambda *a,**k:self.fail('duplicate call'))
|
|
self.assertEqual(same,result)
|
|
|
|
def test_budget_refuses_call_before_network(self):
|
|
self.config['budget_usd']='0.00001'
|
|
with self.assertRaisesRegex(ValueError,'budget'):
|
|
execute(self.root,self.config,self.case,self.model,opener=lambda *a,**k:self.fail('budget bypass'))
|
|
|
|
def test_timeout_retains_reservation_and_refuses_retry(self):
|
|
def fail(*args,**kwargs):raise TimeoutError('timeout')
|
|
with self.assertRaises(TimeoutError):execute(self.root,self.config,self.case,self.model,opener=fail)
|
|
self.assertGreater(float(budget(self.root,'test')['pending_usd']),0)
|
|
with self.assertRaisesRegex(ValueError,'already reserved'):
|
|
execute(self.root,self.config,self.case,self.model,opener=self.response)
|
|
|
|
def test_verifier_rejects_boolean_as_integer_and_non_json(self):
|
|
self.assertFalse(verify(self.case,'{"answer":true}')['accepted'])
|
|
self.assertFalse(verify(self.case,'```json\n{"answer":1}\n```')['accepted'])
|
|
|
|
def test_uncertain_billing_blocks_other_cases(self):
|
|
def fail(*args, **kwargs):
|
|
raise TimeoutError('timeout')
|
|
with self.assertRaises(TimeoutError):
|
|
execute(self.root, self.config, self.case, self.model, opener=fail)
|
|
other = dict(self.case, id='two')
|
|
with self.assertRaisesRegex(ValueError, 'unresolved reservations'):
|
|
execute(self.root, self.config, other, self.model,
|
|
opener=lambda *a, **k: self.fail('unreconciled campaign made a call'))
|
|
|
|
def test_overrun_halts_campaign_across_restarts(self):
|
|
def expensive(*args, **kwargs):
|
|
value = json.loads(self.response().read())
|
|
value['usage']['cost'] = 1
|
|
return io.BytesIO(json.dumps(value).encode())
|
|
with self.assertRaisesRegex(ValueError, 'exceeded reservation'):
|
|
execute(self.root, self.config, self.case, self.model, opener=expensive)
|
|
self.assertEqual(budget(self.root, 'test')['spent_usd'], '1')
|
|
for case in [self.case, dict(self.case, id='two')]:
|
|
with self.assertRaisesRegex(ValueError, 'campaign halted'):
|
|
execute(self.root, self.config, case, self.model,
|
|
opener=lambda *a, **k: self.fail('halted campaign made a call'))
|