96 lines
4.5 KiB
Python
96 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)
|