khodowany
57 lines
3.4 KiB
Python
57 lines
3.4 KiB
Python
import modulus.build_context
|
|
import os
|
|
import pathlib
|
|
import tempfile
|
|
import unittest
|
|
import unittest.mock
|
|
import urllib.error
|
|
|
|
import modulus.artifacts
|
|
import modulus.config
|
|
|
|
class ArtifactTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = pathlib.Path(self.temporary.name)
|
|
config = modulus.config.normalize({'package': {'name': 'example', 'artifact': {'enabled': True, 'targets': ['arm_64_gcc12_release']}}})
|
|
self.context = modulus.build_context.build_context(self.root, config)
|
|
self.directory = self.root / 'build/packages'
|
|
self.directory.mkdir(parents=True)
|
|
self.archive = self.directory / modulus.artifacts.archive_name('example', 'arm_64_gcc12_release')
|
|
self.archive.write_bytes(b'package')
|
|
self.checksum = modulus.artifacts.write_checksum(self.archive)
|
|
self.enterContext(unittest.mock.patch.dict(os.environ, {'PACKAGE_REGISTRY_USER': 'test', 'PACKAGE_REGISTRY_TOKEN': 'secret'}))
|
|
def test_discovery_uses_the_same_names_as_creation(self):
|
|
self.assertEqual(self.archive.name, 'example_arm_64_gcc12_release.tar.gz')
|
|
self.assertEqual(modulus.artifacts.discover_artifacts(self.directory, 'example', ['arm_64_gcc12_release']), [self.archive, self.checksum])
|
|
def test_missing_or_invalid_checksums_never_touch_registry(self):
|
|
with unittest.mock.patch.object(modulus.artifacts, 'request') as request:
|
|
self.archive.write_bytes(b'corrupt')
|
|
with self.assertRaisesRegex(SystemExit, 'checksum mismatch'):
|
|
modulus.artifacts.main(self.context, [])
|
|
self.checksum.unlink()
|
|
with self.assertRaisesRegex(SystemExit, 'missing artifact'):
|
|
modulus.artifacts.main(self.context, [])
|
|
request.assert_not_called()
|
|
def test_replacement_and_cleanup_are_ordered_after_validation(self):
|
|
events = []
|
|
with unittest.mock.patch.object(modulus.artifacts, 'list_versions', return_value=['latest', 'old']), \
|
|
unittest.mock.patch.object(modulus.artifacts, 'delete_version', side_effect=lambda *args: events.append(('delete', args[-2]))), \
|
|
unittest.mock.patch.object(modulus.artifacts, 'upload_file', side_effect=lambda *args: events.append(('upload', args[2].name))):
|
|
self.assertEqual(modulus.artifacts.main(self.context, []), 0)
|
|
self.assertEqual(events, [('delete', 'latest'), ('upload', self.archive.name), ('upload', self.checksum.name), ('delete', 'old')])
|
|
def test_failed_upload_does_not_delete_other_slots(self):
|
|
with unittest.mock.patch.object(modulus.artifacts, 'list_versions', return_value=['latest', 'old']), \
|
|
unittest.mock.patch.object(modulus.artifacts, 'delete_version') as delete, \
|
|
unittest.mock.patch.object(modulus.artifacts, 'upload_file', side_effect=SystemExit('upload failed')):
|
|
with self.assertRaisesRegex(SystemExit, 'upload failed'):
|
|
modulus.artifacts.main(self.context, [])
|
|
self.assertEqual([call.args[-2] for call in delete.call_args_list], ['latest'])
|
|
def test_request_errors_and_credentials_are_not_logged(self):
|
|
for code in (401, 500):
|
|
with unittest.mock.patch.object(modulus.artifacts.urllib.request, 'urlopen', side_effect=urllib.error.HTTPError('https://example.test', code, 'failed', {}, None)):
|
|
with self.assertRaises(SystemExit) as caught:
|
|
modulus.artifacts.request('PUT', 'https://example.test', 'secret')
|
|
self.assertNotIn('secret', str(caught.exception))
|