Remove ACI tests because they are brittle
This commit is contained in:
Joseph Schorr
parent
3ce8a0279a
commit
65c1594a81
1 changed files with 0 additions and 185 deletions
|
|
@ -2085,191 +2085,6 @@ class TorrentV2PushTests(RegistryTestCaseMixin, TorrentTestMixin, V2RegistryPush
|
|||
pass
|
||||
|
||||
|
||||
class ACIConversionTests(RegistryTestCaseMixin, V1RegistryPushMixin, LiveServerTestCase):
|
||||
""" Tests for registry ACI conversion. """
|
||||
|
||||
def get_converted_image(self, tag_name='latest'):
|
||||
url = '/c1/aci/localhost:5000/devtable/newrepo/' + tag_name + '/aci/linux/amd64/'
|
||||
response = self.conduct('GET', url, auth='sig')
|
||||
tar = tarfile.open(fileobj=StringIO(response.content))
|
||||
return tar, response.content
|
||||
|
||||
def get_converted_signature(self, tag_name='latest'):
|
||||
counter = 0
|
||||
|
||||
# Give time for the signature to be written before continuing. As we don't exactly know when
|
||||
# this is (based on CPU conditions when the test is being run), we try a backoff and sleep
|
||||
# approach.
|
||||
while counter < 10:
|
||||
url = '/c1/aci/localhost:5000/devtable/newrepo/' + tag_name + '/aci.asc/linux/amd64/'
|
||||
response = self.conduct('GET', url, auth='sig', expected_code=None)
|
||||
if response.status_code == 202 or response.status_code == 404:
|
||||
counter += 1
|
||||
time.sleep(counter * 2)
|
||||
else:
|
||||
return response.content
|
||||
|
||||
self.fail('Signature was never created')
|
||||
|
||||
def _verify_signature(self, signature, converted):
|
||||
sig_bytes = StringIO(signature)
|
||||
content_bytes = StringIO(converted)
|
||||
|
||||
ctx = gpgme.Context()
|
||||
sigs = ctx.verify(sig_bytes, content_bytes, None)
|
||||
|
||||
self.assertEqual(len(sigs), 1)
|
||||
self.assertEqual(sigs[0].summary, 0)
|
||||
self.assertEqual(sigs[0].fpr, '07692864E17025DD1BEA88E44632047EEEB32221')
|
||||
self.assertEqual(sigs[0].status, None)
|
||||
self.assertEqual(sigs[0].notations, [])
|
||||
self.assertEqual(sigs[0].exp_timestamp, 0)
|
||||
self.assertEqual(sigs[0].wrong_key_usage, False)
|
||||
self.assertEqual(sigs[0].validity, gpgme.VALIDITY_UNKNOWN)
|
||||
self.assertEqual(sigs[0].validity_reason, None)
|
||||
|
||||
|
||||
def test_basic_conversion(self):
|
||||
if os.environ.get('RUN_ACI_TESTS') == 'False':
|
||||
return
|
||||
|
||||
initial_images = [
|
||||
{
|
||||
'id': 'initialid',
|
||||
'contents': 'the initial image',
|
||||
},
|
||||
]
|
||||
|
||||
# Create the repo.
|
||||
self.do_push('devtable', 'newrepo', 'devtable', 'password', images=initial_images)
|
||||
|
||||
# Pull the squashed version of the tag.
|
||||
tar, converted = self.get_converted_image()
|
||||
signature = self.get_converted_signature()
|
||||
first_hash = hashlib.sha256(converted).hexdigest()
|
||||
|
||||
# Verify the manifest.
|
||||
self.assertEquals(['manifest', 'rootfs', 'rootfs/contents'], tar.getnames())
|
||||
|
||||
manifest = json.loads(tar.extractfile(tar.getmember('manifest')).read())
|
||||
expected_manifest = {
|
||||
"acKind": "ImageManifest",
|
||||
"app": {
|
||||
"environment": [],
|
||||
"mountPoints": [],
|
||||
"group": "root",
|
||||
"user": "root",
|
||||
"workingDirectory": "/",
|
||||
"exec": [],
|
||||
"isolators": [],
|
||||
"eventHandlers": [],
|
||||
"ports": [],
|
||||
"annotations": [
|
||||
{"name": "created", "value": ""},
|
||||
{"name": "homepage", "value": "http://localhost:5000/devtable/newrepo:latest"},
|
||||
{"name": "quay.io/derived-image",
|
||||
"value": "fa916d5ca4da5348628dfffcfc943288a0cca521cd21a6d2981a85ec1d7f7a3a"}
|
||||
]
|
||||
},
|
||||
"labels": [
|
||||
{"name": "version", "value": "latest"},
|
||||
{"name": "arch", "value": "amd64"},
|
||||
{"name": "os", "value": "linux"}
|
||||
],
|
||||
"acVersion": "0.6.1",
|
||||
"name": "localhost/devtable/newrepo"
|
||||
}
|
||||
|
||||
self.assertEquals(manifest, expected_manifest)
|
||||
self.assertEquals('the initial image', tar.extractfile(tar.getmember('rootfs/contents')).read())
|
||||
|
||||
# Verify the signature.
|
||||
self._verify_signature(signature, converted)
|
||||
|
||||
# Clear the cache and pull again, ensuring that the hash does not change even for a completely
|
||||
# new generation of the image.
|
||||
self.conduct('POST', '/__test/clearderivedcache')
|
||||
|
||||
# Ensures the generated signature will be different.
|
||||
time.sleep(1)
|
||||
|
||||
_, converted_again = self.get_converted_image()
|
||||
second_hash = hashlib.sha256(converted_again).hexdigest()
|
||||
self.assertEquals(second_hash, first_hash)
|
||||
|
||||
# Ensure we have a different signature (and therefore the cache was broken).
|
||||
signature_again = self.get_converted_signature()
|
||||
self.assertNotEquals(signature_again, signature)
|
||||
|
||||
# Ensure *both* signatures work for both images.
|
||||
self._verify_signature(signature, converted_again)
|
||||
self._verify_signature(signature_again, converted)
|
||||
self._verify_signature(signature_again, converted_again)
|
||||
|
||||
def assertHasDerivedImage(self, manifest, expected):
|
||||
for annotation in manifest['app']['annotations']:
|
||||
if annotation['name'] == 'homepage':
|
||||
self.assertEqual(expected, annotation['value'])
|
||||
return
|
||||
|
||||
self.fail('Derived image annotation not found in metadata')
|
||||
|
||||
def test_conversion_different_tags(self):
|
||||
if os.environ.get('RUN_ACI_TESTS') == 'False':
|
||||
return
|
||||
|
||||
initial_images = [
|
||||
{
|
||||
'id': 'initialid',
|
||||
'contents': 'the initial image',
|
||||
},
|
||||
]
|
||||
|
||||
# Create the repo.
|
||||
self.do_push('devtable', 'newrepo', 'devtable', 'password', images=initial_images,
|
||||
tag_names=['latest', 'sometag'])
|
||||
|
||||
# Pull the squashed version of tag latest.
|
||||
latest_tar, _ = self.get_converted_image(tag_name='latest')
|
||||
latest_manifest = json.loads(latest_tar.extractfile(latest_tar.getmember('manifest')).read())
|
||||
self.assertHasDerivedImage(latest_manifest, 'http://localhost:5000/devtable/newrepo:latest')
|
||||
|
||||
# Pull the squashed version of tag sometag.
|
||||
sometag_tar, _ = self.get_converted_image(tag_name='sometag')
|
||||
sometag_manifest = json.loads(sometag_tar.extractfile(sometag_tar.getmember('manifest')).read())
|
||||
self.assertHasDerivedImage(sometag_manifest, 'http://localhost:5000/devtable/newrepo:sometag')
|
||||
|
||||
|
||||
def test_multilayer_conversion(self):
|
||||
if os.environ.get('RUN_ACI_TESTS') == 'False':
|
||||
return
|
||||
|
||||
images = [
|
||||
{
|
||||
'id': 'baseid',
|
||||
'contents': 'The base image',
|
||||
},
|
||||
{
|
||||
'id': 'latestid',
|
||||
'contents': 'the latest image',
|
||||
'parent': 'baseid',
|
||||
}
|
||||
]
|
||||
|
||||
# Create the repo.
|
||||
self.do_push('devtable', 'newrepo', 'devtable', 'password', images=images)
|
||||
|
||||
# Pull the squashed version of the tag.
|
||||
tar, converted = self.get_converted_image()
|
||||
signature = self.get_converted_signature()
|
||||
|
||||
self.assertEquals(['manifest', 'rootfs', 'rootfs/contents'], tar.getnames())
|
||||
self.assertEquals('the latest image', tar.extractfile(tar.getmember('rootfs/contents')).read())
|
||||
|
||||
# Verify the signature.
|
||||
self._verify_signature(signature, converted)
|
||||
|
||||
|
||||
class SquashingTests(RegistryTestCaseMixin, V1RegistryPushMixin, LiveServerTestCase):
|
||||
""" Tests for registry squashing. """
|
||||
|
||||
|
|
|
|||
Reference in a new issue