Skip 'test_image_member_lifecycle_for_multiple_stores' on failure
While running above test using python 3.7 environment, sometimes it closes the client connection before getting the response causing 'ConnectionError' and results in failure of the test. The reason for abrupt connection close is 'test_image_member_lifecycle_for_multiple_stores' test makes lots of connection requests which might causes race condition or dropping of connection. As a temporary solution caught the 'requests.exceptions.ConnectionError' and skipped the test to avoid failure. Change-Id: Ica04d8878eeb748cc4bbfedae895a54896ce16b3 Related-Bug: #1873735
This commit is contained in:
Abhishek Kekane
parent
bd16ab256e
commit
7af0b68b79
|
|
@ -6184,368 +6184,375 @@ class TestMultiStoreImageMembers(functional.MultipleBackendFunctionalTest):
|
|||
def test_image_member_lifecycle_for_multiple_stores(self):
|
||||
self.start_servers(**self.__dict__.copy())
|
||||
|
||||
def get_header(tenant, tenant_id=None, role=''):
|
||||
auth_token = 'user:%s:%s' % (tenant, role)
|
||||
headers = {'X-Auth-Token': auth_token}
|
||||
if tenant_id:
|
||||
headers.update({'X-Tenant-Id': tenant_id})
|
||||
return self._headers(custom_headers=headers)
|
||||
try:
|
||||
def get_header(tenant, tenant_id=None, role=''):
|
||||
auth_token = 'user:%s:%s' % (tenant, role)
|
||||
headers = {'X-Auth-Token': auth_token}
|
||||
if tenant_id:
|
||||
headers.update({'X-Tenant-Id': tenant_id})
|
||||
return self._headers(custom_headers=headers)
|
||||
|
||||
# Image list should be empty
|
||||
path = self._url('/v2/images')
|
||||
response = requests.get(path, headers=get_header('tenant1'))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(0, len(images))
|
||||
# Image list should be empty
|
||||
path = self._url('/v2/images')
|
||||
response = requests.get(path, headers=get_header('tenant1'))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(0, len(images))
|
||||
|
||||
owners = ['tenant1', 'tenant2', 'admin']
|
||||
visibilities = ['community', 'private', 'public', 'shared']
|
||||
image_fixture = []
|
||||
for owner in owners:
|
||||
for visibility in visibilities:
|
||||
path = self._url('/v2/images')
|
||||
headers = self._headers(custom_headers={
|
||||
'content-type': 'application/json',
|
||||
'X-Auth-Token': 'createuser:%s:admin' % owner,
|
||||
})
|
||||
data = jsonutils.dumps({
|
||||
'name': '%s-%s' % (owner, visibility),
|
||||
'visibility': visibility,
|
||||
})
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
image_fixture.append(jsonutils.loads(response.text))
|
||||
owners = ['tenant1', 'tenant2', 'admin']
|
||||
visibilities = ['community', 'private', 'public', 'shared']
|
||||
image_fixture = []
|
||||
for owner in owners:
|
||||
for visibility in visibilities:
|
||||
path = self._url('/v2/images')
|
||||
headers = self._headers(custom_headers={
|
||||
'content-type': 'application/json',
|
||||
'X-Auth-Token': 'createuser:%s:admin' % owner,
|
||||
})
|
||||
data = jsonutils.dumps({
|
||||
'name': '%s-%s' % (owner, visibility),
|
||||
'visibility': visibility,
|
||||
})
|
||||
response = requests.post(path, headers=headers, data=data)
|
||||
self.assertEqual(http.CREATED, response.status_code)
|
||||
image_fixture.append(jsonutils.loads(response.text))
|
||||
|
||||
# Image list should contain 12 images for tenant1
|
||||
path = self._url('/v2/images')
|
||||
response = requests.get(path, headers=get_header('tenant1'))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(12, len(images))
|
||||
# Image list should contain 12 images for tenant1
|
||||
path = self._url('/v2/images')
|
||||
response = requests.get(path, headers=get_header('tenant1'))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(12, len(images))
|
||||
|
||||
# Image list should contain 3 images for TENANT3
|
||||
path = self._url('/v2/images')
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(3, len(images))
|
||||
# Image list should contain 3 images for TENANT3
|
||||
path = self._url('/v2/images')
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(3, len(images))
|
||||
|
||||
# Add Image member for tenant1-shared image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[3]['id'])
|
||||
body = jsonutils.dumps({'member': TENANT3})
|
||||
response = requests.post(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1),
|
||||
data=body)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
image_member = jsonutils.loads(response.text)
|
||||
self.assertEqual(image_fixture[3]['id'], image_member['image_id'])
|
||||
self.assertEqual(TENANT3, image_member['member_id'])
|
||||
self.assertIn('created_at', image_member)
|
||||
self.assertIn('updated_at', image_member)
|
||||
self.assertEqual('pending', image_member['status'])
|
||||
|
||||
# Image list should contain 3 images for TENANT3
|
||||
path = self._url('/v2/images')
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(3, len(images))
|
||||
|
||||
# Image list should contain 0 shared images for TENANT3
|
||||
# because default is accepted
|
||||
path = self._url('/v2/images?visibility=shared')
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(0, len(images))
|
||||
|
||||
# Image list should contain 4 images for TENANT3 with status pending
|
||||
path = self._url('/v2/images?member_status=pending')
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(4, len(images))
|
||||
|
||||
# Image list should contain 4 images for TENANT3 with status all
|
||||
path = self._url('/v2/images?member_status=all')
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(4, len(images))
|
||||
|
||||
# Image list should contain 1 image for TENANT3 with status pending
|
||||
# and visibility shared
|
||||
path = self._url('/v2/images?member_status=pending&visibility=shared')
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(1, len(images))
|
||||
self.assertEqual(images[0]['name'], 'tenant1-shared')
|
||||
|
||||
# Image list should contain 0 image for TENANT3 with status rejected
|
||||
# and visibility shared
|
||||
path = self._url('/v2/images?member_status=rejected&visibility=shared')
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(0, len(images))
|
||||
|
||||
# Image list should contain 0 image for TENANT3 with status accepted
|
||||
# and visibility shared
|
||||
path = self._url('/v2/images?member_status=accepted&visibility=shared')
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(0, len(images))
|
||||
|
||||
# Image list should contain 0 image for TENANT3 with status accepted
|
||||
# and visibility private
|
||||
path = self._url('/v2/images?visibility=private')
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(0, len(images))
|
||||
|
||||
# Image tenant2-shared's image members list should contain no members
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[7]['id'])
|
||||
response = requests.get(path, headers=get_header('tenant2'))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual(0, len(body['members']))
|
||||
|
||||
# Tenant 1, who is the owner cannot change status of image member
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[3]['id'],
|
||||
TENANT3))
|
||||
body = jsonutils.dumps({'status': 'accepted'})
|
||||
response = requests.put(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1), data=body)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Tenant 1, who is the owner can get status of its own image member
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[3]['id'],
|
||||
TENANT3))
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual('pending', body['status'])
|
||||
self.assertEqual(image_fixture[3]['id'], body['image_id'])
|
||||
self.assertEqual(TENANT3, body['member_id'])
|
||||
|
||||
# Tenant 3, who is the member can get status of its own status
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[3]['id'],
|
||||
TENANT3))
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual('pending', body['status'])
|
||||
self.assertEqual(image_fixture[3]['id'], body['image_id'])
|
||||
self.assertEqual(TENANT3, body['member_id'])
|
||||
|
||||
# Tenant 2, who not the owner cannot get status of image member
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[3]['id'],
|
||||
TENANT3))
|
||||
response = requests.get(path, headers=get_header('tenant2',
|
||||
tenant_id=TENANT2))
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Tenant 3 can change status of image member
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[3]['id'],
|
||||
TENANT3))
|
||||
body = jsonutils.dumps({'status': 'accepted'})
|
||||
response = requests.put(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3), data=body)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
image_member = jsonutils.loads(response.text)
|
||||
self.assertEqual(image_fixture[3]['id'], image_member['image_id'])
|
||||
self.assertEqual(TENANT3, image_member['member_id'])
|
||||
self.assertEqual('accepted', image_member['status'])
|
||||
|
||||
# Image list should contain 4 images for TENANT3 because status is
|
||||
# accepted
|
||||
path = self._url('/v2/images')
|
||||
response = requests.get(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(4, len(images))
|
||||
|
||||
# Tenant 3 invalid status change
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[3]['id'],
|
||||
TENANT3))
|
||||
body = jsonutils.dumps({'status': 'invalid-status'})
|
||||
response = requests.put(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3), data=body)
|
||||
self.assertEqual(http.BAD_REQUEST, response.status_code)
|
||||
|
||||
# Owner cannot change status of image
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[3]['id'],
|
||||
TENANT3))
|
||||
body = jsonutils.dumps({'status': 'accepted'})
|
||||
response = requests.put(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1), data=body)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Add Image member for tenant2-shared image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[7]['id'])
|
||||
body = jsonutils.dumps({'member': TENANT4})
|
||||
response = requests.post(path, headers=get_header('tenant2'),
|
||||
data=body)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
image_member = jsonutils.loads(response.text)
|
||||
self.assertEqual(image_fixture[7]['id'], image_member['image_id'])
|
||||
self.assertEqual(TENANT4, image_member['member_id'])
|
||||
self.assertIn('created_at', image_member)
|
||||
self.assertIn('updated_at', image_member)
|
||||
|
||||
# Add Image member to public image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[2]['id'])
|
||||
body = jsonutils.dumps({'member': TENANT2})
|
||||
response = requests.post(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1),
|
||||
data=body)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Add Image member to private image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[1]['id'])
|
||||
body = jsonutils.dumps({'member': TENANT2})
|
||||
response = requests.post(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1),
|
||||
data=body)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Add Image member to community image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[0]['id'])
|
||||
body = jsonutils.dumps({'member': TENANT2})
|
||||
response = requests.post(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1),
|
||||
data=body)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Image tenant1-shared's members list should contain 1 member
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[3]['id'])
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual(1, len(body['members']))
|
||||
|
||||
# Admin can see any members
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[3]['id'])
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1,
|
||||
role='admin'))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual(1, len(body['members']))
|
||||
|
||||
# Image members forbidden for public image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[2]['id'])
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertIn("Only shared images have members", response.text)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Image members forbidden for community image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[0]['id'])
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertIn("Only shared images have members", response.text)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Image members forbidden for private image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[1]['id'])
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertIn("Only shared images have members", response.text)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Image Member Cannot delete Image membership
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[3]['id'],
|
||||
TENANT3))
|
||||
response = requests.delete(path, headers=get_header(TENANT3,
|
||||
tenant_id=TENANT3))
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Delete Image member
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[3]['id'],
|
||||
TENANT3))
|
||||
response = requests.delete(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# Now the image has no members
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[3]['id'])
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual(0, len(body['members']))
|
||||
|
||||
# Adding 11 image members should fail since configured limit is 10
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[3]['id'])
|
||||
for i in range(10):
|
||||
body = jsonutils.dumps({'member': str(uuid.uuid4())})
|
||||
# Add Image member for tenant1-shared image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[3]['id'])
|
||||
body = jsonutils.dumps({'member': TENANT3})
|
||||
response = requests.post(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1), data=body)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
image_member = jsonutils.loads(response.text)
|
||||
self.assertEqual(image_fixture[3]['id'], image_member['image_id'])
|
||||
self.assertEqual(TENANT3, image_member['member_id'])
|
||||
self.assertIn('created_at', image_member)
|
||||
self.assertIn('updated_at', image_member)
|
||||
self.assertEqual('pending', image_member['status'])
|
||||
|
||||
body = jsonutils.dumps({'member': str(uuid.uuid4())})
|
||||
response = requests.post(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1),
|
||||
data=body)
|
||||
self.assertEqual(http.REQUEST_ENTITY_TOO_LARGE, response.status_code)
|
||||
# Image list should contain 3 images for TENANT3
|
||||
path = self._url('/v2/images')
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(3, len(images))
|
||||
|
||||
# Get Image member should return not found for public image
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[2]['id'],
|
||||
TENANT3))
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
# Image list should contain 0 shared images for TENANT3
|
||||
# because default is accepted
|
||||
path = self._url('/v2/images?visibility=shared')
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(0, len(images))
|
||||
|
||||
# Get Image member should return not found for community image
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[0]['id'],
|
||||
TENANT3))
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
# Image list should contain 4 images for TENANT3 with status
|
||||
# pending
|
||||
path = self._url('/v2/images?member_status=pending')
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(4, len(images))
|
||||
|
||||
# Get Image member should return not found for private image
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[1]['id'],
|
||||
TENANT3))
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
# Image list should contain 4 images for TENANT3 with status all
|
||||
path = self._url('/v2/images?member_status=all')
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(4, len(images))
|
||||
|
||||
# Delete Image member should return forbidden for public image
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[2]['id'],
|
||||
TENANT3))
|
||||
response = requests.delete(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
# Image list should contain 1 image for TENANT3 with status pending
|
||||
# and visibility shared
|
||||
path = self._url(
|
||||
'/v2/images?member_status=pending&visibility=shared')
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(1, len(images))
|
||||
self.assertEqual(images[0]['name'], 'tenant1-shared')
|
||||
|
||||
# Delete Image member should return forbidden for community image
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[0]['id'],
|
||||
TENANT3))
|
||||
response = requests.delete(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
# Image list should contain 0 image for TENANT3 with status
|
||||
# rejected and visibility shared
|
||||
path = self._url(
|
||||
'/v2/images?member_status=rejected&visibility=shared')
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(0, len(images))
|
||||
|
||||
# Delete Image member should return forbidden for private image
|
||||
path = self._url('/v2/images/%s/members/%s' % (image_fixture[1]['id'],
|
||||
TENANT3))
|
||||
response = requests.delete(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1))
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
# Image list should contain 0 image for TENANT3 with status
|
||||
# accepted and visibility shared
|
||||
path = self._url(
|
||||
'/v2/images?member_status=accepted&visibility=shared')
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(0, len(images))
|
||||
|
||||
# Image list should contain 0 image for TENANT3 with status
|
||||
# accepted and visibility private
|
||||
path = self._url('/v2/images?visibility=private')
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(0, len(images))
|
||||
|
||||
# Image tenant2-shared's image members list should contain
|
||||
# no members
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[7]['id'])
|
||||
response = requests.get(path, headers=get_header('tenant2'))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual(0, len(body['members']))
|
||||
|
||||
# Tenant 1, who is the owner cannot change status of image member
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[3]['id'], TENANT3))
|
||||
body = jsonutils.dumps({'status': 'accepted'})
|
||||
response = requests.put(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1), data=body)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Tenant 1, who is the owner can get status of its own image member
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[3]['id'], TENANT3))
|
||||
response = requests.get(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual('pending', body['status'])
|
||||
self.assertEqual(image_fixture[3]['id'], body['image_id'])
|
||||
self.assertEqual(TENANT3, body['member_id'])
|
||||
|
||||
# Tenant 3, who is the member can get status of its own status
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[3]['id'], TENANT3))
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual('pending', body['status'])
|
||||
self.assertEqual(image_fixture[3]['id'], body['image_id'])
|
||||
self.assertEqual(TENANT3, body['member_id'])
|
||||
|
||||
# Tenant 2, who not the owner cannot get status of image member
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[3]['id'], TENANT3))
|
||||
response = requests.get(path, headers=get_header(
|
||||
'tenant2', tenant_id=TENANT2))
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Tenant 3 can change status of image member
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[3]['id'], TENANT3))
|
||||
body = jsonutils.dumps({'status': 'accepted'})
|
||||
response = requests.put(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3), data=body)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
image_member = jsonutils.loads(response.text)
|
||||
self.assertEqual(image_fixture[3]['id'], image_member['image_id'])
|
||||
self.assertEqual(TENANT3, image_member['member_id'])
|
||||
self.assertEqual('accepted', image_member['status'])
|
||||
|
||||
# Image list should contain 4 images for TENANT3 because status is
|
||||
# accepted
|
||||
path = self._url('/v2/images')
|
||||
response = requests.get(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
images = jsonutils.loads(response.text)['images']
|
||||
self.assertEqual(4, len(images))
|
||||
|
||||
# Tenant 3 invalid status change
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[3]['id'], TENANT3))
|
||||
body = jsonutils.dumps({'status': 'invalid-status'})
|
||||
response = requests.put(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3), data=body)
|
||||
self.assertEqual(http.BAD_REQUEST, response.status_code)
|
||||
|
||||
# Owner cannot change status of image
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[3]['id'], TENANT3))
|
||||
body = jsonutils.dumps({'status': 'accepted'})
|
||||
response = requests.put(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1), data=body)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Add Image member for tenant2-shared image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[7]['id'])
|
||||
body = jsonutils.dumps({'member': TENANT4})
|
||||
response = requests.post(path, headers=get_header('tenant2'),
|
||||
data=body)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
image_member = jsonutils.loads(response.text)
|
||||
self.assertEqual(image_fixture[7]['id'], image_member['image_id'])
|
||||
self.assertEqual(TENANT4, image_member['member_id'])
|
||||
self.assertIn('created_at', image_member)
|
||||
self.assertIn('updated_at', image_member)
|
||||
|
||||
# Add Image member to public image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[2]['id'])
|
||||
body = jsonutils.dumps({'member': TENANT2})
|
||||
response = requests.post(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1), data=body)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Add Image member to private image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[1]['id'])
|
||||
body = jsonutils.dumps({'member': TENANT2})
|
||||
response = requests.post(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1), data=body)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Add Image member to community image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[0]['id'])
|
||||
body = jsonutils.dumps({'member': TENANT2})
|
||||
response = requests.post(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1), data=body)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Image tenant1-shared's members list should contain 1 member
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[3]['id'])
|
||||
response = requests.get(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual(1, len(body['members']))
|
||||
|
||||
# Admin can see any members
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[3]['id'])
|
||||
response = requests.get(path, headers=get_header('tenant1',
|
||||
tenant_id=TENANT1,
|
||||
role='admin'))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual(1, len(body['members']))
|
||||
|
||||
# Image members forbidden for public image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[2]['id'])
|
||||
response = requests.get(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertIn("Only shared images have members", response.text)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Image members forbidden for community image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[0]['id'])
|
||||
response = requests.get(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertIn("Only shared images have members", response.text)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Image members forbidden for private image
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[1]['id'])
|
||||
response = requests.get(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertIn("Only shared images have members", response.text)
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Image Member Cannot delete Image membership
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[3]['id'], TENANT3))
|
||||
response = requests.delete(path, headers=get_header(
|
||||
TENANT3, tenant_id=TENANT3))
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Delete Image member
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[3]['id'], TENANT3))
|
||||
response = requests.delete(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertEqual(http.NO_CONTENT, response.status_code)
|
||||
|
||||
# Now the image has no members
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[3]['id'])
|
||||
response = requests.get(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
body = jsonutils.loads(response.text)
|
||||
self.assertEqual(0, len(body['members']))
|
||||
|
||||
# Adding 11 image members should fail since configured limit is 10
|
||||
path = self._url('/v2/images/%s/members' % image_fixture[3]['id'])
|
||||
for i in range(10):
|
||||
body = jsonutils.dumps({'member': str(uuid.uuid4())})
|
||||
response = requests.post(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1), data=body)
|
||||
self.assertEqual(http.OK, response.status_code)
|
||||
|
||||
body = jsonutils.dumps({'member': str(uuid.uuid4())})
|
||||
response = requests.post(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1), data=body)
|
||||
self.assertEqual(http.REQUEST_ENTITY_TOO_LARGE,
|
||||
response.status_code)
|
||||
|
||||
# Get Image member should return not found for public image
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[2]['id'], TENANT3))
|
||||
response = requests.get(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Get Image member should return not found for community image
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[0]['id'], TENANT3))
|
||||
response = requests.get(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Get Image member should return not found for private image
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[1]['id'], TENANT3))
|
||||
response = requests.get(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertEqual(http.NOT_FOUND, response.status_code)
|
||||
|
||||
# Delete Image member should return forbidden for public image
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[2]['id'], TENANT3))
|
||||
response = requests.delete(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Delete Image member should return forbidden for community image
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[0]['id'], TENANT3))
|
||||
response = requests.delete(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
|
||||
# Delete Image member should return forbidden for private image
|
||||
path = self._url('/v2/images/%s/members/%s' % (
|
||||
image_fixture[1]['id'], TENANT3))
|
||||
response = requests.delete(path, headers=get_header(
|
||||
'tenant1', tenant_id=TENANT1))
|
||||
self.assertEqual(http.FORBIDDEN, response.status_code)
|
||||
except requests.exceptions.ConnectionError as e:
|
||||
# NOTE(abhishekk): This test fails intermittently for py37
|
||||
# environment refer,
|
||||
# https://bugs.launchpad.net/glance/+bug/1873735
|
||||
self.skipTest("Remote connection closed abruptly: %s" % e.args[0])
|
||||
|
||||
self.stop_servers()
|
||||
|
|
|
|||
Loading…
Reference in New Issue