vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Add blob mounting tests to the new registry test suite

Browse source
This commit is contained in:
Joseph Schorr 2018-05-30 16:49:46 -04:00
parent 0fa1a1d5fd
commit f6eaf7ce9d
3 changed files with 112 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

128
test/registry/protocol_v2.py
View file

@ -15,6 +15,7 @@ class V2ProtocolSteps(Enum):
BLOB_HEAD_CHECK = 'blob-head-check'
GET_MANIFEST = 'get-manifest'
PUT_MANIFEST = 'put-manifest'
MOUNT_BLOB = 'mount-blob'
class V2Protocol(RegistryProtocol):
@ -27,6 +28,9 @@ class V2Protocol(RegistryProtocol):
Failures.INVALID_REPOSITORY: 400,
Failures.NAMESPACE_DISABLED: 400,
},
V2ProtocolSteps.MOUNT_BLOB: {
Failures.UNAUTHORIZED_FOR_MOUNT: 202,
},
V2ProtocolSteps.GET_MANIFEST: {
Failures.UNKNOWN_TAG: 404,
Failures.UNAUTHORIZED: 403,
@ -92,8 +96,7 @@ class V2Protocol(RegistryProtocol):
}
if scopes:
params['scope'] = 'repository:%s:%s' % (self.repo_name(namespace, repo_name),
','.join(scopes))
params['scope'] = scopes
response = self.conduct(session, 'GET', '/v2/auth', params=params, auth=auth,
expected_status=(200, expected_failure, V2ProtocolSteps.AUTH))
@ -107,7 +110,7 @@ class V2Protocol(RegistryProtocol):
def push(self, session, namespace, repo_name, tag_names, images, credentials=None,
expected_failure=None, options=None):
options = options or ProtocolOptions()
scopes = options.scopes or ['push', 'pull']
scopes = options.scopes or ['repository:%s:push,pull' % self.repo_name(namespace, repo_name)]
tag_names = [tag_names] if isinstance(tag_names, str) else tag_names
# Ping!
@ -160,64 +163,79 @@ class V2Protocol(RegistryProtocol):
expected_status=(404, expected_failure, V2ProtocolSteps.BLOB_HEAD_CHECK),
headers=headers)
# Start a new upload of the layer data.
response = self.conduct(session, 'POST',
'/v2/%s/blobs/uploads/' % self.repo_name(namespace, repo_name),
expected_status=202,
headers=headers)
upload_uuid = response.headers['Docker-Upload-UUID']
new_upload_location = response.headers['Location']
assert new_upload_location.startswith('http://localhost:5000')
# We need to make this relative just for the tests because the live server test
# case modifies the port.
location = response.headers['Location'][len('http://localhost:5000'):]
# PATCH the image data into the layer.
if options.chunks_for_upload is None:
self.conduct(session, 'PATCH', location, data=image.bytes, expected_status=204,
# Check for mounting of blobs.
if options.mount_blobs and image.id in options.mount_blobs:
self.conduct(session, 'POST',
'/v2/%s/blobs/uploads/' % self.repo_name(namespace, repo_name),
params={
'mount': checksum,
'from': options.mount_blobs[image.id],
},
expected_status=(201, expected_failure, V2ProtocolSteps.MOUNT_BLOB),
headers=headers)
if expected_failure is not None:
return
else:
# If chunked upload is requested, upload the data as a series of chunks, checking
# status at every point.
for chunk_data in options.chunks_for_upload:
if len(chunk_data) == 3:
(start_byte, end_byte, expected_code) = chunk_data
else:
(start_byte, end_byte) = chunk_data
expected_code = 204
# Start a new upload of the layer data.
response = self.conduct(session, 'POST',
'/v2/%s/blobs/uploads/' % self.repo_name(namespace, repo_name),
expected_status=202,
headers=headers)
patch_headers = {'Range': 'bytes=%s-%s' % (start_byte, end_byte)}
patch_headers.update(headers)
upload_uuid = response.headers['Docker-Upload-UUID']
new_upload_location = response.headers['Location']
assert new_upload_location.startswith('http://localhost:5000')
contents_chunk = image.bytes[start_byte:end_byte]
self.conduct(session, 'PATCH', location, data=contents_chunk,
expected_status=expected_code,
headers=patch_headers)
if expected_code != 204:
return
# We need to make this relative just for the tests because the live server test
# case modifies the port.
location = response.headers['Location'][len('http://localhost:5000'):]
# Retrieve the upload status at each point, and ensure it is valid.
# PATCH the image data into the layer.
if options.chunks_for_upload is None:
self.conduct(session, 'PATCH', location, data=image.bytes, expected_status=204,
headers=headers)
else:
# If chunked upload is requested, upload the data as a series of chunks, checking
# status at every point.
for chunk_data in options.chunks_for_upload:
if len(chunk_data) == 3:
(start_byte, end_byte, expected_code) = chunk_data
else:
(start_byte, end_byte) = chunk_data
expected_code = 204
patch_headers = {'Range': 'bytes=%s-%s' % (start_byte, end_byte)}
patch_headers.update(headers)
contents_chunk = image.bytes[start_byte:end_byte]
self.conduct(session, 'PATCH', location, data=contents_chunk,
expected_status=expected_code,
headers=patch_headers)
if expected_code != 204:
return
# Retrieve the upload status at each point, and ensure it is valid.
status_url = '/v2/%s/blobs/uploads/%s' % (self.repo_name(namespace, repo_name),
upload_uuid)
response = self.conduct(session, 'GET', status_url, expected_status=204,
headers=headers)
assert response.headers['Docker-Upload-UUID'] == upload_uuid
assert response.headers['Range'] == "bytes=0-%s" % end_byte
if options.cancel_blob_upload:
self.conduct(session, 'DELETE', location, params=dict(digest=checksum),
expected_status=204, headers=headers)
# Ensure the upload was canceled.
status_url = '/v2/%s/blobs/uploads/%s' % (self.repo_name(namespace, repo_name),
upload_uuid)
response = self.conduct(session, 'GET', status_url, expected_status=204, headers=headers)
assert response.headers['Docker-Upload-UUID'] == upload_uuid
assert response.headers['Range'] == "bytes=0-%s" % end_byte
self.conduct(session, 'GET', status_url, expected_status=404, headers=headers)
return
if options.cancel_blob_upload:
self.conduct(session, 'DELETE', location, params=dict(digest=checksum), expected_status=204,
headers=headers)
# Ensure the upload was canceled.
status_url = '/v2/%s/blobs/uploads/%s' % (self.repo_name(namespace, repo_name), upload_uuid)
self.conduct(session, 'GET', status_url, expected_status=404, headers=headers)
return
# Finish the layer upload with a PUT.
response = self.conduct(session, 'PUT', location, params=dict(digest=checksum),
expected_status=201, headers=headers)
assert response.headers['Docker-Content-Digest'] == checksum
# Finish the layer upload with a PUT.
response = self.conduct(session, 'PUT', location, params=dict(digest=checksum),
expected_status=201, headers=headers)
assert response.headers['Docker-Content-Digest'] == checksum
# Ensure the layer exists now.
response = self.conduct(session, 'HEAD',
@ -258,7 +276,7 @@ class V2Protocol(RegistryProtocol):
def delete(self, session, namespace, repo_name, tag_names, credentials=None,
expected_failure=None, options=None):
options = options or ProtocolOptions()
scopes = options.scopes or ['*']
scopes = options.scopes or ['repository:%s:*' % self.repo_name(namespace, repo_name)]
tag_names = [tag_names] if isinstance(tag_names, str) else tag_names
# Ping!
@ -284,7 +302,7 @@ class V2Protocol(RegistryProtocol):
def pull(self, session, namespace, repo_name, tag_names, images, credentials=None,
expected_failure=None, options=None):
options = options or ProtocolOptions()
scopes = options.scopes or ['pull']
scopes = options.scopes or ['repository:%s:pull' % self.repo_name(namespace, repo_name)]
tag_names = [tag_names] if isinstance(tag_names, str) else tag_names
# Ping!

2
test/registry/protocols.py
View file

@ -51,6 +51,7 @@ class Failures(Enum):
UNSUPPORTED_CONTENT_TYPE = 'unsupported-content-type'
INVALID_BLOB = 'invalid-blob'
NAMESPACE_DISABLED = 'namespace-disabled'
UNAUTHORIZED_FOR_MOUNT = 'unauthorized-for-mount'
class ProtocolOptions(object):
@ -62,6 +63,7 @@ class ProtocolOptions(object):
self.chunks_for_upload = None
self.skip_head_checks = False
self.manifest_content_type = None
self.mount_blobs = None
@add_metaclass(ABCMeta)

37
test/registry/registry_tests.py
View file

@ -783,6 +783,43 @@ def test_squashed_images(use_estimates, pusher, sized_images, liveserver_session
tarfile.open(fileobj=tar.extractfile(tar.getmember('%s/layer.tar' % expected_image_id)))
@pytest.mark.parametrize('push_user, push_namespace, push_repo, mount_repo_name, expected_failure', [
# Successful mount, same namespace.
('devtable', 'devtable', 'baserepo', 'devtable/baserepo', None),
# Successful mount, cross namespace.
('devtable', 'buynlarge', 'baserepo', 'buynlarge/baserepo', None),
# Unsuccessful mount, unknown repo.
('devtable', 'devtable', 'baserepo', 'unknown/repohere', Failures.UNAUTHORIZED_FOR_MOUNT),
# Unsuccessful mount, no access.
('public', 'public', 'baserepo', 'public/baserepo', Failures.UNAUTHORIZED_FOR_MOUNT),
])
def test_blob_mounting(push_user, push_namespace, push_repo, mount_repo_name, expected_failure,
manifest_protocol, pusher, puller, basic_images, liveserver_session,
app_reloader):
# Push an image so we can attempt to mount it.
pusher.push(liveserver_session, push_namespace, push_repo, 'latest', basic_images,
credentials=(push_user, 'password'))
# Push again, trying to mount the image layer(s) from the mount repo.
options = ProtocolOptions()
options.scopes = ['repository:devtable/newrepo:push,pull',
'repository:%s:pull' % (mount_repo_name)]
options.mount_blobs = {image.id: mount_repo_name for image in basic_images}
manifest_protocol.push(liveserver_session, 'devtable', 'newrepo', 'latest', basic_images,
credentials=('devtable', 'password'),
options=options,
expected_failure=expected_failure)
if expected_failure is None:
# Pull to ensure it worked.
puller.pull(liveserver_session, 'devtable', 'newrepo', 'latest', basic_images,
credentials=('devtable', 'password'))
def get_robot_password(api_caller):
api_caller.conduct_auth('devtable', 'password')
resp = api_caller.get('/api/v1/organization/buynlarge/robots/ownerbot')