<s> import math <EOL> from oslo_log import log <EOL> from tempest . lib . common . utils import data_utils <EOL> import testtools <EOL> from ec2api . tests . functional import base <EOL> from ec2api . tests . functional import config <EOL> CONF = config . CONF <EOL> LOG = log . getLogger ( __name__ ) <EOL> class EC2_EBSInstanceTuneBDM ( base . EC2TestCase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> @ base . safe_setup <EOL> def setUpClass ( cls ) : <EOL> super ( EC2_EBSInstanceTuneBDM , cls ) . setUpClass ( ) <EOL> if not CONF . aws . ebs_image_id : <EOL> raise cls . skipException ( '<STR_LIT>' ) <EOL> cls . image_id = CONF . aws . ebs_image_id <EOL> cls . zone = CONF . aws . aws_zone <EOL> data = cls . client . describe_images ( ImageIds = [ cls . image_id ] ) <EOL> assert <NUM_LIT:1> == len ( data [ '<STR_LIT>' ] ) <EOL> image = data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> cls . root_device_name = image [ '<STR_LIT>' ] <EOL> bdm = image [ '<STR_LIT>' ] <EOL> bdm = [ v for v in bdm if v [ '<STR_LIT>' ] == cls . root_device_name ] <EOL> assert <NUM_LIT:1> == len ( bdm ) <EOL> ebs = bdm [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> cls . root_device_size = ebs . get ( '<STR_LIT>' ) <EOL> if not cls . root_device_size : <EOL> snapshotId = ebs . get ( '<STR_LIT>' ) <EOL> data = cls . client . describe_snapshots ( SnapshotIds = [ snapshotId ] ) <EOL> assert <NUM_LIT:1> == len ( data [ '<STR_LIT>' ] ) <EOL> cls . root_device_size = data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> def test_launch_ebs_instance_with_persistent_root_device ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . run_instance ( ImageId = self . image_id , <EOL> BlockDeviceMappings = [ { '<STR_LIT>' : self . root_device_name , <EOL> '<STR_LIT>' : { '<STR_LIT>' : False } } ] ) <EOL> bdt = self . get_instance_bdm ( instance_id , self . root_device_name ) <EOL> self . assertIsNotNone ( bdt ) <EOL> volume_id = bdt [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> res_clean_vol = self . addResourceCleanUp ( self . client . delete_volume , <EOL> VolumeId = volume_id ) <EOL> self . assertIsNotNone ( volume_id ) <EOL> self . assertFalse ( bdt [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> data = self . client . describe_volumes ( VolumeIds = [ volume_id ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( data [ '<STR_LIT>' ] ) ) <EOL> self . client . terminate_instances ( InstanceIds = [ instance_id ] ) <EOL> self . get_instance_waiter ( ) . wait_delete ( instance_id ) <EOL> self . get_volume_waiter ( ) . wait_available ( volume_id ) <EOL> self . client . delete_volume ( VolumeId = volume_id ) <EOL> self . cancelResourceCleanUp ( res_clean_vol ) <EOL> self . get_volume_waiter ( ) . wait_delete ( volume_id ) <EOL> def test_launch_ebs_instance_with_resized_root_device ( self ) : <EOL> """<STR_LIT>""" <EOL> new_size = int ( math . ceil ( self . root_device_size * <NUM_LIT> ) ) <EOL> instance_id = self . run_instance ( ImageId = self . image_id , <EOL> BlockDeviceMappings = [ { '<STR_LIT>' : self . root_device_name , <EOL> '<STR_LIT>' : { '<STR_LIT>' : new_size } } ] ) <EOL> bdt = self . get_instance_bdm ( instance_id , self . root_device_name ) <EOL> self . assertIsNotNone ( bdt ) <EOL> volume_id = bdt [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( volume_id ) <EOL> self . assertTrue ( bdt [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> data = self . client . describe_volumes ( VolumeIds = [ volume_id ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( data [ '<STR_LIT>' ] ) ) <EOL> volume = data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( new_size , volume [ '<STR_LIT>' ] ) <EOL> self . client . terminate_instances ( InstanceIds = [ instance_id ] ) <EOL> self . get_instance_waiter ( ) . wait_delete ( instance_id ) <EOL> def test_launch_ebs_instance_with_creating_blank_volume ( self ) : <EOL> """<STR_LIT>""" <EOL> device_name_prefix = base . get_device_name_prefix ( self . root_device_name ) <EOL> device_name = device_name_prefix + '<STR_LIT:b>' <EOL> instance_id = self . run_instance ( ImageId = self . image_id , <EOL> BlockDeviceMappings = [ { '<STR_LIT>' : device_name , <EOL> '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:1> } } ] ) <EOL> bdt = self . get_instance_bdm ( instance_id , device_name ) <EOL> self . assertIsNotNone ( bdt ) <EOL> volume_id = bdt [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( volume_id ) <EOL> self . assertTrue ( bdt [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> data = self . client . describe_volumes ( VolumeIds = [ volume_id ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( data [ '<STR_LIT>' ] ) ) <EOL> volume = data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( <NUM_LIT:1> , volume [ '<STR_LIT>' ] ) <EOL> self . client . terminate_instances ( InstanceIds = [ instance_id ] ) <EOL> self . get_instance_waiter ( ) . wait_delete ( instance_id ) <EOL> class EC2_EBSInstanceAttaching ( base . EC2TestCase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> @ base . safe_setup <EOL> def setUpClass ( cls ) : <EOL> super ( EC2_EBSInstanceAttaching , cls ) . setUpClass ( ) <EOL> if not CONF . aws . run_incompatible_tests : <EOL> raise cls . skipException ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if not CONF . aws . ebs_image_id : <EOL> raise cls . skipException ( '<STR_LIT>' ) <EOL> cls . image_id = CONF . aws . ebs_image_id <EOL> cls . zone = CONF . aws . aws_zone <EOL> data = cls . client . describe_images ( ImageIds = [ cls . image_id ] ) <EOL> assert <NUM_LIT:1> == len ( data [ '<STR_LIT>' ] ) <EOL> image = data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> root_device_name = image [ '<STR_LIT>' ] <EOL> device_name_prefix = base . get_device_name_prefix ( root_device_name ) <EOL> cls . full_device_name_prefix = device_name_prefix <EOL> cls . short_device_name_prefix = device_name_prefix [ len ( "<STR_LIT>" ) : ] <EOL> data = cls . client . create_volume ( AvailabilityZone = cls . zone , <EOL> Size = <NUM_LIT:1> ) <EOL> cls . volume_id = data [ '<STR_LIT>' ] <EOL> cls . addResourceCleanUpStatic ( cls . client . delete_volume , <EOL> VolumeId = cls . volume_id ) <EOL> cls . get_volume_waiter ( ) . wait_available ( cls . volume_id ) <EOL> data = cls . client . create_snapshot ( VolumeId = cls . volume_id ) <EOL> cls . snapshot_id = data [ '<STR_LIT>' ] <EOL> cls . addResourceCleanUpStatic ( cls . client . delete_snapshot , <EOL> SnapshotId = cls . snapshot_id ) <EOL> cls . get_snapshot_waiter ( ) . wait_available ( cls . snapshot_id , <EOL> final_set = ( '<STR_LIT>' ) ) <EOL> instance_type = CONF . aws . instance_type <EOL> cls . device1_name = cls . full_device_name_prefix + "<STR_LIT:d>" <EOL> cls . device2_name = cls . short_device_name_prefix + "<STR_LIT:h>" <EOL> data = cls . client . run_instances ( <EOL> ImageId = cls . image_id , InstanceType = instance_type , <EOL> Placement = { '<STR_LIT>' : cls . zone } , MinCount = <NUM_LIT:1> , MaxCount = <NUM_LIT:1> , <EOL> BlockDeviceMappings = [ { '<STR_LIT>' : cls . device1_name , <EOL> '<STR_LIT>' : { '<STR_LIT>' : cls . snapshot_id , <EOL> '<STR_LIT>' : True } } , <EOL> { '<STR_LIT>' : cls . device2_name , <EOL> '<STR_LIT>' : { '<STR_LIT>' : cls . snapshot_id , <EOL> '<STR_LIT>' : True } } ] ) <EOL> instance_id = data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> cls . instance_id = instance_id <EOL> cls . addResourceCleanUpStatic ( cls . client . terminate_instances , <EOL> InstanceIds = [ instance_id ] ) <EOL> cls . get_instance_waiter ( ) . wait_available ( instance_id , <EOL> final_set = ( '<STR_LIT>' ) ) <EOL> data = cls . client . describe_instances ( InstanceIds = [ instance_id ] ) <EOL> assert <NUM_LIT:1> == len ( data . get ( '<STR_LIT>' , [ ] ) ) <EOL> instances = data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] . get ( '<STR_LIT>' , [ ] ) <EOL> assert <NUM_LIT:1> == len ( instances ) <EOL> instance = instances [ <NUM_LIT:0> ] <EOL> bdms = instance [ '<STR_LIT>' ] <EOL> for bdt in bdms : <EOL> if bdt [ '<STR_LIT>' ] == cls . device1_name : <EOL> cls . volume_id1 = bdt [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> if bdt [ '<STR_LIT>' ] == cls . device2_name : <EOL> cls . volume_id2 = bdt [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> assert cls . volume_id1 <EOL> assert cls . volume_id2 <EOL> @ classmethod <EOL> def tearDownClass ( cls ) : <EOL> super ( EC2_EBSInstanceAttaching , cls ) . tearDownClass ( ) <EOL> for volume_id in [ cls . volume_id1 , cls . volume_id2 ] : <EOL> try : <EOL> cls . cleanUpItem ( cls . client . delete_volume , [ ] , <EOL> { '<STR_LIT>' : volume_id } ) <EOL> except BaseException : <EOL> LOG . exception ( '<STR_LIT>' ) <EOL> def _test_attaching ( self , volume_id , device_name , device_prefix , <EOL> new_device_name_letter ) : <EOL> self . client . detach_volume ( VolumeId = volume_id ) <EOL> clean_v = self . addResourceCleanUp ( self . client . delete_volume , <EOL> VolumeId = volume_id ) <EOL> self . get_volume_attachment_waiter ( ) . wait_delete ( volume_id ) <EOL> bdt = self . get_instance_bdm ( self . instance_id , device_name ) <EOL> self . assertIsNone ( bdt ) <EOL> self . client . attach_volume ( InstanceId = self . instance_id , <EOL> VolumeId = volume_id , <EOL> Device = device_name ) <EOL> self . cancelResourceCleanUp ( clean_v ) <EOL> self . get_volume_attachment_waiter ( ) . wait_available ( <EOL> volume_id , final_set = ( '<STR_LIT>' ) ) <EOL> data = self . client . describe_volumes ( VolumeIds = [ volume_id ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( data [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( '<STR_LIT>' , data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> bdt = self . get_instance_bdm ( self . instance_id , device_name ) <EOL> self . assertIsNotNone ( bdt ) <EOL> self . client . detach_volume ( VolumeId = volume_id ) <EOL> clean_v = self . addResourceCleanUp ( self . client . delete_volume , <EOL> VolumeId = volume_id ) <EOL> self . get_volume_attachment_waiter ( ) . wait_delete ( volume_id ) <EOL> bdt = self . get_instance_bdm ( self . instance_id , device_name ) <EOL> self . assertIsNone ( bdt ) <EOL> new_device_name = device_prefix + new_device_name_letter <EOL> self . client . attach_volume ( InstanceId = self . instance_id , <EOL> VolumeId = volume_id , <EOL> Device = new_device_name ) <EOL> self . cancelResourceCleanUp ( clean_v ) <EOL> self . get_volume_attachment_waiter ( ) . wait_available ( <EOL> volume_id , final_set = ( '<STR_LIT>' ) ) <EOL> data = self . client . describe_volumes ( VolumeIds = [ volume_id ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( data [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( '<STR_LIT>' , data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> bdt = self . get_instance_bdm ( self . instance_id , new_device_name ) <EOL> self . assertIsNotNone ( bdt ) <EOL> def test_attaching_by_full_name ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_attaching ( self . volume_id1 , self . device1_name , <EOL> self . full_device_name_prefix , "<STR_LIT:e>" ) <EOL> def test_attaching_by_short_name ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _test_attaching ( self . volume_id2 , self . device2_name , <EOL> self . short_device_name_prefix , "<STR_LIT:i>" ) <EOL> class EC2_EBSInstanceSnapshot ( base . EC2TestCase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> @ base . safe_setup <EOL> def setUpClass ( cls ) : <EOL> super ( EC2_EBSInstanceSnapshot , cls ) . setUpClass ( ) <EOL> if not CONF . aws . ebs_image_id : <EOL> raise cls . skipException ( '<STR_LIT>' ) <EOL> cls . image_id = CONF . aws . ebs_image_id <EOL> cls . zone = CONF . aws . aws_zone <EOL> def test_create_ebs_instance_snapshot ( self ) : <EOL> """<STR_LIT>""" <EOL> instance_id = self . run_instance ( ImageId = self . image_id ) <EOL> instance = self . get_instance ( instance_id ) <EOL> bdt = self . get_instance_bdm ( instance_id , None ) <EOL> self . assertIsNotNone ( bdt ) <EOL> volume_id = bdt [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( volume_id ) <EOL> self . client . stop_instances ( InstanceIds = [ instance_id ] ) <EOL> self . get_instance_waiter ( ) . wait_available ( instance_id , <EOL> final_set = ( '<STR_LIT>' ) ) <EOL> self . addResourceCleanUp ( self . client . delete_volume , <EOL> VolumeId = volume_id ) <EOL> data = self . client . create_snapshot ( VolumeId = volume_id ) <EOL> snapshot_id = data [ '<STR_LIT>' ] <EOL> self . addResourceCleanUp ( self . client . delete_snapshot , <EOL> SnapshotId = snapshot_id ) <EOL> self . get_snapshot_waiter ( ) . wait_available ( snapshot_id , <EOL> final_set = ( '<STR_LIT>' ) ) <EOL> self . client . terminate_instances ( InstanceIds = [ instance_id ] ) <EOL> self . get_instance_waiter ( ) . wait_delete ( instance_id ) <EOL> kwargs = { <EOL> '<STR_LIT:Name>' : data_utils . rand_name ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : instance [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : instance [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : { '<STR_LIT>' : snapshot_id , <EOL> '<STR_LIT>' : True } } ] <EOL> } <EOL> if '<STR_LIT>' in instance : <EOL> kwargs [ '<STR_LIT>' ] = instance [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in instance : <EOL> kwargs [ '<STR_LIT>' ] = instance [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in instance : <EOL> kwargs [ '<STR_LIT>' ] = instance [ '<STR_LIT>' ] <EOL> data = self . client . register_image ( * [ ] , ** kwargs ) <EOL> image_id = data [ '<STR_LIT>' ] <EOL> clean_i = self . addResourceCleanUp ( self . client . deregister_image , <EOL> ImageId = image_id ) <EOL> self . get_image_waiter ( ) . wait_available ( image_id ) <EOL> instance_id = self . run_instance ( ImageId = image_id ) <EOL> self . client . terminate_instances ( InstanceIds = [ instance_id ] ) <EOL> self . get_instance_waiter ( ) . wait_delete ( instance_id ) <EOL> self . client . deregister_image ( ImageId = image_id ) <EOL> self . cancelResourceCleanUp ( clean_i ) <EOL> class EC2_EBSInstanceResizeRootDevice ( base . EC2TestCase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> @ base . safe_setup <EOL> def setUpClass ( cls ) : <EOL> super ( EC2_EBSInstanceResizeRootDevice , cls ) . setUpClass ( ) <EOL> if not CONF . aws . ebs_image_id : <EOL> raise cls . skipException ( '<STR_LIT>' ) <EOL> cls . image_id = CONF . aws . ebs_image_id <EOL> cls . zone = CONF . aws . aws_zone <EOL> @ testtools . skipUnless ( <EOL> CONF . aws . run_incompatible_tests , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_resize_root_ebs_device ( self ) : <EOL> """<STR_LIT>""" <EOL> clean_dict = dict ( ) <EOL> instance_id = self . run_instance ( clean_dict = clean_dict , <EOL> ImageId = self . image_id ) <EOL> res_clean = clean_dict [ '<STR_LIT>' ] <EOL> instance = self . get_instance ( instance_id ) <EOL> bdt = self . get_instance_bdm ( instance_id , None ) <EOL> self . assertIsNotNone ( bdt ) <EOL> volume_id = bdt [ '<STR_LIT>' ] . get ( '<STR_LIT>' ) <EOL> self . assertIsNotNone ( volume_id ) <EOL> self . client . stop_instances ( InstanceIds = [ instance_id ] ) <EOL> self . get_instance_waiter ( ) . wait_available ( instance_id , <EOL> final_set = ( '<STR_LIT>' ) ) <EOL> self . client . detach_volume ( VolumeId = volume_id ) <EOL> clean_v = self . addResourceCleanUp ( self . client . delete_volume , <EOL> VolumeId = volume_id ) <EOL> self . get_volume_attachment_waiter ( ) . wait_delete ( volume_id ) <EOL> data = self . client . create_snapshot ( VolumeId = volume_id ) <EOL> snapshot_id = data [ '<STR_LIT>' ] <EOL> clean_s = self . addResourceCleanUp ( self . client . delete_snapshot , <EOL> SnapshotId = snapshot_id ) <EOL> self . get_snapshot_waiter ( ) . wait_available ( snapshot_id , <EOL> final_set = ( '<STR_LIT>' ) ) <EOL> new_size = int ( math . ceil ( data [ '<STR_LIT>' ] * <NUM_LIT> ) ) <EOL> data = self . client . create_volume ( AvailabilityZone = self . zone , <EOL> Size = new_size , <EOL> SnapshotId = snapshot_id ) <EOL> volume_id2 = data [ '<STR_LIT>' ] <EOL> clean_v2 = self . addResourceCleanUp ( self . client . delete_volume , <EOL> VolumeId = volume_id2 ) <EOL> self . get_volume_waiter ( ) . wait_available ( volume_id2 ) <EOL> self . client . delete_snapshot ( SnapshotId = snapshot_id ) <EOL> self . cancelResourceCleanUp ( clean_s ) <EOL> self . client . delete_volume ( VolumeId = volume_id ) <EOL> self . cancelResourceCleanUp ( clean_v ) <EOL> self . get_volume_waiter ( ) . wait_delete ( volume_id ) <EOL> self . client . attach_volume ( <EOL> InstanceId = instance_id , VolumeId = volume_id2 , <EOL> Device = instance [ '<STR_LIT>' ] ) <EOL> self . get_volume_attachment_waiter ( ) . wait_available ( <EOL> volume_id2 , final_set = ( '<STR_LIT>' ) ) <EOL> self . cancelResourceCleanUp ( res_clean ) <EOL> res_clean = self . addResourceCleanUp ( self . client . terminate_instances , <EOL> InstanceIds = [ instance_id ] ) <EOL> self . client . start_instances ( InstanceIds = [ instance_id ] ) <EOL> self . get_instance_waiter ( ) . wait_available ( instance_id , <EOL> final_set = ( '<STR_LIT>' ) ) <EOL> self . client . terminate_instances ( InstanceIds = [ instance_id ] ) <EOL> self . cancelResourceCleanUp ( res_clean ) <EOL> self . get_instance_waiter ( ) . wait_delete ( instance_id ) <EOL> self . client . delete_volume ( VolumeId = volume_id2 ) <EOL> self . cancelResourceCleanUp ( clean_v2 ) </s>
