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

Implement the remaining registry tests in the new py.test format

Browse source
This commit is contained in:
Joseph Schorr 2018-05-01 13:26:47 +03:00
parent 77adf9dd77
commit 8c1b0e673c
7 changed files with 1200 additions and 62 deletions
Download patch file Download diff file Expand all files Collapse all files

95
test/registry/protocol_v1.py
View file

@ -12,6 +12,7 @@ class V1ProtocolSteps(Enum):
""" Defines the various steps of the protocol, for matching failures. """
PUT_IMAGES = 'put-images'
GET_IMAGES = 'get-images'
PUT_TAG = 'put-tag'
class V1Protocol(RegistryProtocol):
@ -19,10 +20,21 @@ class V1Protocol(RegistryProtocol):
V1ProtocolSteps.PUT_IMAGES: {
Failures.UNAUTHENTICATED: 401,
Failures.UNAUTHORIZED: 403,
Failures.APP_REPOSITORY: 405,
Failures.INVALID_REPOSITORY: 404,
Failures.DISALLOWED_LIBRARY_NAMESPACE: 400,
},
V1ProtocolSteps.GET_IMAGES: {
Failures.UNAUTHENTICATED: 403,
Failures.UNAUTHORIZED: 403,
Failures.APP_REPOSITORY: 405,
Failures.ANONYMOUS_NOT_ALLOWED: 401,
Failures.DISALLOWED_LIBRARY_NAMESPACE: 400,
},
V1ProtocolSteps.PUT_TAG: {
Failures.MISSING_TAG: 404,
Failures.INVALID_TAG: 400,
Failures.INVALID_IMAGES: 400,
},
}
@ -38,53 +50,70 @@ class V1Protocol(RegistryProtocol):
def ping(self, session):
assert session.get('/v1/_ping').status_code == 200
def login(self, session, username, password, scopes, expect_success):
data = {
'username': username,
'password': password,
}
response = self.conduct(session, 'POST', '/v1/users/', json_data=data, expected_status=400)
assert (response.text == '"Username or email already exists"') == expect_success
def pull(self, session, namespace, repo_name, tag_names, images, credentials=None,
expected_failure=None, options=None):
options = options or ProtocolOptions()
auth = self._auth_for_credentials(credentials)
tag_names = [tag_names] if isinstance(tag_names, str) else tag_names
prefix = '/v1/repositories/%s/%s/' % (namespace, repo_name)
prefix = '/v1/repositories/%s/' % self.repo_name(namespace, repo_name)
# Ping!
self.ping(session)
# GET /v1/repositories/{namespace}/{repository}/images
result = self.conduct(session, 'GET', prefix + 'images', auth=auth,
headers = {'X-Docker-Token': 'true'}
result = self.conduct(session, 'GET', prefix + 'images', auth=auth, headers=headers,
expected_status=(200, expected_failure, V1ProtocolSteps.GET_IMAGES))
if expected_failure is not None:
return
headers = {}
if credentials is not None:
headers['Authorization'] = 'token ' + result.headers['www-authenticate']
else:
assert not 'www-authenticate' in result.headers
# GET /v1/repositories/{namespace}/{repository}/tags
tags_result = self.conduct(session, 'GET', prefix + 'tags', auth=auth).json()
assert len(tags_result.values()) == len(tag_names)
image_ids = self.conduct(session, 'GET', prefix + 'tags', headers=headers).json()
assert len(image_ids.values()) == len(tag_names)
tag_image_id = tags_result['latest']
if not options.munge_shas:
# Ensure we have a matching image ID.
known_ids = {image.id for image in images}
assert tag_image_id in known_ids
for tag_name in tag_names:
tag_image_id = image_ids[tag_name]
if not options.munge_shas:
# Ensure we have a matching image ID.
known_ids = {image.id for image in images}
assert tag_image_id in known_ids
# Retrieve the ancestry of the tagged image.
image_prefix = '/v1/images/%s/' % tag_image_id
ancestors = self.conduct(session, 'GET', image_prefix + 'ancestry', auth=auth).json()
# Retrieve the ancestry of the tagged image.
image_prefix = '/v1/images/%s/' % tag_image_id
ancestors = self.conduct(session, 'GET', image_prefix + 'ancestry', headers=headers).json()
assert len(ancestors) == len(images)
for index, image_id in enumerate(reversed(ancestors)):
# /v1/images/{imageID}/{ancestry, json, layer}
image_prefix = '/v1/images/%s/' % image_id
self.conduct(session, 'GET', image_prefix + 'ancestry', auth=auth)
assert len(ancestors) == len(images)
for index, image_id in enumerate(reversed(ancestors)):
# /v1/images/{imageID}/{ancestry, json, layer}
image_prefix = '/v1/images/%s/' % image_id
self.conduct(session, 'GET', image_prefix + 'ancestry', headers=headers)
result = self.conduct(session, 'GET', image_prefix + 'json', auth=auth)
assert result.json()['id'] == image_id
result = self.conduct(session, 'GET', image_prefix + 'json', headers=headers)
assert result.json()['id'] == image_id
# Ensure we can HEAD the image layer.
self.conduct(session, 'HEAD', image_prefix + 'layer', auth=auth)
# Ensure we can HEAD the image layer.
self.conduct(session, 'HEAD', image_prefix + 'layer', headers=headers)
# And retrieve the layer data.
result = self.conduct(session, 'GET', image_prefix + 'layer', auth=auth)
assert result.content == images[index].bytes
# And retrieve the layer data.
result = self.conduct(session, 'GET', image_prefix + 'layer', headers=headers)
assert result.content == images[index].bytes
return PullResult(manifests=None)
return PullResult(manifests=None, image_ids=image_ids)
def push(self, session, namespace, repo_name, tag_names, images, credentials=None,
expected_failure=None, options=None):
@ -95,7 +124,8 @@ class V1Protocol(RegistryProtocol):
self.ping(session)
# PUT /v1/repositories/{namespace}/{repository}/
result = self.conduct(session, 'PUT', '/v1/repositories/%s/%s/' % (namespace, repo_name),
result = self.conduct(session, 'PUT',
'/v1/repositories/%s/' % self.repo_name(namespace, repo_name),
expected_status=(201, expected_failure, V1ProtocolSteps.PUT_IMAGES),
json_data={},
auth=auth)
@ -115,6 +145,9 @@ class V1Protocol(RegistryProtocol):
if image.parent_id is not None:
image_json_data['parent'] = image.parent_id
if image.config is not None:
image_json_data['config'] = image.config
self.conduct(session, 'PUT', '/v1/images/%s/json' % image.id,
json_data=image_json_data, headers=headers)
@ -133,12 +166,14 @@ class V1Protocol(RegistryProtocol):
# PUT /v1/repositories/{namespace}/{repository}/tags/latest
for tag_name in tag_names:
self.conduct(session, 'PUT',
'/v1/repositories/%s/%s/tags/%s' % (namespace, repo_name, tag_name),
'/v1/repositories/%s/tags/%s' % (self.repo_name(namespace, repo_name), tag_name),
data='"%s"' % images[-1].id,
headers=headers)
headers=headers,
expected_status=(200, expected_failure, V1ProtocolSteps.PUT_TAG))
# PUT /v1/repositories/{namespace}/{repository}/images
self.conduct(session, 'PUT', '/v1/repositories/%s/%s/images' % (namespace, repo_name),
self.conduct(session, 'PUT',
'/v1/repositories/%s/images' % self.repo_name(namespace, repo_name),
expected_status=204, headers=headers)
return PushResult(checksums=None, manifests=None)
return PushResult(checksums=None, manifests=None, headers=headers)