122 lines
5.2 KiB
Python
122 lines
5.2 KiB
Python
import copy
|
|
from pathlib import Path
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from urllib.error import HTTPError
|
|
import distribution as d
|
|
import release as r
|
|
|
|
|
|
class API:
|
|
"""Release-API ohne Downloadpfad; protokolliert jede Mutation."""
|
|
def __init__(self):
|
|
self.release = None
|
|
self.fail_at = None
|
|
self.writes = []
|
|
self.omit_upload = False
|
|
self.wrong_size = False
|
|
|
|
def request(self, method, path, data=None):
|
|
if path.startswith('/tags/'):
|
|
return {'commit': {'sha': 'a' * 40}}
|
|
if method == 'GET':
|
|
if self.release is None:
|
|
raise HTTPError(path, 404, 'Not Found', {}, None)
|
|
return copy.deepcopy(self.release)
|
|
self.writes.append((method, path))
|
|
if method == 'POST' and path == '/releases':
|
|
self.release = dict(id=1, assets=[], **data)
|
|
elif method == 'POST':
|
|
name = path.split('name=')[1]
|
|
if name == self.fail_at:
|
|
raise OSError('simulierter Uploadfehler')
|
|
if not self.omit_upload:
|
|
self.release['assets'].append(dict(name=name, size=len(data) + self.wrong_size))
|
|
elif method == 'PATCH':
|
|
self.release.update(data)
|
|
return copy.deepcopy(self.release)
|
|
|
|
|
|
class DistributionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.packages = Path(self.temp.name)
|
|
self.revision = 'a' * 40
|
|
self.version = '1.0.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)
|
|
mock_verify = patch.object(d, 'verify', side_effect=lambda archive, target, *args: self.manifests[target])
|
|
mock_verify.start(); self.addCleanup(mock_verify.stop)
|
|
|
|
def publish(self, api):
|
|
return r.publish(api, self.packages, 'v' + self.version, self.revision)
|
|
|
|
def test_release_tags_are_exact_numeric_versions(self):
|
|
for tag in ('v1.0.0', 'v0.0.0', 'v12.34.56'):
|
|
self.assertEqual(r.release_version(tag), tag[1:])
|
|
for tag in ('main', '1.0.0', 'v1.0', 'v1.0.0-rc1', 'v1.0.0+build',
|
|
'v1.0.0/extra', 'v01.0.0', 'v1x.0.0', 'v1.0.0\n'):
|
|
with self.assertRaises(ValueError, msg=tag): r.release_version(tag)
|
|
|
|
def test_bad_packages_prevent_any_release_mutation(self):
|
|
api = API()
|
|
first = next(iter(d.TARGETS))
|
|
for key, value in [('source_dirty', True), ('revision', 'b' * 40)]:
|
|
original = self.manifests[first][key]
|
|
self.manifests[first][key] = value
|
|
with self.assertRaises(ValueError): self.publish(api)
|
|
self.manifests[first][key] = original
|
|
path = self.packages / d.name(self.version, first)
|
|
path.write_bytes(b'broken')
|
|
with self.assertRaises(ValueError): self.publish(api)
|
|
path.unlink()
|
|
with self.assertRaises(FileNotFoundError): self.publish(api)
|
|
self.assertEqual(api.writes, [])
|
|
|
|
def test_archive_traversal_rejected_before_extract(self):
|
|
path = self.packages / 'evil.tar.gz'
|
|
with tarfile.open(path, 'w:gz') as tar:
|
|
tar.addfile(tarfile.TarInfo('../escape'))
|
|
dest = self.packages / 'unpack'; dest.mkdir()
|
|
with self.assertRaises(ValueError): d.unpack(path, dest, list(d.TARGETS)[1])
|
|
self.assertFalse((self.packages / 'escape').exists())
|
|
|
|
def test_upload_failure_then_retry_publishes_directly_without_evidence_or_download(self):
|
|
api = API(); api.fail_at = r.expected(self.version)[2]
|
|
with self.assertRaises(OSError): self.publish(api)
|
|
self.assertTrue(api.release['draft'])
|
|
self.assertFalse(any(method == 'PATCH' for method, _ in api.writes))
|
|
self.assertEqual(len(api.release['assets']), 2)
|
|
before_retry = len(api.writes)
|
|
api.fail_at = None
|
|
self.publish(api)
|
|
self.assertFalse(api.release['draft'])
|
|
self.assertEqual(len(api.release['assets']), 8)
|
|
self.assertEqual(len(api.writes[before_retry:]), 7) # sechs fehlende Assets + Freigabe
|
|
self.assertEqual(api.writes[-1], ('PATCH', '/releases/1'))
|
|
before = list(api.writes)
|
|
with self.assertRaises(ValueError): self.publish(api)
|
|
self.assertEqual(before, api.writes)
|
|
|
|
def test_incomplete_or_wrong_sized_upload_never_publishes(self):
|
|
for fault in ('omit_upload', 'wrong_size'):
|
|
api = API(); setattr(api, fault, True)
|
|
with self.assertRaises(ValueError): self.publish(api)
|
|
self.assertTrue(api.release['draft'])
|
|
self.assertFalse(any(method == 'PATCH' for method, _ in api.writes))
|
|
api.wrong_size = False
|
|
before = list(api.writes)
|
|
with self.assertRaises(ValueError): self.publish(api)
|
|
self.assertEqual(before, api.writes) # vorhandene abweichende Assets bleiben erhalten
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|