8434876e2d
update sync to sync from base class for alt_region tests, update flavor delete tag test to ensure tag exists before delete Change-Id: Iea42bc1566203a9959fe74a0ab99fb70f1360119
532 lines
22 KiB
Python
532 lines
22 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):
|
|
# setup public image for tempest testing
|
|
cls.image_params_public = \
|
|
cls._get_image_params(set_private=False)
|
|
cls.image_params_private = \
|
|
cls._get_image_params(set_enabled=False)
|
|
|
|
cls.addClassResourceCleanup(
|
|
cls._del_img_validate_deletion_on_dcp_and_lcp,
|
|
cls.image_params_public['id'])
|
|
cls.addClassResourceCleanup(
|
|
cls._del_img_validate_deletion_on_dcp_and_lcp,
|
|
cls.image_params_private['id'])
|
|
|
|
# setup public image for tempest testing
|
|
cls.public_image = \
|
|
cls._create_img_and_validate_creation_on_dcp_and_lcp(
|
|
**cls.image_params_public)
|
|
|
|
# setup private image for tempest testing
|
|
cls.private_image = \
|
|
cls._create_img_and_validate_creation_on_dcp_and_lcp(
|
|
**cls.image_params_private)
|
|
|
|
super(TestTempestIms, cls).resource_setup()
|
|
|
|
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):
|
|
self.addCleanup(
|
|
self._del_img_validate_deletion_on_dcp_and_lcp,
|
|
post_body['id'])
|
|
image = self._create_img_and_validate_creation_on_dcp_and_lcp(
|
|
**post_body)
|
|
# 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/private)
|
|
"""
|
|
|
|
# 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"
|
|
private_filter = "?visibility=private"
|
|
|
|
# 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 private images
|
|
_, body = self.client.list_images(private_filter)
|
|
image_ids = [img['id'] for img in body['images']]
|
|
self.assertIn(self.private_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.private_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.private_image and check if request is successful
|
|
self.client.enabled_image(self.private_image['id'], True)
|
|
self._wait_for_image_status_on_dcp(self.private_image['id'], 'Success')
|
|
_, body = self.client.get_image(self.private_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.private_image & check if status = "Success"
|
|
self.client.add_customer_to_image(self.private_image['id'],
|
|
self.alt_tenant_id)
|
|
self._wait_for_image_status_on_dcp(self.private_image['id'],
|
|
'Success')
|
|
|
|
# check that alt tenant successfully added to image tenants array
|
|
_, body = self.client.get_image(self.private_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.private_image['id'],
|
|
self.alt_tenant_id)
|
|
self._wait_for_image_status_on_dcp(self.private_image['id'], 'Success')
|
|
|
|
# image region array should no longer contain alt tenant
|
|
_, body = self.client.get_image(self.private_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.private_image with alt tenant
|
|
self.client.update_customer(self.private_image['id'],
|
|
self.alt_tenant_id)
|
|
self._wait_for_image_status_on_dcp(self.private_image['id'], 'Success')
|
|
|
|
# check that image tenants array contains only alt tenant
|
|
_, body = self.client.get_image(self.private_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_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)
|
|
|
|
@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_private = False to create image with visibility = 'public'
|
|
post_body = self._get_image_params(set_private=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_private_images(self):
|
|
"""List images with visibility = 'private' """
|
|
# image data created with visibility = private set by default
|
|
post_body = self._get_image_params()
|
|
image = self._data_setup(post_body)
|
|
# confirm image visibility is set to "private" after image is created
|
|
self.assertEqual(image["visibility"], "private")
|
|
filter_private_images = "?visibility=%s" % image["visibility"]
|
|
|
|
# list all public images and check if image id is in the list
|
|
_, body = self.client.list_images(filter_private_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)
|