70 lines
4.3 KiB
Python
70 lines
4.3 KiB
Python
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from scripts import image_inputs as inputs
|
|
|
|
class ImageInputsTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory(); self.addCleanup(self.temp.cleanup)
|
|
self.root = Path(self.temp.name); self.repo = self.root/'repo'; self.repo.mkdir()
|
|
subprocess.run(['git','init','-q',str(self.repo)],check=True)
|
|
(self.repo/'source').write_text('reviewed')
|
|
subprocess.run(['git','-C',str(self.repo),'add','.'],check=True)
|
|
subprocess.run(['git','-C',str(self.repo),'-c','user.name=Fixture','-c','user.email=fixture@example.invalid','commit','-qm','fixture'],check=True)
|
|
self.ref = subprocess.check_output(['git','-C',str(self.repo),'rev-parse','HEAD'],text=True).strip()
|
|
self.data = dict(schema=1, source_ref=self.ref, architecture='amd64', build_number=1, source_date_epoch=1, artifacts={}, container_archives={})
|
|
for name in inputs.REQUIRED | {'filebrowser.tar','fmcd.tar'}:
|
|
path = self.root/name
|
|
if name == 'web-ui.tar':
|
|
with tarfile.open(path,'w') as tar:
|
|
for member_name in ('index.html','packages/archipelago-companion.apk','packages/archipelago-companion.json','archipelago-runtime/scripts/c01_network_owner.py'):
|
|
item=tarfile.TarInfo(member_name); item.size=6; tar.addfile(item,io.BytesIO(b'native'))
|
|
else: path.write_bytes(b'qualified fixture')
|
|
group = 'artifacts' if name in inputs.REQUIRED else 'container_archives'
|
|
self.data[group][name] = dict(path=str(path),sha256=inputs.digest(path),provenance='fixture only')
|
|
self.manifest=self.root/'inputs.json'
|
|
self.save()
|
|
def save(self): self.manifest.write_text(json.dumps(self.data))
|
|
def validate(self): return inputs.validate(self.manifest,self.repo,self.ref,'amd64')
|
|
def test_stage_exact_payload_and_source(self):
|
|
out=self.root/'stage'; inputs.stage(self.manifest,self.repo,self.ref,'amd64',out)
|
|
self.assertEqual((out/'web-ui/index.html').read_text(),'native')
|
|
with tarfile.open(out/'source.tar') as tar:
|
|
self.assertEqual(tar.extractfile('source').read(),b'reviewed')
|
|
with self.assertRaises(FileExistsError): inputs.stage(self.manifest,self.repo,self.ref,'amd64',out)
|
|
def test_modified_artifact_refused(self):
|
|
(self.root/'archipelago').write_bytes(b'stale')
|
|
with self.assertRaisesRegex(ValueError,'hash mismatch'): self.validate()
|
|
def test_dirty_source_refused(self):
|
|
(self.repo/'source').write_text('unreviewed')
|
|
with self.assertRaisesRegex(ValueError,'clean reviewed'): self.validate()
|
|
def test_wrong_source_and_arch_refused(self):
|
|
for key,value in [('source_ref','a'*40),('architecture','arm64')]:
|
|
old=self.data[key]; self.data[key]=value; self.save()
|
|
with self.assertRaisesRegex(ValueError,'differs'): self.validate()
|
|
self.data[key]=old
|
|
def test_missing_offline_baseline_refused(self):
|
|
del self.data['container_archives']['fmcd.tar']; self.save()
|
|
with self.assertRaisesRegex(ValueError,'baseline'): self.validate()
|
|
def test_colliding_archive_refused(self):
|
|
self.data['container_archives']['web-ui.tar']=self.data['artifacts']['web-ui.tar']; self.save()
|
|
with self.assertRaisesRegex(ValueError,'distinct'): self.validate()
|
|
def test_missing_provenance_refused(self):
|
|
self.data['artifacts']['archipelago']['provenance']=''; self.save()
|
|
with self.assertRaisesRegex(ValueError,'provenance'): self.validate()
|
|
def test_archive_traversal_links_and_standalone_refused(self):
|
|
for name,kind in [('../escape',tarfile.REGTYPE),('/escape',tarfile.REGTYPE),('linked',tarfile.SYMTYPE),('aiui/index.html',tarfile.REGTYPE)]:
|
|
path=self.root/'bad.tar'
|
|
with tarfile.open(path,'w') as tar:
|
|
base=tarfile.TarInfo('index.html'); tar.addfile(base)
|
|
member=tarfile.TarInfo(name); member.type=kind; member.linkname='/escape'; tar.addfile(member)
|
|
with self.assertRaises(ValueError): inputs.unpack_web(path,self.root/'out')
|
|
self.assertFalse((self.root/'escape').exists())
|
|
|
|
if __name__ == '__main__': unittest.main()
|