Phase 6: Gitea-Cross-Builds, Paketprüfung und gesperrten Releasepfad implementieren

.gitea/scripts/test_distribution.py

@@ -0,0 +1,122 @@
+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()