Chi Lo 446168b7bd Display Ranger error response messasge in tempest exception
When resource creation failed, the region's error message is
retrieved and included as part of the tempest exception
message.

Change-Id: I6309b29ad516651b9a89ffe54763d11fb56949a4
2020-11-10 07:14:33 -08:00

587 lines
24 KiB
Python

# Copyright 2016 AT&T Corp
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo_concurrency import lockutils
from ranger_tempest_plugin import data_utils as orm_data_utils
from ranger_tempest_plugin.tests.api import base
from ranger_tempest_plugin.tests.api import ims_base
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions
import testtools
CONF = config.CONF
PREFIX = 'alt_region'
SYNC = lockutils.synchronized_with_prefix(PREFIX)
class TestTempestIms(ims_base.ImsBaseOrmTest):
@classmethod
def skip_checks(cls):
super(TestTempestIms, cls).skip_checks()
if not CONF.ranger.image_url:
raise cls.skipException("Image not available")
@classmethod
def setup_clients(cls):
super(TestTempestIms, cls).setup_clients()
@classmethod
def resource_setup(cls):
super(TestTempestIms, cls).resource_setup()
# setup image parameters for tempest testing
cls.image_params_public = \
cls._get_image_params(set_shared=False)
cls.image_params_shared = \
cls._get_image_params(set_enabled=False)
# setup public image for tempest testing
cls.public_image = \
cls._create_img_and_validate_creation_on_dcp_and_lcp(
**cls.image_params_public)
cls.addClassResourceCleanup(
cls._del_img_validate_deletion_on_dcp_and_lcp,
cls.image_params_public['id'])
# setup shared image for tempest testing
cls.shared_image = \
cls._create_img_and_validate_creation_on_dcp_and_lcp(
**cls.image_params_shared)
cls.addClassResourceCleanup(
cls._del_img_validate_deletion_on_dcp_and_lcp,
cls.image_params_shared['id'])
def _update_region(self, region_name, status=None):
if status is None:
status = {'status': 'functional'}
self.os_admin.rms_client.update_region_status(
region_name, status)
def _data_setup(self, post_body):
image = self._create_img_and_validate_creation_on_dcp_and_lcp(
**post_body)
self.addCleanup(
self._del_img_validate_deletion_on_dcp_and_lcp,
post_body['id'])
# only check for Success image status if "regions" is not empty
if image["regions"]:
self._wait_for_image_status_on_dcp(image['id'], 'Success')
return image
def _delete_image(self, id_):
"""Try to delete from from dcp only"""
# delete the data and do get_image to ensure 404-NotFound response
self._del_img_validate_deletion_on_dcp_and_lcp(id_)
self.assertRaises(exceptions.NotFound, self.client.get_image, id_)
def _create_region_group(self):
_, body = self.rms_client.create_region_group(
**orm_data_utils.rand_region_group([CONF.identity.region,
CONF.ranger.alt_region]))
group_id = body['group']['id']
self.addCleanup(self.rms_client.delete_region_group, group_id)
return group_id
@decorators.idempotent_id('2b1bb28b-4151-4e75-ae1b-d21089c3418c')
def test_get_image(self):
"""Execute 'get_image' using the following options:
- get image by id (using cust_id parameter)
- get image by name (using cust_name parameter)
"""
# execute get_image using image ID and iamge_name
for identifier in [self.public_image['id'], self.public_image['name']]:
_, body = self.client.get_image(identifier)
self.assertIn(self.public_image['id'], body['image']['id'])
@decorators.idempotent_id('2cb3eec0-d65b-4aae-9af2-d8fd1dde84be')
def test_list_images_with_filters(self):
"""This function executes 'list customer' with all available filters:
- no filter (i.e. list all images)
- filter by region
- filter by customer
- filter by visibility (public/shared)
"""
# define the list customer filters to be used for this test
no_filter = None
customer_filter = "?customer=%s" % self.tenant_id
region_filter = "?region=%s" % self.region_id
public_filter = "?visibility=public"
shared_filter = "?visibility=shared"
# list public images
_, body = self.client.list_images(public_filter)
image_ids = [img['id'] for img in body['images']]
self.assertIn(self.public_image['id'], image_ids)
# list shared images
_, body = self.client.list_images(shared_filter)
image_ids = [img['id'] for img in body['images']]
self.assertIn(self.shared_image['id'], image_ids)
# execute list_customers with the rest of the filters
for list_filter in [no_filter, region_filter,
customer_filter]:
_, body = self.client.list_images(list_filter)
images = [image['id'] for image in body['images']]
self.assertIn(self.shared_image['id'], images)
@decorators.idempotent_id('4435fef4-49a9-435b-8463-cf8a1e0b7cd8')
def test_disable_image(self):
# disable self.public_image and check if request is successful
self.client.enabled_image(self.public_image['id'], False)
self._wait_for_image_status_on_dcp(self.public_image['id'], 'Success')
_, body = self.client.get_image(self.public_image['id'])
image = body["image"]
# assert that the image["enabled"] value is 'False'
self.assertTrue(not image['enabled'])
@decorators.idempotent_id('f32a13e3-6f38-423b-a616-09c8d4e1c277')
def test_enable_image(self):
# enable self.shared_image and check if request is successful
self.client.enabled_image(self.shared_image['id'], True)
self._wait_for_image_status_on_dcp(self.shared_image['id'], 'Success')
_, body = self.client.get_image(self.shared_image['id'])
image = body["image"]
# assert that the image["enabled"] value is 'True'
self.assertTrue(image['enabled'])
@decorators.idempotent_id('cb9e3022-00d7-4a21-bdb2-67d3cd15a4f8')
def test_add_delete_image_region(self):
# skip region assignment in data setup
post_body = self._get_image_params(set_region=False)
image = self._data_setup(post_body)
# add region to image then check to confirm image status = "Success"
self.client.add_region_to_image(image['id'], self.region_id)
# image status must show 'Success' when assigned to a region
self._wait_for_image_status_on_dcp(image['id'], 'Success')
# check that region is successfully added
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertEqual(image["regions"][0]["name"], self.region_id)
# delete the region then check to confirm image status = "no regions"
_, body = self.client.delete_region_from_image(image['id'],
self.region_id)
self._wait_for_image_status_on_dcp(image['id'], 'no regions')
# image region array should be empty after the region was removed
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertFalse(image["regions"])
@decorators.idempotent_id('0ee68189-66a8-4213-ad68-bc12991c174a')
def test_add_delete_image_tenant(self):
# add alt tenant to self.shared_image & check if status = "Success"
self.client.add_customer_to_image(self.shared_image['id'],
self.alt_tenant_id)
self._wait_for_image_status_on_dcp(self.shared_image['id'],
'Success')
# check that alt tenant successfully added to image tenants list
_, body = self.client.get_image(self.shared_image['id'])
image = body["image"]
self.assertEqual(len(image["customers"]), 2)
self.assertIn(self.alt_tenant_id, image['customers'])
# now delete alt_tenant_id and ensure operation is successful
_, body = self.client.delete_customer_from_image(
self.shared_image['id'],
self.alt_tenant_id)
self._wait_for_image_status_on_dcp(self.shared_image['id'], 'Success')
# image region array should no longer contain alt tenant
_, body = self.client.get_image(self.shared_image['id'])
image = body["image"]
self.assertNotIn(self.alt_tenant_id, image['customers'])
@decorators.idempotent_id('bac99348-6b13-4b30-958b-3c039b27eda3')
def test_update_image_tenant(self):
# replace current tenant in self.shared_image with alt tenant
self.client.update_customer(self.shared_image['id'],
self.alt_tenant_id)
self._wait_for_image_status_on_dcp(self.shared_image['id'], 'Success')
# check that image tenants array contains only alt tenant
_, body = self.client.get_image(self.shared_image['id'])
image = body["image"]
self.assertEqual(len(image["customers"]), 1)
self.assertIn(self.alt_tenant_id, image['customers'])
@decorators.idempotent_id('0331e02a-ab52-4341-b676-a02462244277')
def test_create_shared_image(self):
post_body = self._get_image_params()
self.addCleanup(
self._del_img_validate_deletion_on_dcp_and_lcp,
post_body['id'])
# call client create_image and wait till status equals 'Success'
_, body = self.client.create_image(**post_body)
image = body["image"]
self._wait_for_image_status_on_dcp(image['id'], 'Success')
# verify image record created successfully
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertEqual(image["regions"][0]["name"], CONF.identity.region)
self.assertEqual(image["visibility"], post_body['visibility'])
@decorators.idempotent_id('28fb9788-b3a2-4ea1-a3a2-fac63018c05b')
def test_create_private_image(self):
post_body = self._get_image_params(set_shared=False)
post_body["visibility"] = "private"
self.addCleanup(
self._del_img_validate_deletion_on_dcp_and_lcp,
post_body['id'])
# call client create_image and wait till status equals 'Success'
_, body = self.client.create_image(**post_body)
image = body["image"]
self._wait_for_image_status_on_dcp(image['id'], 'Success')
# verify image record created successfully
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertEqual(image["regions"][0]["name"], CONF.identity.region)
self.assertEqual(image["visibility"], post_body['visibility'])
@decorators.idempotent_id('01806310-2a14-4ebb-9679-cce5ee7401b0')
def test_create_private_image_with_customer(self):
post_body = self._get_image_params()
post_body["visibility"] = "private"
# private image can not have associated customers list
self.assertRaises(exceptions.BadRequest,
self.client.create_image, **post_body)
@decorators.idempotent_id('06defa6d-3ce9-4428-8b36-09a351dba872')
def test_create_community_image(self):
post_body = self._get_image_params(set_shared=False)
post_body["visibility"] = "community"
self.addCleanup(
self._del_img_validate_deletion_on_dcp_and_lcp,
post_body['id'])
# call client create_image and wait till status equals 'Success'
_, body = self.client.create_image(**post_body)
image = body["image"]
self._wait_for_image_status_on_dcp(image['id'], 'Success')
# verify image record created successfully
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertEqual(image["regions"][0]["name"], CONF.identity.region)
self.assertEqual(image["visibility"], post_body['visibility'])
@decorators.idempotent_id('298976ae-a6d6-4cdd-b679-786a29eefe2f')
def test_create_community_image_with_customer(self):
post_body = self._get_image_params()
post_body["visibility"] = "community"
# private image can not have associated customers list
self.assertRaises(exceptions.BadRequest,
self.client.create_image, **post_body)
@decorators.idempotent_id('01160918-e217-401d-a6a0-e7992ab76e41')
def test_update_image(self):
region = {}
# define data used to update image
post_body = self._get_image_params(set_region=False)
# create image that will be updated
image = self._data_setup(post_body)
# setup region, change properties for update data
region["name"] = self.region_id
region["type"] = "single"
region["checksum"] = "7297321c2fa6424417a548c85edd6e98"
region["virtual_size"] = "None"
region["size"] = "38797312"
post_body["regions"] = [region]
post_body["enabled"] = False
post_body["customers"] = [self.alt_tenant_id]
# empty tags list
post_body["tags"] = []
_, body = self.client.update_image(image['id'], para=None,
**post_body)
self._wait_for_image_status_on_dcp(image['id'], 'Success')
# verify image record updated successfully
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertEqual(image["regions"][0]["name"], CONF.identity.region)
self.assertIn(self.alt_tenant_id, image['customers'])
self.assertFalse(image['enabled'])
self.assertFalse(image['tags'])
@decorators.idempotent_id('23e2e7e2-5b19-4c66-b35c-7c686a986627')
def test_delete_image(self):
# setup data for test case
post_body = self._get_image_params()
image = self._create_img_and_validate_creation_on_dcp_and_lcp(
**post_body)
# delete the data and do get_image to ensure 404-NotFound response
self._del_img_validate_deletion_on_dcp_and_lcp(image['id'])
self.assertRaises(exceptions.NotFound, self.client.get_image,
image['id'])
@base.SYNC('alt_region', external=True)
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('e642fa39-1b69-4d17-8bd1-aee90ea042a3')
def test_image_while_region_down(self):
# update region with status in 'down' state
self._update_region(CONF.ranger.alt_region,
status={'status': 'down'})
# create image within that newly updated region in 'down' state
post_body = self._get_image_params()
post_body['regions'][0]['name'] = CONF.ranger.alt_region
self.assertRaises(exceptions.BadRequest,
self.client.create_image, **post_body)
# restore region with status in 'functional' state
self._update_region(CONF.ranger.alt_region)
@SYNC('region')
@decorators.idempotent_id('a1fee342-3000-41a6-97f9-b33fd2734e4d')
def test_image_while_region_building(self):
self._update_region(CONF.identity.region,
status={'status': 'building'})
# create image within that newly created region
post_body = self._get_image_params()
post_body['regions'][0]['name'] = CONF.identity.region
# add image to tempest cleanup
self.addCleanup(
self._del_img_validate_deletion_on_dcp_and_lcp,
post_body["id"])
_, body = self.client.create_image(**post_body)
image = body['image']
self._wait_for_image_status_on_dcp(image['id'], 'Success')
self.assertIn('id', body['image'])
_, body = self.client.get_image(image['id'])
self.assertEqual(body['image']['id'], image['id'])
self.assertEqual(body['image']['regions'][0]['status'], 'Success')
self._update_region(CONF.identity.region)
@SYNC('region')
@decorators.idempotent_id('b967ce58-5d24-4af2-8416-a336772c8087')
def test_image_while_region_maintenance(self):
self._update_region(CONF.identity.region,
status={'status': 'maintenance'})
# get image params and add region to them
post_body = self._get_image_params()
post_body['regions'][0]['name'] = CONF.identity.region
# add image to tempest cleanup
self.addCleanup(
self._del_img_validate_deletion_on_dcp_and_lcp,
post_body['id'])
_, body = self.client.create_image(**post_body)
image = body['image']
self._wait_for_image_status_on_dcp(image['id'], 'Success')
self.assertIn('id', body['image'])
_, body = self.client.get_image(image['id'])
self.assertEqual(body['image']['id'], image['id'])
self.assertEqual(body['image']['regions'][0]['status'], 'Success')
self._update_region(CONF.identity.region)
@decorators.idempotent_id('eae7ca20-5383-4579-9f73-0138b8b3ec85')
def test_list_public_images(self):
"""List images with visibility = 'public'"""
# set_shared = False to create image with visibility = 'public'
post_body = self._get_image_params(set_shared=False)
image = self._data_setup(post_body)
test_image_id = image['id']
# confirm image visibility is set to "public" after image is created
self.assertEqual(image["visibility"], "public")
filter_public_images = "?visibility=%s" % image["visibility"]
# list all public images and check if test_image_id is in the list
_, body = self.client.list_images(filter_public_images)
image_ids = [img['id'] for img in body['images']]
self.assertIn(test_image_id, image_ids)
@decorators.idempotent_id('dc321d60-f3bd-477c-b7bf-1594626f0a12')
def test_list_shared_images(self):
"""List images with visibility = 'shared' """
# image data created with visibility = shared set by default
post_body = self._get_image_params()
image = self._data_setup(post_body)
# confirm image visibility is set to "shared" after image is created
self.assertEqual(image["visibility"], "shared")
filter_shared_images = "?visibility=%s" % image["visibility"]
# list all shared images and check if image id is in the list
_, body = self.client.list_images(filter_shared_images)
image_ids = [img['id'] for img in body['images']]
self.assertIn(image['id'], image_ids)
@decorators.idempotent_id('59887b26-8e73-4781-87a4-3b505ece0021')
def test_create_update_delete_image_protected_true(self):
post_body = self._get_image_params()
# set protected True
post_body['protected'] = True
# call client create_image and wait till status equals 'Success'
_, body = self.client.create_image(**post_body)
image = body["image"]
self._wait_for_image_status_on_dcp(image['id'], 'Success')
# verify image record created successfully
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertEqual(image["regions"][0]["name"], CONF.identity.region)
self.assertEqual(image["protected"], True)
# unset protected to False for delete
post_body['protected'] = False
_, body = self.client.update_image(post_body['id'],
para=None,
**post_body)
self._wait_for_image_status_on_dcp(image['id'], 'Success')
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertEqual(image["protected"], False)
# delete image
self._del_img_validate_deletion_on_dcp_and_lcp(post_body['id'])
@decorators.idempotent_id('56cd1de0-3908-41d5-af98-45ad95463817')
def test_create_image_with_tags_properties(self):
post_body = self._get_image_params()
# set tags and properties
tags = ["brocade", "mediumImage", "vCEImage", "vyatta"]
properties = {
"OS": "Debian",
"OS-Version": "7",
}
post_body["tags"] = tags
post_body["properties"] = properties
# add image to tempest cleanup
self.addCleanup(
self._del_img_validate_deletion_on_dcp_and_lcp,
post_body['id'])
# call client create_IMAGE and wait till status equals 'Success'
_, body = self.client.create_image(**post_body)
image = body["image"]
self._wait_for_image_status_on_dcp(image['id'], 'Success')
# verify image record created successfully
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertListEqual(image["tags"], tags)
self.assertDictEqual(image["properties"], properties)
@decorators.idempotent_id('67aa7014-4dbb-4d66-bc7b-1a95a57494f8')
def test_create_image_with_uuid(self):
post_body = self._get_image_params()
# add image to tempest cleanup
self.addCleanup(
self._del_img_validate_deletion_on_dcp_and_lcp,
post_body['id'])
# call client create_IMAGE and wait till status equals 'Success'
_, body = self.client.create_image(**post_body)
image = body["image"]
self._wait_for_image_status_on_dcp(image['id'], 'Success')
# verify image record created successfully
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertEqual(image['id'], post_body['id'])
@base.SYNC('alt_region', external=True)
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('7ae78584-55af-45cb-a4a2-b28cc679699e')
def test_create_image_with_region_group(self):
# create region group
group_id = self._create_region_group()
post_body = self._get_image_params()
post_body['regions'] = [
{
'name': group_id,
'type': 'group'
}
]
# call client create_image and wait till status equals 'Success'
_, body = self.client.create_image(**post_body)
image = body["image"]
self.addCleanup(
self._del_img_validate_deletion_on_dcp_and_lcp, image['id'])
self._wait_for_image_status_on_dcp(image['id'], 'Success')
# verify image record created successfully
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertEqual(image['status'], 'Success')
self.assertEqual(len(image['regions']), 2)
@base.SYNC('alt_region', external=True)
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('73c97918-2081-4f42-9c1e-6fd7a9fb8735')
def test_create_image_with_two_regions(self):
post_body = self._get_image_params()
region1 = {'name': CONF.identity.region,
'type': 'single'}
region2 = {'name': CONF.ranger.alt_region,
'type': 'single'}
post_body['regions'] = [region1, region2]
# call client create_image and wait till status equals 'Success'
_, body = self.client.create_image(**post_body)
image = body["image"]
self.addCleanup(
self._del_img_validate_deletion_on_dcp_and_lcp, image['id'])
self._wait_for_image_status_on_dcp(image['id'], 'Success')
# verify image record created successfully
_, body = self.client.get_image(image['id'])
image = body["image"]
self.assertEqual(image['status'], 'Success')
self.assertEqual(len(image['regions']), 2)