diff --git a/glance/tests/functional/v2/test_images.py b/glance/tests/functional/v2/test_images.py index 61825fe219..eb4256ab41 100644 --- a/glance/tests/functional/v2/test_images.py +++ b/glance/tests/functional/v2/test_images.py @@ -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()