intelligence-radar/tests/test_advisor.py
tegwick 29ceb77938 Build reproducible capability advisor and scheduled source capture
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a09c77-b1a9-7eb3-b501-6cb24b458c97
2026-09-13 23:07:41 +02:00

95 lines
4.5 KiB
Python

import copy
import json
from pathlib import Path
import tempfile
import unittest
from intelligence_radar.advisor import advise, outcome, recommend, replay
REPO = Path(__file__).resolve().parents[1]
class AdvisorTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.task = json.loads((REPO/'examples/task.json').read_text())
self.catalog = json.loads((REPO/'examples/advisor-catalog.json').read_text())
self.policy = json.loads((REPO/'policies/advisor-v0.2.json').read_text())
def report(self):
return advise(self.root, self.task, self.catalog, self.policy, '2026-09-13', True)
def test_rank_cost_and_no_unearned_sufficiency(self):
r = self.report()
self.assertEqual(r['frontier'], 'frontier-fixture')
self.assertEqual(r['efficient'], 'efficient-fixture')
self.assertIsNone(r['budget'])
self.assertEqual(len(r['selected_benchmarks']), 5)
self.assertAlmostEqual(sum(b['weight'] for b in r['selected_benchmarks']), 1)
efficient = next(x for x in r['profiles'] if x['id']=='efficient-fixture')
self.assertAlmostEqual(efficient['cost']['scenario_usd'], .07)
def test_missing_score_never_imputed(self):
self.catalog['results'].pop(0)
report = self.report()
self.assertEqual(report['frontier'], 'efficient-fixture')
row = next(x for x in report['profiles'] if x['id']=='frontier-fixture')
self.assertIsNone(row['fitness_proxy'])
self.assertEqual(len(row['missing_measurements']), 1)
def test_constraints_remove_highest_scoring_profile(self):
self.catalog['profiles'][0]['attributes']['requester_access'] = False
self.assertEqual(self.report()['frontier'], 'efficient-fixture')
def test_missing_or_conflicting_price_prevents_cheapest_claim(self):
self.catalog['prices'].pop(0)
self.assertIsNone(self.report()['efficient'])
self.catalog['prices'].append(copy.deepcopy(self.catalog['prices'][0]))
self.assertIsNone(self.report()['efficient'])
def test_stale_future_and_wrong_version_results(self):
for field, value in [('observed_at','2026-09-14T00:00:00Z'),
('observed_at','2025-01-01T00:00:00Z'),
('benchmark_version','other')]:
with self.subTest(field=field,value=value):
original=self.catalog['results'][0].copy()
self.catalog['results'][0][field]=value
self.assertEqual(self.report()['frontier'], 'efficient-fixture')
self.catalog['results'][0]=original
def test_aggregate_and_short_selection_withhold_ranking(self):
self.catalog['benchmarks'][0]['aggregate']=True
self.assertIsNone(self.report()['frontier'])
def test_calibration_duplicate_protection_and_frozen_replay(self):
ident, initial = recommend(self.root,self.task,self.catalog,self.policy,'2026-09-13',True)
value=dict(recommendation_id=ident,profile_id='budget-fixture',run_id='run-0',
accepted=True,independent_verification=True,verifier_evidence='synthetic test run',
actual_usd=.01,runtime_seconds=1,completed_at='2026-09-13T22:00:00Z')
first=outcome(self.root,value)
self.assertEqual(outcome(self.root,value),first)
with self.assertRaises(ValueError):
outcome(self.root,dict(value,accepted=False))
for i in range(1,20):
outcome(self.root,dict(value,run_id=f'run-{i}'))
self.assertEqual(self.report()['budget'],'budget-fixture')
self.assertEqual(replay(self.root,ident), initial)
self.task['acceptance']+=' Different acceptance test.'
self.assertIsNone(self.report()['budget'])
def test_synthetic_requires_explicit_opt_in(self):
with self.assertRaises(ValueError):
advise(self.root,self.task,self.catalog,self.policy,'2026-09-13')
def test_invalid_numeric_inputs_rejected(self):
self.catalog['results'][0]['raw_score']=float('nan')
with self.assertRaises(ValueError):
self.report()
def test_replay_is_independent_of_json_object_key_order(self):
self.task['constraints']['unavailable_attribute'] = True
self.task['capabilities']['coding'] = .31
ident, initial = recommend(self.root,self.task,self.catalog,self.policy,'2026-09-13',True)
self.assertEqual(replay(self.root,ident), initial)