Chi Lo fc92e28653 Added new test cases for CMS service
Change-Id: I6f09d000652ad3a4967cc16874e23d2cada70cb8
2020-06-18 15:51:47 +00:00

537 lines
24 KiB
Python
Executable File

# 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 copy
import random
import uuid
from oslo_concurrency import lockutils
from ranger_tempest_plugin import data_utils as orm_data_utils
from ranger_tempest_plugin.tests.api import cms_base
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions
import testtools
CONF = config.CONF
PREFIX = 'ranger'
SYNC = lockutils.synchronized_with_prefix(PREFIX)
class TestTempestCms(cms_base.CmsBaseOrmTest):
def _delete_customer(self, id_, region):
"""Try to delete customer from dcp only"""
self.client.delete_region_from_customer(id_, region)
self._wait_for_status(id_, 'no regions')
self.client.delete_customer(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 _create_customer(self, post_body, cleanup=True):
customer_id = self._create_cust_validate_creation_on_dcp_and_lcp(
**post_body)
if cleanup:
self.addCleanup(
self._del_cust_validate_deletion_on_dcp_and_lcp, customer_id)
return customer_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('6072c438-1e45-4c0b-97a6-e5127bd33d89')
def test_get_customer(self):
"""Execute 'get_customer' using the following options
- get customer by id (using cust_id parameter)
- get customer by name (using cust_name parameter)
"""
# execute get_customer using cust_id and cust_name
for identifier in [self.setup_customer_id,
self.setup_customer['name']]:
_, body = self.client.get_customer(identifier)
self.assertIn(self.setup_customer_id, body['uuid'])
@decorators.idempotent_id('6072c438-1e45-4c0b-97a6-e5127bd33d90')
def test_list_customers_with_filters(self):
"""This function executes 'list customer' with all available filters
- no filter (i.e. list all customers)
- filter by metadata key
- filter by metadata key value pair
- filter by region
- filter by default user id
- customer name contains a substring
- customer name starting with a string
"""
# format filter parameter values
region_name = [
region['name'] for region in self.setup_customer['regions']]
userid = [user['id'] for user in self.setup_customer['users']]
cust_name = self.setup_customer['name']
substr_name = random.randint(0, len(cust_name))
# get the first key from metadata as the metadata key filter
metadata_key = list(self.setup_customer['metadata'].keys())[0]
metadata_value = self.setup_customer['metadata'][metadata_key]
# define the list customer filters to be used for this test
no_filter = None
metadata_filter = {'metadata': metadata_key}
metadata_kv_filter = {
'metadata': '%s:%s' % (metadata_key, metadata_value)
}
region_filter = {'region': region_name[0]}
user_filter = {'user': userid[0]}
contains_filter = {'contains': cust_name[substr_name:]}
startswith_filter = {'starts_with': cust_name[:substr_name]}
# execute list_customers with the available filters
for list_filter in [no_filter, metadata_filter, metadata_kv_filter,
region_filter, user_filter,
contains_filter, startswith_filter]:
_, body = self.client.list_customers(list_filter)
customers = [cust['id'] for cust in body['customers']]
self.assertIn(self.setup_customer_id, customers)
@decorators.idempotent_id('ac132678-fdb6-4037-a310-813313044055')
def test_enable_customer(self):
# setup data for test case
post_body = self._get_customer_params(enabled=False)
test_customer_id = self._create_customer(post_body)
# update enabled status from 'False' to 'True' and validate update OK
_, body = self.client.get_customer(test_customer_id)
self.assertFalse(body['enabled'])
self.client.enable_customer(test_customer_id, True)
self._wait_for_status(test_customer_id, 'Success')
_, body = self.client.get_customer(test_customer_id)
self.assertTrue(body['enabled'])
@decorators.idempotent_id('7dfd5f7e-7031-4ee1-b355-cd5cdeb21bd1')
def test_add_default_user(self):
# setup data for "add_default_user" test case; leave default
# and region users blank at the initial data creation
post_body = self._get_customer_params(default_users=False,
region_users=False)
test_customer_id = self._create_customer(post_body)
_, body = self.client.get_customer(test_customer_id)
self.assertFalse(body['users'])
self.assertFalse(body["regions"][0]['users'])
# now add a default user, then validate that new user is added to
# BOTH default user AND region user successfully
post_default_user = self._get_user_params()
new_user_id = post_default_user[0]["id"]
_, body = self.client.add_default_user(test_customer_id,
*post_default_user)
self._wait_for_status(test_customer_id, 'Success')
_, body = self.client.get_customer(test_customer_id)
self.assertIn(new_user_id, [x['id'] for x in body['users']])
self.assertIn(new_user_id, [x['id']
for x in body['regions'][0]['users']])
@decorators.idempotent_id('699e8487-035e-4ae0-97b4-ca51b9a08aea')
def test_delete_default_user(self):
# setup data for delete_default_user test case
post_body = self._get_customer_params()
default_user_id = post_body["users"][0]["id"]
test_customer_id = self._create_customer(post_body)
_, body = self.client.get_customer(test_customer_id)
self.assertEqual(default_user_id, body['users'][0]['id'])
self.assertIn(default_user_id,
[x['id'] for x in body['regions'][0]['users']])
# delete default user and validate operation success by confirming
# user id no longer in both default users and region users list
_, body = self.client.delete_default_user(test_customer_id,
default_user_id)
self._wait_for_status(test_customer_id, 'Success')
_, body = self.client.get_customer(test_customer_id)
self.assertFalse(body['users'])
self.assertNotIn(default_user_id,
[x['id'] for x in body['regions'][0]['users']])
@decorators.idempotent_id('48ffd49f-2b36-40b4-b1b4-0c805b7ba7c2')
def test_replace_default_user(self):
# setup data for "replace_default_user" test case; no need to
# set region user as default user will also be assigned to it
post_body = self._get_customer_params(region_users=False)
default_user_id = post_body["users"][0]["id"]
test_customer_id = self._create_customer(post_body)
_, body = self.client.get_customer(test_customer_id)
self.assertIn(default_user_id, [x['id'] for x in body['users']])
self.assertEqual(body['users'], body['regions'][0]['users'])
# replace default user
put_default_user = self._get_user_params(alt=True)
updated_user_id = put_default_user[0]["id"]
_, body = self.client.replace_default_user(test_customer_id,
*put_default_user)
self._wait_for_status(test_customer_id, 'Success')
# validate that BOTH the customer default user and region user
# are replaced with the new default_user successfully
_, body = self.client.get_customer(test_customer_id)
self.assertEqual(len(body['users']),
len(body['regions'][0]['users']))
self.assertIn(updated_user_id, [x['id'] for x in body['users']])
self.assertEqual(body['users'], body['regions'][0]['users'])
@decorators.idempotent_id('07a631f9-3aa5-4797-9ead-4531ced89e2a')
def test_add_region_user(self):
# We are leaving both default and region users as blank to ensure
# region user is empty on initial data creation for this test case
post_body = self._get_customer_params(region_users=False,
default_users=False)
test_customer_id = self._create_customer(post_body)
_, body = self.client.get_customer(test_customer_id)
# confirm that the region users body is empty after data creation
self.assertFalse(body["regions"][0]["users"])
# now we shall add user to regions[0]
post_region_user = self._get_user_params()
_, body = self.client.add_region_user(test_customer_id,
CONF.identity.region,
*post_region_user)
self._wait_for_status(test_customer_id, 'Success')
_, customer = self.client.get_customer(test_customer_id)
# validate that the region user is no longer empty after the add
self.assertTrue(customer["regions"][0]["users"])
@decorators.idempotent_id('9b2b3af8-2444-4143-a9e6-78c33b36c823')
def test_delete_region_user(self):
# To test delete_region_users scenario, leave default user blank
# for the test customer data, or else default user info will be
# added as region user as well
post_body = self._get_customer_params(default_users=False)
region_user_id = post_body["regions"][0]["users"][0]["id"]
test_customer_id = self._create_customer(post_body)
_, body = self.client.get_customer(test_customer_id)
self.assertTrue(body["regions"][0]["users"])
# delete the user from the region, then validate operation success
_, body = self.client.delete_region_user(test_customer_id,
CONF.identity.region,
region_user_id)
self._wait_for_status(test_customer_id, 'Success')
_, customer = self.client.get_customer(test_customer_id)
# validate that the region user is now empty
self.assertFalse(customer["regions"][0]["users"])
@decorators.idempotent_id('0ca59977-ef29-46b9-be92-14980a12c573')
def test_replace_region_user(self):
post_body = self._get_customer_params()
test_customer_id = self._create_customer(post_body)
# update region user then confirm that update is successful
put_region_user = self._get_user_params(alt=True)
new_region_user_id = put_region_user[0]["id"]
_, body = self.client.replace_region_user(test_customer_id,
CONF.identity.region,
*put_region_user)
self._wait_for_status(test_customer_id, 'Success')
_, customer = self.client.get_customer(test_customer_id)
self.assertIn(new_region_user_id, [x['id']
for x in customer["regions"][0]['users']])
@decorators.idempotent_id('f1444965-c711-438d-ab86-a2412acbe8e0')
def test_replace_metadata(self):
metadata = {'metadata': {'replace_key': 'replace_value'}}
_, body = self.client.replace_metadata(self.setup_customer_id,
metadata)
self._wait_for_status(self.setup_customer_id, 'Success')
_, body = self.client.list_customers({'metadata': 'replace_key'})
self.assertIn(self.setup_customer_id,
[x['id'] for x in body['customers']])
@decorators.idempotent_id('80713a87-8e95-481f-a198-6b4515d48362')
def test_add_metadata(self):
metadata = {'metadata': {'add_key': 'add_value'}}
_, body = self.client.add_metadata(self.setup_customer_id,
metadata)
self._wait_for_status(self.setup_customer_id, 'Success')
_, body = self.client.list_customers({'metadata': 'add_key'})
self.assertIn(self.setup_customer_id,
[x['id'] for x in body['customers']])
@decorators.idempotent_id('19a6bbe2-92b9-46be-95b7-aa0d9f247d88')
def test_add_regions(self):
region = self._get_region_params()
_, body = self.client.add_regions(self.bare_customer_id,
region)
self._wait_for_status(self.setup_customer_id, 'Success')
_, body = self.client.list_customers({'region': region[0]['name']})
self.assertIn(self.setup_customer_id,
[x['id'] for x in body['customers']])
@decorators.idempotent_id('09a43bc1-439e-4497-a026-f2c3de451d29')
def test_delete_regions(self):
# setup data for delete_region
post_body = self._get_customer_params()
region_name = post_body["regions"][0]["name"]
test_customer_id = self._create_customer(post_body)
_, customer = self.client.get_customer(test_customer_id)
self.assertTrue(customer["regions"])
_, body = self.client.delete_region_from_customer(test_customer_id,
region_name)
self._wait_for_status(test_customer_id, 'no regions')
_, customer = self.client.get_customer(test_customer_id)
self.assertFalse(customer["regions"])
@decorators.idempotent_id('c7a24667-2a99-41ac-a42d-6c1163ef48af')
def test_create_customer(self):
post_body = self._get_customer_params(quota=False)
test_cust_name = post_body['name']
_, body = self.client.create_customer(**post_body)
test_customer_id = body['customer']['id']
self.addCleanup(
self._del_cust_validate_deletion_on_dcp_and_lcp,
test_customer_id)
self._wait_for_status(test_customer_id, 'Success')
_, body = self.client.get_customer(test_cust_name)
self.assertIn(test_customer_id, body['uuid'])
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('c2f4e842-c0c3-4747-9b10-e86c25fe8283')
def test_create_customer_with_region_group(self):
# create region group
group_id = self._create_region_group()
post_body = self._get_bare_customer_params()
post_body['description'] = 'test create customer region group'
post_body['regions'] = [
{
'name': group_id,
'type': 'group'
}
]
test_customer_id = self._create_customer(post_body)
_, body = self.client.get_customer(test_customer_id)
self.assertEqual(body['status'], 'Success')
self.assertEqual(len(body['regions']), 2)
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('5e6f1b6b-bff1-4d30-ba97-4ff66ad47ba9')
def test_create_customer_with_two_regions(self):
post_body = self._get_bare_customer_params()
post_body['description'] = 'test create customer multi regions'
region1 = {'name': CONF.identity.region,
'type': 'single'}
region2 = {'name': CONF.ranger.alt_region,
'type': 'single'}
post_body['regions'] = [region1, region2]
test_customer_id = self._create_customer(post_body)
_, body = self.client.get_customer(test_customer_id)
self.assertEqual(body['status'], 'Success')
self.assertEqual(len(body['regions']), 2)
@SYNC('customer')
@decorators.idempotent_id('43785f87-27d5-408d-997f-de602caeb698')
def test_replace_customer(self):
customer = self._get_bare_customer_params()
customer['name'] = self.setup_customer['name']
customer['description'] = 'test customer'
# Remove users and quotas under regions
customer['regions'] = [{'name': CONF.identity.region}]
_, body = self.client.update_customer(self.setup_customer_id,
customer)
self._wait_for_status(self.setup_customer_id, 'Success')
_, body = self.client.get_customer(self.setup_customer_id)
self.assert_expected(customer, body, ['name', 'regions'])
for region in customer['regions']:
self.assertIn(region['name'], [x['name'] for x in body['regions']])
# verify no users and quotas under regions
for region in body['regions']:
self.assertEqual(len(region['quotas']), 0)
self.assertEqual(len(region['users']), 0)
@decorators.idempotent_id('e8b9077a-d45c-4e24-a433-e7dfa07486b9')
def test_delete_customer(self):
# setup data for test case
post_body = self._get_customer_params()
test_customer_id = self._create_customer(post_body, cleanup=False)
# delete the data and do get_customer to ensure 404-NotFound response
self._del_cust_validate_deletion_on_dcp_and_lcp(test_customer_id)
self.assertRaises(exceptions.NotFound, self.client.get_customer,
test_customer_id)
@decorators.idempotent_id('b8493b3f-e64d-448e-a965-b0eeff415981')
def test_customer_while_region_down(self):
# create region with status down
self._update_region(CONF.identity.region, status={'status': 'down'})
# create customer within that newly created region
cust_body = self._get_customer_params()
cust_body['regions'][0]['name'] = CONF.identity.region
self.assertRaises(exceptions.BadRequest,
self.client.create_customer, **cust_body)
@decorators.idempotent_id('1aa52c36-4b1e-459e-9633-12b6cbd53ae7')
def test_customer_while_region_building(self):
# create region with status maintenance
self._update_region(CONF.identity.region,
status={'status': 'building'})
cust_body = self._get_customer_params()
cust_body['regions'][0]['name'] = CONF.identity.region
resp, body = self.client.create_customer(**cust_body)
customer_id = body['customer']['id']
self._wait_for_status(customer_id, 'Success')
self.addCleanup(
self._delete_customer,
customer_id,
CONF.identity.region)
_, body = self.client.get_customer(customer_id)
self.assertEqual(body['status'], 'Success')
self.assertEqual(body['custId'], customer_id)
self._update_region(CONF.identity.region)
@decorators.idempotent_id('d3cd949e-7895-421c-aeee-2c3d862c391f')
def test_customer_while_region_maintenance(self):
# create region with status maintenance
self._update_region(CONF.identity.region,
status={'status': 'maintenance'})
cust_body = self._get_customer_params()
cust_body['regions'][0]['name'] = CONF.identity.region
resp, body = self.client.create_customer(**cust_body)
customer_id = body['customer']['id']
self._wait_for_status(customer_id, 'Success')
self.addCleanup(
self._delete_customer,
customer_id,
CONF.identity.region)
_, body = self.client.get_customer(customer_id)
self.assertEqual(body['status'], 'Success')
self.assertEqual(body['custId'], customer_id)
self._update_region(CONF.identity.region)
@decorators.idempotent_id('49e08cd0-4d33-4ec4-95a0-500974b3a9bf')
def test_list_customer_additional_fields(self):
_, customer = self.client.get_customer(self.setup_customer_id)
self.assertIn('status', customer)
self.assertIn('regions', customer)
@decorators.idempotent_id('edf3a30a-fca8-49d0-acc6-c3190c6feb43')
def test_disable_customer(self):
self.client.enable_customer(self.setup_customer_id, False)
self._wait_for_status(self.setup_customer_id, 'Success')
_, body = self.client.get_customer(self.setup_customer_id)
self.assertFalse(body['enabled'])
@decorators.idempotent_id('7787b712-c398-47b6-8bf6-345be85e6daa')
def test_create_customer_with_uuid(self):
post_body = self._get_customer_params(quota=False)
uuid_ = uuid.uuid4().hex
post_body['uuid'] = uuid_
test_customer_id = self._create_customer(post_body)
_, customer = self.client.get_customer(test_customer_id)
self.assertEqual(customer['uuid'], uuid_)
@decorators.idempotent_id('8bd31cee-825a-4542-aa1d-bf2d79dbac62')
def test_replace_customer_default_and_region_quota(self):
customer = self._get_bare_customer_params()
customer['name'] = self.setup_customer['name']
# Remove quotas under regions
customer['regions'] = [{'name': CONF.identity.region}]
# Modify defaultQuotas for compute.ram
quota = self.setup_customer['defaultQuotas'][0]
quota['compute'][0]['ram'] = '20'
customer['defaultQuotas'] = [quota]
_, body = self.client.update_customer(self.setup_customer_id,
customer)
self._wait_for_status(self.setup_customer_id, 'Success')
_, body = self.client.get_customer(self.setup_customer_id)
self.assertEqual(len(body['defaultQuotas']), 1)
self.assertEqual(len(body['regions'][0]['quotas']), 0)
self.assertDictEqual(quota, body['defaultQuotas'][0])
@decorators.idempotent_id('710ee9a9-7629-4166-941c-75dbf1c10c9d')
def test_update_region_quota(self):
# Get original region quota and change a value
region_quotas = self.setup_customer['regions'][0]['quotas']
region = {}
region['type'] = 'single'
region['name'] = CONF.identity.region
region['quotas'] = copy.deepcopy(region_quotas)
region['quotas'][0]['compute'][0]['ram'] = '30'
_, body = self.client.update_region(self.setup_customer_id, *[region])
self._wait_for_status(self.setup_customer_id, 'Success')
_, body = self.client.get_customer(self.setup_customer_id)
self.assertDictEqual(region['quotas'][0],
body['regions'][0]['quotas'][0])
# Default quotas should not be affected
self.assertDictEqual(self.setup_customer['defaultQuotas'][0],
body['defaultQuotas'][0])
@testtools.skipUnless(CONF.ranger.alt_region_available,
'Alt region not provided, skipping this test')
@decorators.idempotent_id('4bd683c6-1e1c-4f5d-9f42-6af8ac0b2183')
def test_update_customer_region(self):
post_body = self._get_customer_params(
default_users=False, region_users=False)
test_customer_id = self._create_customer(post_body)
_, customer = self.client.get_customer(test_customer_id)
self.assertEqual(customer['regions'][0]['name'],
CONF.identity.region)
region = {'name': CONF.ranger.alt_region}
self.client.update_region(test_customer_id, *[region])
self._wait_for_status(test_customer_id, 'Success')
_, customer = self.client.get_customer(test_customer_id)
self.assertEqual(customer['regions'][0]['name'],
CONF.ranger.alt_region)