tegwick
919aefbef5
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a09cbc-910e-77a0-b336-3130194350b3
82 lines
3.8 KiB
Python
82 lines
3.8 KiB
Python
import io
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
|
|
from intelligence_radar.advisor import recommend, replay
|
|
from intelligence_radar.bundles import export_bundle, import_bundle
|
|
from intelligence_radar.ingest import capture
|
|
from intelligence_radar.store import canonical, get, put
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class BundleTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name)
|
|
self.source = self.root/'source'
|
|
task = json.loads((REPO/'examples/task.json').read_text())
|
|
catalog = json.loads((REPO/'examples/advisor-catalog.json').read_text())
|
|
policy = json.loads((REPO/'policies/advisor-v0.2.json').read_text())
|
|
self.ident,self.report = recommend(self.source,task,catalog,policy,'2026-09-13',True)
|
|
self.bundle=self.root/'report.tar'
|
|
|
|
def test_portable_replay_and_idempotent_import(self):
|
|
export_bundle(self.source,self.ident,self.bundle)
|
|
target=self.root/'imported'
|
|
import_bundle(target,self.bundle)
|
|
self.assertEqual(replay(target,self.ident),self.report)
|
|
import_bundle(target,self.bundle)
|
|
export_bundle(self.source,self.ident,self.root/'second.tar')
|
|
self.assertEqual(self.bundle.read_bytes(),(self.root/'second.tar').read_bytes())
|
|
|
|
def test_path_traversal_is_rejected_before_writes(self):
|
|
with tarfile.open(self.bundle,'w') as archive:
|
|
member=tarfile.TarInfo('../escape');member.size=1
|
|
archive.addfile(member,io.BytesIO(b'x'))
|
|
with self.assertRaisesRegex(ValueError,'path'):
|
|
import_bundle(self.root/'target',self.bundle)
|
|
self.assertFalse((self.root/'escape').exists())
|
|
self.assertFalse((self.root/'target').exists())
|
|
|
|
def test_tampered_member_is_rejected_before_writes(self):
|
|
export_bundle(self.source,self.ident,self.bundle)
|
|
changed=self.root/'tampered.tar'
|
|
with tarfile.open(self.bundle) as original, tarfile.open(changed,'w') as archive:
|
|
for member in original:
|
|
blob=original.extractfile(member).read()
|
|
if member.name!='manifest.json':blob+=b' '
|
|
member.size=len(blob)
|
|
archive.addfile(member,io.BytesIO(blob))
|
|
with self.assertRaisesRegex(ValueError,'checksum'):
|
|
import_bundle(self.root/'target',changed)
|
|
self.assertFalse((self.root/'target').exists())
|
|
|
|
def test_invalid_size_override_rejected(self):
|
|
source={'id':'x','publisher':'x','url':'https://example.invalid','max_bytes':True}
|
|
with self.assertRaises(ValueError):
|
|
capture(self.source,source)
|
|
|
|
def test_missing_dependency_is_rejected_even_when_replay_ignores_it(self):
|
|
snapshot = get(self.source, 'recommendations', self.ident)
|
|
snapshot['provenance'] = {'capture_id': 'a' * 64}
|
|
ident = put(self.source, 'recommendations', snapshot)
|
|
# Synthetic replay does not consult this provenance, but a portable
|
|
# bundle must still carry every dependency it claims to retain.
|
|
name = f'data/recommendations/{ident}.json'
|
|
blob = (self.source / name).read_bytes()
|
|
manifest = {'schema_version': 1, 'recommendation_id': ident,
|
|
'files': {name: hashlib.sha256(blob).hexdigest()}}
|
|
with tarfile.open(self.bundle, 'w') as archive:
|
|
for path, data in [(name, blob), ('manifest.json', canonical(manifest))]:
|
|
member = tarfile.TarInfo(path)
|
|
member.size = len(data)
|
|
archive.addfile(member, io.BytesIO(data))
|
|
with self.assertRaises(FileNotFoundError):
|
|
import_bundle(self.root / 'target', self.bundle)
|
|
self.assertFalse((self.root / 'target').exists())
|