3db2d2c746
When Ranger is not reachable, test cases will need to be skipped. In addition, remove the extra skip check for RMS and IMS services. If setup fails for these modules, let the test cases fail as well. Change-Id: Ibadb065ef9816cf1cb6ce44e5229753d679f8bf6
531 lines
22 KiB
Python
531 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):
|
|
super(TestTempestIms, cls).resource_setup()
|
|
# 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)
|
|
|
|
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)
|