65 lines
2.8 KiB
Python
65 lines
2.8 KiB
Python
|
import hashlib
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
import tempfile
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
from intelligence_radar.profile_review import SOURCE, review
|
||
|
|
from intelligence_radar.store import put
|
||
|
|
|
||
|
|
|
||
|
|
class ProfileReviewTests(unittest.TestCase):
|
||
|
|
def setUp(self):
|
||
|
|
self.tmp = tempfile.TemporaryDirectory()
|
||
|
|
self.addCleanup(self.tmp.cleanup)
|
||
|
|
self.root = Path(self.tmp.name)
|
||
|
|
self.config = {'campaign': 'test', 'models': [
|
||
|
|
{'id': 'model-a', 'max_usd_per_million': {'prompt': '2', 'completion': '10'}},
|
||
|
|
{'id': 'missing', 'max_usd_per_million': {'prompt': '2', 'completion': '10'}}]}
|
||
|
|
self.item = {'id': 'model-a', 'canonical_slug': 'model-a-version',
|
||
|
|
'pricing': {'prompt': '0.000002', 'completion': '0.000011'}}
|
||
|
|
|
||
|
|
def capture(self, entries, url=SOURCE):
|
||
|
|
raw = json.dumps({'data': entries}).encode()
|
||
|
|
sha = hashlib.sha256(raw).hexdigest()
|
||
|
|
path = self.root / 'data/artifacts' / sha
|
||
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
path.write_bytes(raw)
|
||
|
|
return put(self.root, 'captures', {'url': url, 'final_url': url,
|
||
|
|
'content_sha256': sha, 'retrieved_at': '2026-09-14T00:00:00Z'}), path
|
||
|
|
|
||
|
|
def test_prices_and_listing_do_not_establish_access(self):
|
||
|
|
ident, _ = self.capture([self.item])
|
||
|
|
result = review(self.root, ident, self.config)
|
||
|
|
self.assertFalse(result['execution_authorized'])
|
||
|
|
row = result['models'][0]
|
||
|
|
self.assertIsNone(row['requester_access_verified'])
|
||
|
|
self.assertIsNone(row['benchmark_comparability_verified'])
|
||
|
|
self.assertTrue(row['base_prices']['prompt']['within_configured_cap'])
|
||
|
|
self.assertFalse(row['base_prices']['completion']['within_configured_cap'])
|
||
|
|
self.assertFalse(result['models'][1]['listed'])
|
||
|
|
|
||
|
|
def test_invalid_or_missing_prices_remain_unknown(self):
|
||
|
|
for price in ['NaN', '-1', None]:
|
||
|
|
self.item['pricing'] = {'prompt': price}
|
||
|
|
ident, _ = self.capture([self.item])
|
||
|
|
row = review(self.root, ident, self.config)['models'][0]
|
||
|
|
for field in ['prompt', 'completion']:
|
||
|
|
self.assertIsNone(row['base_prices'][field]['within_configured_cap'])
|
||
|
|
|
||
|
|
def test_artifact_tamper_refused(self):
|
||
|
|
ident, path = self.capture([self.item])
|
||
|
|
path.write_text('{}')
|
||
|
|
with self.assertRaisesRegex(ValueError, 'integrity'):
|
||
|
|
review(self.root, ident, self.config)
|
||
|
|
|
||
|
|
def test_duplicate_models_refused(self):
|
||
|
|
ident, _ = self.capture([self.item, self.item])
|
||
|
|
with self.assertRaisesRegex(ValueError, 'duplicate'):
|
||
|
|
review(self.root, ident, self.config)
|
||
|
|
|
||
|
|
def test_wrong_source_refused(self):
|
||
|
|
ident, _ = self.capture([self.item], 'https://example.com/models')
|
||
|
|
with self.assertRaisesRegex(ValueError, 'expected'):
|
||
|
|
review(self.root, ident, self.config)
|