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, plan, reconcile, 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')) def test_plan_needs_no_key_and_writes_nothing(self): config = dict(self.config, models=[self.model]) with patch.dict(os.environ, {}, clear=True): result = plan(self.root, config, {'cases': [self.case]}) self.assertFalse(result['execution_authorized']) self.assertEqual(result['budget_blockers'], []) self.assertEqual(list(self.root.iterdir()), []) executed = execute(self.root, config, self.case, self.model, opener=self.response) self.assertEqual(result['trials'][0]['run_id'], executed['run_id']) after = plan(self.root, config, {'cases': [self.case]}) self.assertEqual(after['new_reservations_usd'], '0') self.assertEqual(after['planned_committed_usd'], '0.001') def test_plan_counts_pending_once_and_exposes_blocker(self): def fail(*args, **kwargs): raise TimeoutError('timeout') with self.assertRaises(TimeoutError): execute(self.root, self.config, self.case, self.model, opener=fail) result = plan(self.root, dict(self.config, models=[self.model]), {'cases': [self.case]}) self.assertEqual(result['budget_blockers'], ['unresolved billing reservations']) self.assertEqual(result['new_reservations_usd'], '0') self.assertEqual(result['planned_committed_usd'], budget(self.root, 'test')['pending_usd']) def test_plan_refuses_duplicates_and_marks_over_budget(self): config = dict(self.config, models=[self.model], budget_usd='0') with self.assertRaisesRegex(ValueError, 'unique'): plan(self.root, config, {'cases': [self.case, self.case]}) result = plan(self.root, config, {'cases': [self.case]}) self.assertIn('planned reservations exceed campaign budget', result['budget_blockers']) def _pending_run_id(self): def fail(*args, **kwargs): raise TimeoutError('timeout') with self.assertRaises(TimeoutError): execute(self.root, self.config, self.case, self.model, opener=fail) return budget(self.root, 'test') def test_reconcile_unknown_reservation_refused(self): with self.assertRaisesRegex(ValueError, 'unknown reservation'): reconcile(self.root, 'test', 'nope', '0.001', 'evidence-sha', 'codex') def test_reconcile_settles_pending_reservation_without_fabricating_trial(self): self._pending_run_id() from intelligence_radar.store import all_records run_id = [e for i, e in all_records(self.root, 'trial-events') if e['kind'] == 'reserved'][0]['run_id'] reconcile(self.root, 'test', run_id, '0.001', 'evidence-sha256:abc', 'codex') state = budget(self.root, 'test') self.assertEqual(state['pending_usd'], '0') self.assertEqual(state['spent_usd'], '0.001') trials_dir = self.root / 'data/trials' self.assertEqual(list(trials_dir.glob('*.json')) if trials_dir.exists() else [], []) def test_reconcile_refuses_duplicate_settlement(self): self._pending_run_id() from intelligence_radar.store import all_records run_id = [e for i, e in all_records(self.root, 'trial-events') if e['kind'] == 'reserved'][0]['run_id'] reconcile(self.root, 'test', run_id, '0.001', 'evidence-sha256:abc', 'codex') with self.assertRaisesRegex(ValueError, 'already settled'): reconcile(self.root, 'test', run_id, '0.001', 'evidence-sha256:abc', 'codex') def test_reconcile_refuses_cross_campaign_reference(self): self._pending_run_id() from intelligence_radar.store import all_records run_id = [e for i, e in all_records(self.root, 'trial-events') if e['kind'] == 'reserved'][0]['run_id'] with self.assertRaisesRegex(ValueError, 'cross-campaign'): reconcile(self.root, 'other-campaign', run_id, '0.001', 'evidence-sha256:abc', 'codex') def test_reconcile_refuses_invalid_amount(self): self._pending_run_id() from intelligence_radar.store import all_records run_id = [e for i, e in all_records(self.root, 'trial-events') if e['kind'] == 'reserved'][0]['run_id'] with self.assertRaises(ValueError): reconcile(self.root, 'test', run_id, '-1', 'evidence-sha256:abc', 'codex') def test_reconcile_refuses_budget_overrun(self): self.config['budget_usd'] = '0.02' self._pending_run_id() from intelligence_radar.store import all_records run_id = [e for i, e in all_records(self.root, 'trial-events') if e['kind'] == 'reserved'][0]['run_id'] with self.assertRaisesRegex(ValueError, 'exceeds campaign budget'): reconcile(self.root, 'test', run_id, '5', 'evidence-sha256:abc', 'codex') def test_reconcile_requires_evidence_and_reviewer(self): self._pending_run_id() from intelligence_radar.store import all_records run_id = [e for i, e in all_records(self.root, 'trial-events') if e['kind'] == 'reserved'][0]['run_id'] with self.assertRaisesRegex(ValueError, 'evidence_ref'): reconcile(self.root, 'test', run_id, '0.001', '', 'codex') with self.assertRaisesRegex(ValueError, 'reviewer'): reconcile(self.root, 'test', run_id, '0.001', 'evidence-sha256:abc', '') def test_reconcile_is_safe_to_retry_after_success(self): self._pending_run_id() from intelligence_radar.store import all_records run_id = [e for i, e in all_records(self.root, 'trial-events') if e['kind'] == 'reserved'][0]['run_id'] first = reconcile(self.root, 'test', run_id, '0.001', 'evidence-sha256:abc', 'codex') with self.assertRaisesRegex(ValueError, 'already settled'): reconcile(self.root, 'test', run_id, '0.001', 'evidence-sha256:abc', 'codex') self.assertEqual(budget(self.root, 'test')['spent_usd'], '0.001')