# 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 uuid from ranger_tempest_plugin.data_utils import data_utils from ranger_tempest_plugin.tests.api import ims_base from tempest import config from tempest import test from tempest.lib import decorators from tempest.lib import exceptions CONF = config.CONF class TestTempestIms(ims_base.ImsBaseOrmTest): @classmethod def setup_credentials(cls): super(TestTempestIms, cls).setup_credentials() @classmethod def setup_clients(cls): super(TestTempestIms, cls).setup_clients() @classmethod def resource_setup(cls): # setup public image for tempest testing cls.image_params = \ cls._get_image_params(set_private=False) cls.public_image = \ cls._create_img_and_validate_creation_on_dcp_and_lcp( **cls.image_params) # setup private image for tempest testing cls.image_params = cls._get_image_params(set_enabled=False) cls.private_image = \ cls._create_img_and_validate_creation_on_dcp_and_lcp( **cls.image_params) super(TestTempestIms, cls).resource_setup() @classmethod def resource_cleanup(cls): cls._del_img_validate_deletion_on_dcp_and_lcp(cls.public_image['id']) cls._del_img_validate_deletion_on_dcp_and_lcp(cls.private_image['id']) super(TestTempestIms, cls).resource_cleanup() def _data_setup(self, post_body): image = self._create_img_and_validate_creation_on_dcp_and_lcp( **post_body) test.addClassResourceCleanup( self._del_img_validate_deletion_on_dcp_and_lcp, image["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(self, status='functional'): region_name = data_utils.rand_name() _, region = self.os_admin.rms_client.create_region( **{ 'region_id': region_name, 'status': status, } ) test.addClassResourceCleanup( self.os_admin.rms_client.delete_region, region_name) return region @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('6072c438-1e45-4c0b-97a6-e5127bd33d90') 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) test_image_id = image['id'] # add region to image then check to confirm image status = "Success" self.client.add_region_to_image(test_image_id, self.region_id) # image status must show 'Success' when assigned to a region self._wait_for_image_status_on_dcp(test_image_id, 'Success') # check that region is successfully added _, body = self.client.get_image(test_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(test_image_id, self.region_id) self._wait_for_image_status_on_dcp(test_image_id, 'no regions') # image region array should be empty after the region was removed _, body = self.client.get_image(test_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() # call client create_IMAGE and wait till status equals 'Success' _, body = self.client.create_image(**post_body) image = body["image"] test_image_id = image["id"] self._wait_for_image_status_on_dcp(test_image_id, 'Success') # do not forget to add this account to tempest cleanup test.addClassResourceCleanup( self._del_img_validate_deletion_on_dcp_and_lcp, test_image_id) # verify image record created successfully _, body = self.client.get_image(test_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 = {} post_body = self._get_image_params(set_region=False) image = self._data_setup(post_body) test_image_id = image['id'] test.addClassResourceCleanup( self._del_img_validate_deletion_on_dcp_and_lcp, test_image_id) # setup region and change 'enabled', 'customers' properties 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(test_image_id, para=None, **post_body) self._wait_for_image_status_on_dcp(test_image_id, 'Success') # verify image record updated successfully _, body = self.client.get_image(test_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) test_image_id = image['id'] # delete the data and do get_image to ensure 404-NotFound response self._del_img_validate_deletion_on_dcp_and_lcp(test_image_id) self.assertRaises(exceptions.NotFound, self.client.get_image, test_image_id) @decorators.idempotent_id('e642fa39-1b69-4d17-8bd1-aee90ea042a3') def test_image_while_region_down(self): # create region with status down region = self._create_region(status='down') # create image within that newly created region post_body = self._get_image_params() post_body['regions'][0]['name'] = region['name'] # TODO(JH629G): Determine if flavor resource deletion is necessary here self.assertRaises(exceptions.BadRequest, self.client.create_image, **post_body) @decorators.idempotent_id('a1fee342-3000-41a6-97f9-b33fd2734e4d') def test_image_while_region_building(self): # create region with status building region = self._create_region(status='building') # create image within that newly created region post_body = self._get_image_params() post_body['regions'][0]['name'] = region['name'] _, body = self.client.create_image(**post_body) self.assertIn('id', body['image']) test_image_id = body['image']['id'] # add image to tempest cleanup test.addClassResourceCleanup( self._del_img_validate_deletion_on_dcp_and_lcp, test_image_id) # jh629g: changed out below code for above, # on first glance seems to be stronger # validation of deletion, which should be good # for stopping resource leaks # do not forget to add this account to tempest cleanup # test.addClassResourceCleanup(self._delete_image, test_image_id) _, body = self.client.get_image(test_image_id) # since region is building it will give error # Notification to ORD failed _, body = self.client.get_image(test_image_id) self.assertEqual(body['image']['id'], test_image_id) self.assertEqual(body['image']['status'], 'error') @decorators.idempotent_id('b967ce58-5d24-4af2-8416-a336772c8087') def test_image_while_region_maintenance(self): # create region with status maintenance region = self._create_region(status='maintenance') # create image within that newly created region post_body = self._get_image_params() post_body['regions'][0]['name'] = region['name'] _, body = self.client.create_image(**post_body) self.assertIn('id', body['image']) test_image_id = body['image']['id'] # add image to tempest cleanup test.addClassResourceCleanup( self._del_img_validate_deletion_on_dcp_and_lcp, test_image_id) _, body = self.client.get_image(test_image_id) # since region is maintenance it will give error # Notification to ORD failed self.assertEqual(body['image']['id'], test_image_id) self.assertEqual(body['image']['status'], 'Error') @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) test_image_id = image['id'] # 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 test_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(test_image_id, image_ids) @decorators.idempotent_id('59887b26-8e73-4781-87a4-3b505ece0021') def test_create_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"] test_image_id = image["id"] self._wait_for_image_status_on_dcp(test_image_id, 'Success') # add image to temepst cleanup test.addClassResourceCleanup( self._del_img_validate_deletion_on_dcp_and_lcp, test_image_id) # verify image record created successfully _, body = self.client.get_image(test_image_id) image = body["image"] self.assertEqual(image["regions"][0]["name"], CONF.identity.region) @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", "vyatta", "vCEImage", "mediumImage"] properties = { "Application-Name": "Vyatta", "Application-Type": "VCE", "Application-Vendor": "Brocade", "Application-Version": "3.5.R5.att-V6.0", "hw_vif_model": "VirtualVmxnet3", "OS": "Debian", "OS-Version": "7", "Post-Processing-Networking": "None", "Post-Processing-Tools": "None", "vmware-adaptertype": "ide", "vmware-disktype": "sparse" } post_body["tags"] = tags post_body["properties"] = properties # call client create_IMAGE and wait till status equals 'Success' _, body = self.client.create_image(**post_body) image = body["image"] test_image_id = image["id"] self._wait_for_image_status_on_dcp(test_image_id, 'Success') # add image to tempest cleanup test.addClassResourceCleanup( self._del_img_validate_deletion_on_dcp_and_lcp, test_image_id) # verify image record created successfully _, body = self.client.get_image(test_image_id) image = body["image"] self.assertListEqual(image["regions"][0]["tags"], tags) self.assertDictEqual(image["regions"][0]["properties"], properties) @decorators.idempotent_id('67aa7014-4dbb-4d66-bc7b-1a95a57494f8') def test_create_image_with_uuid(self): post_body = self._get_image_params() # set uuid str_uuid = uuid.uuid4().hex post_body['id'] = str_uuid # call client create_IMAGE and wait till status equals 'Success' _, body = self.client.create_image(**post_body) image = body["image"] test_image_id = image["id"] self._wait_for_image_status_on_dcp(test_image_id, 'Success') # add image to tempest cleanup test.addClassResourceCleanup( self._del_img_validate_deletion_on_dcp_and_lcp, test_image_id) # verify image record created successfully _, body = self.client.get_image(test_image_id) image = body["image"] self.assertEqual(image["regions"][0]["id"], str_uuid) @decorators.idempotent_id('ae1223b5-cb75-442b-82eb-488969acc978') def test_create_flavor_with_region_group(self): post_body = self._get_image_params() # region group region_group = {"name": "NCLargetest", "type": "group"} # update region_group to regions post_body["regions"].append(region_group) # call client create_IMAGE and wait till status equals 'Success' _, body = self.client.create_image(**post_body) image = body["image"] test_image_id = image["id"] self._wait_for_image_status_on_dcp(test_image_id, 'Success') # add image to tempest cleanup test.addClassResourceCleanup( self._del_img_validate_deletion_on_dcp_and_lcp, test_image_id) # verify image record created successfully _, body = self.client.get_image(test_image_id) image = body["image"] # Aggregate Status self.assertEqual(image["status"], 'Success') # Region Status self.assertEqual(image["regions"][1]["status"], 'Success') # region group self.assertDictEqual(image["regions"][1]["name"], "NCLargetest")