import copy import hashlib import json from pathlib import Path import tarfile import tempfile import unittest from unittest.mock import patch import distribution as d import release as r class DistributionTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) self.packages = self.root / 'packages'; self.packages.mkdir() self.evidence = self.root / 'evidence'; self.evidence.mkdir() self.revision = 'a' * 40 self.version = '0.1.0' self.manifests = {} for target in d.TARGETS: path = self.packages / d.name(self.version, target) path.write_bytes(target.encode()) path.with_name(path.name + '.sha256').write_text(f'{d.digest(path)} {path.name}\n') self.manifests[target] = dict(revision=self.revision, version=self.version, source_dirty=False) proof = self.evidence / target; proof.mkdir() report = dict(target=target, revision=self.revision, package=path.name, package_sha256=d.digest(path), tbl_sha256=d.TBL_SHA, automated_passed=True, checks={k:'passed' for k in ('native','library','ide-pty','windows-console','default-template')}) (proof / 'report.json').write_text(json.dumps(report)) (proof / 'manual.json').write_text(json.dumps(dict(target=target, revision=self.revision, package_sha256=d.digest(path), passed=True, tester='Prüfer', terminal_versions={k:'Version 1' for k in ('windows-terminal','terminal.app','xterm','vte')}, cases={k:True for k in ('f1_f12','modifiers','mouse','unicode_colors','resize','input_break','shell')}, evidence=['terminal.txt']))) (proof / 'terminal.txt').write_text('Bedienprotokoll') acceptance = dict(revision=self.revision, passed=True, foreign=True, inventory=True, workspace=True, benchmarks=True, native_ide=True, run_url='https://git.rfc1437.de/run/1', buildhost='Linux arm64') for directory in (self.packages, self.evidence): (directory / 'acceptance.json').write_text(json.dumps(acceptance)) self.mock_verify = patch.object(d, 'verify', side_effect=lambda archive, target, *args: self.manifests[target]) self.mock_verify.start(); self.addCleanup(self.mock_verify.stop) def gate(self): return d.gate(self.packages, self.evidence, self.revision, self.version) def test_complete_gate_and_missing_or_wrong_evidence(self): self.assertEqual(len(self.gate()), 8) target = list(d.TARGETS)[0] path = self.evidence / target / 'report.json' valid = json.loads(path.read_text()) for key, value in [('revision','b'*40), ('target',list(d.TARGETS)[1]), ('package_sha256','0'*64), ('tbl_sha256','0'*64), ('automated_passed',False), ('checks',{'native':'passed'})]: broken = dict(valid, **{key:value}); path.write_text(json.dumps(broken)) with self.assertRaises(ValueError, msg=key): self.gate() path.write_text(json.dumps(valid)) (self.evidence / target / 'manual.json').unlink() with self.assertRaises(FileNotFoundError): self.gate() def test_missing_corrupt_dirty_and_mismatched_packages(self): first = next(iter(d.TARGETS)) self.manifests[first]['source_dirty'] = True with self.assertRaises(ValueError): self.gate() self.manifests[first]['source_dirty'] = False path = self.packages / d.name(self.version, first) path.write_bytes(b'broken') with self.assertRaises(ValueError): self.gate() path.unlink() with self.assertRaises(ValueError): self.gate() def test_archive_traversal_rejected_before_extract(self): path = self.root / 'evil.tar.gz' with tarfile.open(path, 'w:gz') as tar: entry = tarfile.TarInfo('../escape'); tar.addfile(entry) dest = self.root / 'unpack'; dest.mkdir() with self.assertRaises(ValueError): d.unpack(path, dest, list(d.TARGETS)[1]) self.assertFalse((self.root / 'escape').exists()) def test_upload_failure_never_publishes_or_overwrites(self): class API: def __init__(self): self.release = dict(id=1, draft=True, assets=[]) self.data = {}; self.fail_at = None; self.writes = [] def request(self, method, path, data=None): if path.startswith('/tags/'): return {'commit': {'sha': 'a'*40}} if method == 'GET': return copy.deepcopy(self.release) self.writes.append((method,path)) if method == 'POST': name = path.split('name=')[1] if name == self.fail_at: raise OSError('simulierter Uploadfehler') self.data[name] = data self.release['assets'].append(dict(name=name, browser_download_url=name)) return {} if method == 'PATCH': self.release.update(data) def download(self, asset): return self.data[asset['name']] api = API(); api.fail_at = r.expected(self.version)[2] with self.assertRaises(OSError): r.stage(api, self.packages, 'v'+self.version, self.revision) self.assertTrue(api.release['draft']) self.assertFalse(any(method=='PATCH' for method,_ in api.writes)) api.fail_at = None r.stage(api, self.packages, 'v'+self.version, self.revision) r.publish(api, self.packages, self.evidence, 'v'+self.version, self.revision) self.assertFalse(api.release['draft']) before = list(api.writes) with self.assertRaises(ValueError): r.stage(api, self.packages, 'v'+self.version, self.revision) self.assertEqual(before, api.writes) def test_incomplete_remote_upload_blocks_publish(self): class API: def request(self, method, path, data=None): self.assert_get(method) if path.startswith('/tags/'): return {'commit': {'sha': 'a'*40}} return dict(id=1, draft=True, assets=[]) @staticmethod def assert_get(method): if method != 'GET': raise AssertionError('Unerwartete Mutation') with self.assertRaises(ValueError): r.publish(API(), self.packages, self.evidence, 'v'+self.version, self.revision) if __name__ == '__main__': unittest.main()