# 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.
import json
import time
from oslo_concurrency import lockutils
from ranger_tempest_plugin import data_utils as orm_data_utils
from ranger_tempest_plugin.tests.api import fms_base
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions
from tempest.lib.common.utils import data_utils
import testtools
CONF = config.CONF
PREFIX = 'flavor'
SYNC = lockutils.synchronized_with_prefix(PREFIX)
class TestTempestFms(fms_base.FmsBaseOrmTest):
region = CONF.identity.region
alt_region = CONF.ranger.alt_region
@classmethod
def setup_clients(cls):
super(TestTempestFms, cls).setup_clients()
cls.rms_client = cls.os_primary.rms_client
cls.cms_client = cls.os_primary.cms_client
@classmethod
def resource_setup(cls):
# create flavor for use in test cases
body = cls._get_flavor_params()
cls.addClassResourceCleanup(
cls._del_flv_and_validate_deletion_on_dcp_and_lcp,
body['id'])
cls.flavor = cls._create_flv_and_validate_creation_on_dcp_and_lcp(
**body)
cls.tenant_id = cls._create_customer()
cls.alt_tenant_id = cls._create_customer()
# add custom extra specs
cls.custom_es = {'os_extra_specs': {'g': 'guava', 'h': 'honeydew'}}
cls.client.add_extra_specs(cls.flavor['id'], cls.custom_es)
cls._wait_for_flavor_status_on_dcp(cls.flavor['id'], 'Success')
cls.tags = cls.client.get_tags(cls.flavor['id'])
# build customers for related flavor tests
super(TestTempestFms, cls).resource_setup()
@classmethod
def _create_customer(cls):
cust = {}
cust['description'] = ''
cust['enabled'] = True
cust['name'] = data_utils.rand_name('ormTempestFms')
cust['regions'] = [{'name': CONF.identity.region, 'type': 'single'}]
cust['defaultQuotas'] = []
cust['users'] = []
_, body = cls.cms_client.create_customer(**cust)
customer_id = body["customer"]["id"]
_, customer = cls.cms_client.get_customer(customer_id)
if customer["name"] == cust["name"]:
if customer["regions"] == []:
customer_status = "no regions"
else:
customer_status = "Success"
else:
message = "customer %s not created successfully" % cust["name"]
exceptions.TempestException(message)
customer_status = cls.cms_client.get_customer(customer_id)[1]["status"]
start = int(time.time())
while customer_status != 'Success':
time.sleep(cls.build_interval)
customer_status = \
cls.cms_client.get_customer(customer_id)[1]["status"]
if customer_status == 'Error':
message = ('customer %s failed to reach %s status'
' and is in ERROR status on orm' %
(customer_id, customer_status))
raise exceptions.TempestException(message)
if int(time.time()) - start >= cls.build_timeout:
message = ('customer %s failed to reach %s'
'status within the required time (%s s)'
'on orm and is in %s status.'
% (customer_id, customer_status,
cls.build_timeout,
customer_status))
return customer_id
def _get_flavor_details(self, flavor_id):
_, body = self.client.get_flavor(flavor_id)
return body["flavor"]
def _create_region_group(self):
_, body = self.rms_client.create_region_group(
**orm_data_utils.rand_region_group([self.region,
self.alt_region])
)
group_id = body['group']['id']
self.addCleanup(
self.rms_client.delete_region_group,
group_id)
return group_id
def _data_setup(self, post_body):
self.addCleanup(
self._del_flv_and_validate_deletion_on_dcp_and_lcp,
post_body['id'])
flavor = self._create_flv_and_validate_creation_on_dcp_and_lcp(
**post_body)
return flavor
def _exec_tags_function(self, flavor_id, req_json, action, para):
if action == 'add':
_, body = self.client.add_tags(flavor_id, req_json)
elif action == 'update':
_, body = self.client.update_tags(flavor_id, req_json)
elif action == 'delete':
_, body = self.client.delete_tags(flavor_id, para)
self._wait_for_flavor_status_on_dcp(flavor_id, 'Success')
def _exec_ex_spec_function(self, flavor_id, es_body, action, para):
if action == 'add':
_, body = self.client.add_extra_specs(flavor_id, es_body)
elif action == 'update':
_, body = self.client.update_extra_specs(flavor_id, es_body)
elif action == 'delete':
_, body = self.client.delete_extra_specs(flavor_id, para)
self._wait_for_flavor_status_on_dcp(flavor_id, 'Success')
def _wait_for_flavor_region_status(self, flavor, region_name):
region_index = 0
start = int(time.time())
for regions in flavor['regions']:
if regions['name'] == region_name:
region_status = regions['status']
while region_status != 'Success':
time.sleep(self.build_interval)
region_status = \
self.fms_client.get_flavor(flavor['id'])
region_status = \
region_status['regions'][region_index]['status']
if region_status == 'Error':
message = ('Flavor region %s failed to'
' reach success status'
' and is in ERROR status on orm' %
region_name)
raise exceptions.TempestException(message)
if int(time.time()) - start >= self.build_timeout:
message = ('Flavor %s region %s failed to reach'
' success status within the required'
' time (%s s) on orm and is in %s status.'
% (flavor['id'], region_name,
self.build_timeout,
region_status))
region_index += 1
def _restore_custom_es(self, flavor_id):
_, body = self.client.update_extra_specs(flavor_id, self.custom_es)
self._wait_for_flavor_status_on_dcp(flavor_id, 'Success')
_, extra_specs = self.client.get_extra_specs(flavor_id)
es_expected = self.custom_es.get("os_extra_specs")
es_expected.update(self.flavor['extra-specs'])
self.assertDictEqual(es_expected, extra_specs.get("os_extra_specs"))
def _assert_create_flavor(self, public=False):
post_body = self._get_flavor_params(public=public)
visibility = "private" if not public else "public"
# call client create_flavor and wait till status equals 'Success'
body = self._data_setup(post_body)
test_flavor_id = body['id']
self._wait_for_flavor_status_on_dcp(test_flavor_id, 'Success')
# verify flavor record created successfully
flavor = self._get_flavor_details(test_flavor_id)
self.assertEqual(flavor["visibility"], visibility)
self.assertEqual(flavor["regions"][0]["name"], CONF.identity.region)
self.assertEqual(flavor["status"], "Success")
def _update_region(self, r_name, status=None):
if status is None:
status = {'status': 'functional'}
self.os_admin.rms_client.update_region_status(
r_name, status=status)
def _delete_flavor(self, flavor_id):
# delete the data and get_flavor to ensure 404-NotFound response
self._del_flv_and_validate_deletion_on_dcp_and_lcp(flavor_id)
self.assertRaises(exceptions.NotFound,
self.client.get_flavor,
flavor_id)
@SYNC('flavor')
@decorators.idempotent_id('2a4481cd-acce-4a5d-af7c-940222a6238b')
def test_get_flavor(self):
"""Execute get_flavor using flavor_id / flavor_name"""
for para in [self.flavor['id'], self.flavor['name']]:
_, body = self.client.get_flavor(para)
self.assertIn(self.flavor['id'], body["flavor"]["id"])
@decorators.idempotent_id('c46a503a-951c-4d00-afaa-46076b54db16')
def test_list_flavor_with_filters(self):
"""This function executes 'list flavors' with the following filters:
- None (no filters, i.e. list all flavors)
- alias filter
- region filter
- visibility filter
- 'contains' filter
- 'starts_with' filter
- 'series' filter
- "tenant" filter
"""
post_body = self._get_flavor_params()
flavor = self._data_setup(post_body)
# for use by the 'constains and 'starts_with' filter
str_index1 = data_utils.rand_int_id(0, len(flavor['name']) - 1)
str_index2 = data_utils.rand_int_id(str_index1 + 1,
len(flavor['name']))
# define the list flavors filters to be used for this test
alias_filter = "?alias=%s" % flavor['alias']
region_filter = "?region=%s" % CONF.identity.region
visibility_filter = "?visibility=%s" % flavor['visibility']
contains_filter = '?contains=%s' \
% flavor['name'][str_index1:str_index2]
startswith_filter = "?starts_with=%s" % flavor['name'][:str_index2]
series_filter = "?series=%s" % flavor['series']
tenant_filter = "?tenant=%s" % self.tenant_id
for list_filter in [None, region_filter, visibility_filter,
alias_filter, contains_filter, startswith_filter,
series_filter, tenant_filter]:
_, body = self.client.list_flavors(list_filter)
flavor_ids = [test_flavor["id"] for test_flavor in body["flavors"]]
self.assertIn(flavor['id'], flavor_ids)
@decorators.idempotent_id('7b9d1f91-a8a4-458d-aaad-a98b5bf033b4')
def test_create_private_flavor(self):
self._assert_create_flavor(public=False)
@decorators.idempotent_id('790d3417-54a1-4db1-aee9-7f3a7d1e0962')
def test_create_public_flavor(self):
self._assert_create_flavor(public=True)
@SYNC('flavor')
@decorators.idempotent_id('4cad10ce-67d2-4633-b347-2c16783a31b9')
def test_add_flavor_tags(self):
# test add_tags command with two sets of key:values
add_tag_body = {"tags": {"aa": "bb", "cc": "dd"}}
self._exec_tags_function(self.flavor['id'], None, 'delete', None)
_, tag_body = self.client.get_tags(self.flavor['id'])
self.assertDictEqual(tag_body, {"tags": {}})
self._exec_tags_function(self.flavor['id'], add_tag_body, 'add', None)
_, tag_body = self.client.get_tags(self.flavor['id'])
subset = {k: v for k, v in list(tag_body.get("tags").items())
if k in add_tag_body.get("tags")}
self.assertDictEqual(add_tag_body.get("tags"), subset)
@SYNC('flavor')
@decorators.idempotent_id('db8e5c0f-0041-45d4-9939-e079296123d8')
def test_replace_flavor_tags(self):
_, tag_body = self.client.get_tags(self.flavor['id'])
# test replace_tags command
replace_tag_body = {"tags": {"ee": "ff", "gg": "hh"}}
self._exec_tags_function(self.flavor['id'], replace_tag_body,
'update', None)
_, tag_body = self.client.get_tags(self.flavor['id'])
self.assertDictEqual(replace_tag_body.get("tags"),
tag_body.get("tags"))
# restore tags
self._exec_tags_function(self.flavor['id'], tag_body, 'update', None)
@SYNC('flavor')
@decorators.idempotent_id('e0a0eca6-e120-45ab-a1a4-f5b95fdf97f1')
def test_delete_flavor_tag(self):
# test delete_tag command - delete a tag from tags body
delete_tag_key = 'a'
self._exec_tags_function(self.flavor['id'], None, 'delete',
delete_tag_key)
# do get_tag and tries to retrieve the deleted tag
_, tag_body = self.client.get_tags(self.flavor['id'])
tag_present = tag_body.get("tags").get(delete_tag_key, None)
self.assertIsNone(tag_present)
# restore tags
self._exec_tags_function(self.flavor['id'],
{'tags': {'aa': 'bb'}},
'add',
None)
@SYNC('flavor')
@decorators.idempotent_id('9c511020-53ed-4139-8c53-451cb0ea8c75')
def test_delete_all_flavor_tags(self):
# ensure there is at least a tag
_, tag_body = self.client.get_tags(self.flavor['id'])
restore_tags = tag_body
self.assertTrue(bool(tag_body.get("tags")))
# test delete_all_tags command - run get_tag again and confirm
# that the tag dict is now empty
self._exec_tags_function(self.flavor['id'], None, 'delete', None)
_, tag_body = self.client.get_tags(self.flavor['id'])
# assert that tag_body contains nothing
self.assertFalse(tag_body["tags"])
# restore deleted tags
self._exec_tags_function(self.flavor['id'], restore_tags, 'add', None)
@fms_base.base.SYNC('alt_region', external=True)
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('ec74d68f-b42a-41a8-9685-ff5eca25ea0c')
def test_add_delete_flavor_region(self):
# setup data to add region
post_body = self._get_flavor_params()
flavor = self._data_setup(post_body)
test_flavor_id = flavor['id']
post_region_body = \
'{"regions": [{"name": "%s"}]}' % (CONF.ranger.alt_region)
post_region_body = json.loads(post_region_body)
_, body = self.client.add_flavor_regions(test_flavor_id,
post_region_body)
self._wait_for_flavor_status_on_dcp(test_flavor_id, 'Success')
_, body = self.client.get_flavor(test_flavor_id)
self._wait_for_flavor_region_status(flavor, CONF.ranger.alt_region)
assert self.region or self.alt_region in \
body['flavor']['regions'][0]['name']
assert self.region or self.alt_region in \
body['flavor']['regions'][1]['name']
# remove added region and then check to confirm flavor status
_, body = self.client.delete_flavor_region(test_flavor_id,
CONF.ranger.alt_region)
self._wait_for_flavor_status_on_dcp(test_flavor_id, 'Success')
_, body = self.client.get_flavor(test_flavor_id)
self.assertTrue(len(body["flavor"]["regions"]) == 1)
@SYNC('flavor')
@decorators.idempotent_id('71404409-5d95-472c-8dac-b49a1c0c4b37')
def test_add_flavor_extra_specs(self):
# get extra specs before the add
_, es_body = self.client.get_extra_specs(self.flavor['id'])
es_expected = es_body.get("os_extra_specs")
# add custom extra specs
add_es_body = {"os_extra_specs": {"a": "apple", "b": "banana"}}
self._exec_ex_spec_function(self.flavor['id'],
add_es_body,
'add', None)
_, extra_specs = self.client.get_extra_specs(self.flavor['id'])
# assert extra specs add results match expected
es_expected.update(add_es_body["os_extra_specs"])
self.assertDictEqual(es_expected, extra_specs.get("os_extra_specs"))
@SYNC('flavor')
@decorators.idempotent_id('043948fd-125b-4d96-bf40-42464066a7e1')
def test_update_flavor_extra_specs(self):
# add custom extra spec using update_extra_spec
replace_es_body = {"os_extra_specs": {"z": "zebra", "x": "xray"}}
self._exec_ex_spec_function(
self.flavor['id'],
replace_es_body,
'update',
None)
# assert extra specs update results match expected
_, flavor_ex_specs = self.client.get_extra_specs(self.flavor['id'])
es_expected = replace_es_body.get("os_extra_specs")
es_expected.update(self.flavor['extra-specs'])
self.assertDictEqual(es_expected,
flavor_ex_specs.get("os_extra_specs"))
@SYNC('flavor')
@decorators.idempotent_id('df83e2cd-d202-4b2f-8459-391a73067ec5')
def test_create_and_delete_flavor_extra_spec(self):
# get extra specs before the delete
_, es_body = self.client.get_extra_specs(self.flavor['id'])
# now delete one of the custom extra specs
delete_es_key_h = "h"
self._exec_ex_spec_function(self.flavor['id'], None, 'delete',
delete_es_key_h)
# assert extra specs update results match expected
es_body["os_extra_specs"].pop(delete_es_key_h)
_, flavor_ex_specs = self.client.get_extra_specs(self.flavor['id'])
self.assertDictEqual(es_body["os_extra_specs"],
flavor_ex_specs.get("os_extra_specs"))
@SYNC('flavor')
@decorators.idempotent_id('e3fc7ce3-c8fe-4805-8ad3-7be2c94fe7ad')
def test_delete_all_flavor_extra_specs(self):
# run delete ALL extra specs - note that this will only
# delete custom extra specs, NOT the default extra specs
_, specs_to_restore = self.client.get_extra_specs(self.flavor['id'])
self._exec_ex_spec_function(self.flavor['id'], None, 'delete', None)
_, flavor_ex_specs = self.client.get_extra_specs(self.flavor['id'])
# assert that flavor extra specs now contains only
# the default extra specs
self.assertDictEqual(self.flavor['extra-specs'],
flavor_ex_specs.get("os_extra_specs"))
self._exec_ex_spec_function(self.flavor['id'],
specs_to_restore, 'add', None)
@decorators.idempotent_id('187195b5-adfb-4c73-a2f5-42117021f5f2')
def test_add_delete_flavor_tenant(self):
# setup data for test case
post_body = self._get_flavor_params()
flavor = self._data_setup(post_body)
test_flavor_id = flavor['id']
# check that flavor contains one tenant before testing adding tenant
flavor = self._get_flavor_details(test_flavor_id)
self.assertEqual(len(flavor["tenants"]), 1)
self.assertEqual(flavor["tenants"][0], self.tenant_id)
# test add_flavor_tenant by adding one more tenant
alt_tenant_body = '{"tenants": ["%s"]}' % (self.alt_tenant_id)
alt_tenant_body = json.loads(alt_tenant_body)
_, body = self.client.add_flavor_tenants(test_flavor_id,
alt_tenant_body)
self._wait_for_flavor_status_on_dcp(test_flavor_id, 'Success')
# get flavor on same flavor id and confirm we have two tenants now
flavor = self._get_flavor_details(test_flavor_id)
self.assertEqual(len(flavor["tenants"]), 2)
self.assertIn(self.alt_tenant_id, flavor["tenants"])
# delete one tenant, then wait until status = 'Success'
_, body = self.client.delete_flavor_tenant(test_flavor_id,
self.alt_tenant_id)
self._wait_for_flavor_status_on_dcp(test_flavor_id, 'Success')
# get flavor to confirm flavor["tenants"] now shows one tenant only
flavor = self._get_flavor_details(test_flavor_id)
self.assertEqual(len(flavor["tenants"]), 1)
self.assertEqual(flavor["tenants"][0], self.tenant_id)
@decorators.idempotent_id('36958bba-673e-4397-85a9-fddb01e9ca0d')
def test_delete_flavor(self):
# setup data for test case
post_body = self._get_flavor_params()
flavor = self._create_flv_and_validate_creation_on_dcp_and_lcp(
**post_body)
# delete the data and do get_flavor to ensure 404-NotFound response
self._del_flv_and_validate_deletion_on_dcp_and_lcp(flavor['id'])
self.assertRaises(exceptions.NotFound, self.client.get_flavor,
flavor['id'])
@decorators.idempotent_id('2a1240d8-ae30-4c37-b99f-af961a5e16cb')
def test_create_flavor_with_swap_and_ephemeral(self):
post_body = self._get_flavor_params()
# set swap and ephemeral size
swap = ephemeral = '1024'
# update swap and ephemeral
post_body['ephemeral'] = ephemeral
post_body['swap'] = swap
flavor = self._data_setup(post_body)
test_flavor_id = flavor['id']
# verify flavor record created successfully
flavor_details = self._get_flavor_details(test_flavor_id)
self.assertEqual(flavor_details['status'], 'Success')
self.assertEqual(flavor_details['swap'], swap)
self.assertEqual(flavor_details['ephemeral'], ephemeral)
@decorators.idempotent_id('2178be75-c36b-4cfe-9df7-8a4bff6e7f83')
def test_list_flavor_with_additional_field(self):
_, body = self.client.list_flavors()
fields = [
'id',
'description',
'series',
'ram',
'vcpus',
'disk',
'swap',
'ephemeral',
'extra-specs',
'options',
'tag',
'regions',
'visibility',
'tenants',
]
if body['flavors']:
flavor = body['flavors'][0]
# check fields is present in response dict as keys
self.assertTrue(set(fields).issubset(set(flavor.keys())))
@decorators.idempotent_id('53275983-6999-42f2-8ce5-6a83ade3980f')
def test_create_flavor_with_zero_disk(self):
post_body = self._get_flavor_params()
disk = '0'
post_body['disk'] = disk
flavor = self._data_setup(post_body)
test_flavor_id = flavor['id']
flavor_details = self._get_flavor_details(test_flavor_id)
self.assertEqual(flavor_details["status"], "Success")
self.assertEqual(flavor_details["disk"], disk)
@fms_base.base.SYNC('alt_region', external=True)
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('997ca03c-4176-4632-a0c9-7e943b03306c')
def test_create_flavor_with_region_group(self):
# create region group
group_id = self._create_region_group()
post_body = self._get_flavor_params()
region_group = {'name': group_id, 'type': 'group'}
post_body['regions'] = [region_group]
flavor = self._data_setup(post_body)
test_flavor_id = flavor['id']
flavor_details = self._get_flavor_details(test_flavor_id)
self.assertEqual(flavor_details['status'], 'Success')
assert self.region or self.alt_region in \
flavor['regions'][0]['name']
assert self.region or self.alt_region in \
flavor['regions'][1]['name']
@fms_base.base.SYNC('alt_region', external=True)
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('ea2a618e-bd53-460b-bde5-01ea20b417c9')
def test_create_flavor_with_two_regions(self):
post_body = self._get_flavor_params()
post_body['regions'].append(
{'name': self.alt_region}
)
flavor = self._data_setup(post_body)
test_flavor_id = flavor['id']
flavor_details = self._get_flavor_details(test_flavor_id)
self.assertEqual(flavor_details['status'], 'Success')
self.assertEqual(len(flavor['regions']), 2)
@fms_base.base.SYNC('alt_region', external=True)
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('06c81b29-85b6-4edf-ab89-3877c49e23bc')
def test_create_flavor_with_group_region(self):
group_id = self._create_region_group()
post_body = self._get_flavor_params()
post_body['regions'] = [
{
'name': group_id,
'type': 'group'
}
]
flavor = self._data_setup(post_body)
flavor_details = self._get_flavor_details(flavor['id'])
self.assertEqual(flavor_details['status'], 'Success')
assert self.alt_region or self.region in flavor['regions'][0]['name']
assert self.alt_region or self.region in flavor['regions'][1]['name']
@decorators.idempotent_id('997ca03c-4176-4632-a0c9-7e943b03306d')
def test_create_flavor_with_uuid(self):
post_body = self._get_flavor_params()
flavor = self._data_setup(post_body)
flavor_details = self._get_flavor_details(post_body['id'])
self.assertEqual(flavor_details['status'], 'Success')
self.assertEqual(flavor['id'], post_body['id'])
@fms_base.base.SYNC('alt_region', external=True)
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('37f1909f-3ba2-403c-ba0c-0a11b869d6a1')
def test_flavor_while_region_down(self):
# update region to status down
self._update_region(self.alt_region, status={'status': 'down'})
# create flavor within that newly created region
post_body = self._get_flavor_params(set_region=False)
post_body['regions'] = [{'name': self.alt_region}]
self.assertRaises(exceptions.BadRequest,
self.client.create_flavor, **post_body)
self._update_region(self.alt_region)
@decorators.idempotent_id('1c6a24d3-345e-46d4-aaa0-127b7fc8a42d')
def test_flavor_while_region_building(self):
# update region to status building
self._update_region(self.region, status={'status': 'building'})
post_body = self._get_flavor_params(set_region=False)
post_body['regions'] = [{'name': self.region}]
body = self._data_setup(post_body)
self.assertIn('id', body)
test_flavor_id = body['id']
flavor = self._get_flavor_details(test_flavor_id)
self.assertEqual(flavor['id'], test_flavor_id)
self.assertEqual(flavor['status'], 'Success')
self._update_region(self.region)
@decorators.idempotent_id('e17dab64-c900-4a19-a7a2-96a0bf4af0f8')
def test_flavor_while_region_maintenance(self):
# update region to status maintenance
self._update_region(self.region, status={'status': 'maintenance'})
post_body = self._get_flavor_params(set_region=False)
post_body['regions'] = [{'name': self.region}]
body = self._data_setup(post_body)
self.assertIn('id', body)
test_flavor_id = body['id']
flavor = self._get_flavor_details(test_flavor_id)
self.assertEqual(flavor['id'], test_flavor_id)
self.assertEqual(flavor['status'], 'Success')
self._update_region(self.region)