<s> """<STR_LIT>""" <EOL> from cloudcafe . common . tools . datagen import random_string <EOL> from cloudcafe . compute . composites import ComputeIntegrationComposite <EOL> from cloudroast . blockstorage . volumes_api . fixtures import VolumesTestFixture <EOL> class ComputeIntegrationTestFixture ( VolumesTestFixture ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( ComputeIntegrationTestFixture , cls ) . setUpClass ( ) <EOL> cls . compute = ComputeIntegrationComposite ( ) <EOL> cls . servers = cls . compute . servers <EOL> cls . flavors = cls . compute . flavors <EOL> cls . images = cls . compute . images <EOL> cls . volume_attachments = cls . compute . volume_attachments <EOL> @ classmethod <EOL> def random_server_name ( cls ) : <EOL> return random_string ( prefix = "<STR_LIT>" , size = <NUM_LIT:10> ) <EOL> @ classmethod <EOL> def new_server ( <EOL> cls , name = None , image = None , flavor = None , add_cleanup = True ) : <EOL> name = name or cls . random_server_name ( ) <EOL> image = image or cls . images . config . primary_image <EOL> flavor = flavor or cls . flavors . config . primary_flavor <EOL> resp = cls . servers . behaviors . create_active_server ( <EOL> name , image_ref = image , flavor_ref = flavor ) <EOL> if add_cleanup : <EOL> cls . addClassCleanup ( <EOL> cls . servers . client . delete_server , resp . entity . id ) <EOL> return resp . entity <EOL> @ classmethod <EOL> def attach_volume_and_get_device_info ( <EOL> cls , server_connection , server_id , volume_id ) : <EOL> original_details = server_connection . get_all_disk_details ( ) <EOL> attachment = cls . volume_attachments . behaviors . attach_volume_to_server ( <EOL> server_id , volume_id ) <EOL> assert attachment , "<STR_LIT>" . format ( <EOL> volume_id , server_id ) <EOL> new_details = server_connection . get_all_disk_details ( ) <EOL> volume_details = [ d for d in new_details if d not in original_details ] <EOL> cls . fixture_log . debug ( volume_details ) <EOL> assert len ( volume_details ) == <NUM_LIT:1> , ( <EOL> "<STR_LIT>" ) <EOL> setattr ( attachment , '<STR_LIT>' , volume_details ) <EOL> os_disk_device_name = volume_details [ <NUM_LIT:0> ] . get ( '<STR_LIT>' ) or "<STR_LIT>" . format ( <EOL> volume_details [ <NUM_LIT:0> ] . get ( '<STR_LIT:name>' ) ) <EOL> assert os_disk_device_name , ( <EOL> "<STR_LIT>" ) <EOL> setattr ( attachment , '<STR_LIT>' , os_disk_device_name ) <EOL> return attachment <EOL> @ classmethod <EOL> def format_attached_volume ( <EOL> cls , server_connection , device_name , fstype = None ) : <EOL> resp = None <EOL> if device_name . startswith ( '<STR_LIT>' ) : <EOL> resp = server_connection . format_disk ( device_name , fstype or '<STR_LIT>' ) <EOL> else : <EOL> resp = server_connection . format_disk ( device_name , fstype or '<STR_LIT>' ) <EOL> assert resp is not None , ( <EOL> "<STR_LIT>" ) <EOL> return resp <EOL> @ classmethod <EOL> def mount_attached_volume ( <EOL> cls , server_connection , device_name , mount_point = None ) : <EOL> mount_point = mount_point or server_connection . generate_mountpoint ( ) <EOL> if device_name . startswith ( '<STR_LIT>' ) : <EOL> server_connection . create_directory ( mount_point ) <EOL> return server_connection . mount_disk ( <EOL> source_path = device_name , destination_path = mount_point ) <EOL> @ classmethod <EOL> def unmount_attached_volume ( cls , server_connection , device_name ) : <EOL> return server_connection . unmount_disk ( device_name ) <EOL> @ classmethod <EOL> def _add_directory_prefix ( cls , file_directory_string ) : <EOL> if not file_directory_string . startswith ( '<STR_LIT:/>' ) : <EOL> if len ( file_directory_string ) == <NUM_LIT:1> : <EOL> file_directory_string = file_directory_string + "<STR_LIT>" <EOL> return file_directory_string <EOL> @ classmethod <EOL> def get_remote_file_md5_hash ( <EOL> cls , server_connection , file_directory , file_name ) : <EOL> file_directory = cls . _add_directory_prefix ( file_directory ) <EOL> return server_connection . get_md5sum_for_remote_file ( <EOL> file_directory , file_name ) <EOL> @ classmethod <EOL> def create_remote_file ( <EOL> cls , server_connection , file_directory , file_name , <EOL> file_content = None ) : <EOL> file_content = file_content or "<STR_LIT:content>" <EOL> file_directory = cls . _add_directory_prefix ( file_directory ) <EOL> return server_connection . create_file ( <EOL> file_name , file_content , file_directory ) <EOL> @ classmethod <EOL> def _get_remote_client ( cls , client_type ) : <EOL> client = None <EOL> if client_type == '<STR_LIT>' : <EOL> from cloudcafe . compute . common . clients . remote_instance . windows . windows_client import WindowsClient <EOL> client = WindowsClient <EOL> if client_type == '<STR_LIT>' : <EOL> from cloudcafe . compute . common . clients . remote_instance . linux . linux_client import LinuxClient <EOL> client = LinuxClient <EOL> if not client : <EOL> raise Exception ( <EOL> "<STR_LIT>" . format ( client_type ) ) <EOL> return client <EOL> @ classmethod <EOL> def _connect ( <EOL> cls , remote_client , ip_address = None , username = None , <EOL> connection_timeout = None , key = None , password = None ) : <EOL> kwargs = { <EOL> '<STR_LIT>' : ip_address , <EOL> '<STR_LIT:username>' : username , <EOL> '<STR_LIT>' : connection_timeout } <EOL> auth_strategy = "<STR_LIT:key>" if key else "<STR_LIT:password>" <EOL> kwargs [ auth_strategy ] = key or password <EOL> _client = remote_client ( ** kwargs ) <EOL> return _client <EOL> @ classmethod <EOL> def connect_to_server ( <EOL> cls , ip_address , username = '<STR_LIT:root>' , password = None , key = None , <EOL> connection_timeout = None , client_type = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> remote_client = cls . _get_remote_client ( client_type ) <EOL> return cls . _connect ( <EOL> remote_client , ip_address = ip_address , username = username , <EOL> connection_timout = connection_timeout , key = key , <EOL> password = password ) <EOL> @ classmethod <EOL> def get_image_os_type ( cls , image_id ) : <EOL> image = cls . images . client . get_image ( image_id ) . entity <EOL> return image . metadata . get ( '<STR_LIT>' , '<STR_LIT>' ) . lower ( ) <EOL> @ classmethod <EOL> def connect_to_instance ( <EOL> cls , server_instance_model , key = None , connection_timeout = None , <EOL> os_type = None ) : <EOL> """<STR_LIT>""" <EOL> _usernames = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT:root>' } <EOL> ip_address = None <EOL> if hasattr ( server_instance_model , '<STR_LIT>' ) : <EOL> ip_address = server_instance_model . accessIPv4 <EOL> else : <EOL> ip_address = server_instance_model . addresses . public . ipv4 <EOL> if os_type is None : <EOL> os_type = cls . get_image_os_type ( server_instance_model . image . id ) <EOL> username = _usernames . get ( os_type ) <EOL> password = server_instance_model . admin_pass <EOL> connection_timeout = connection_timeout or cls . servers . config . connection_timeout <EOL> remote_client = cls . _get_remote_client ( os_type ) <EOL> return cls . _connect ( <EOL> remote_client , ip_address = ip_address , username = username , <EOL> connection_timeout = connection_timeout , key = key , <EOL> password = password ) <EOL> @ classmethod <EOL> def setup_server_and_attached_volume_with_data ( <EOL> cls , server = None , volume = None ) : <EOL> """<STR_LIT>""" <EOL> cls . test_server = server or cls . new_server ( ) <EOL> cls . server_conn = cls . connect_to_instance ( cls . test_server ) <EOL> cls . volume_mount_point = cls . server_conn . generate_mountpoint ( ) <EOL> cls . test_volume = volume or cls . new_volume ( ) <EOL> cls . test_attachment = cls . attach_volume_and_get_device_info ( <EOL> cls . server_conn , cls . test_server . id , cls . test_volume . id_ ) <EOL> cls . format_attached_volume ( <EOL> cls . server_conn , cls . test_attachment . os_disk_device_name ) <EOL> cls . mount_attached_volume ( <EOL> cls . server_conn , cls . test_attachment . os_disk_device_name , <EOL> mount_point = cls . volume_mount_point ) <EOL> cls . written_data = "<STR_LIT:a>" * <NUM_LIT> <EOL> cls . written_filename = "<STR_LIT>" <EOL> resp = cls . create_remote_file ( <EOL> cls . server_conn , cls . volume_mount_point , cls . written_filename , <EOL> file_content = cls . written_data ) <EOL> assert resp is not None , ( <EOL> "<STR_LIT>" ) <EOL> cls . original_md5hash = cls . get_remote_file_md5_hash ( <EOL> cls . server_conn , cls . volume_mount_point , cls . written_filename ) <EOL> assert cls . original_md5hash is not None , ( <EOL> "<STR_LIT>" ) <EOL> cls . server_conn . filesystem_sync ( ) <EOL> @ classmethod <EOL> def unmount_and_detach_test_volume ( cls ) : <EOL> cls . unmount_attached_volume ( <EOL> cls . server_conn , cls . test_attachment . os_disk_device_name ) <EOL> cls . volume_attachments . behaviors . delete_volume_attachment ( <EOL> cls . test_attachment . id_ , cls . test_server . id ) <EOL> def calculate_volume_size_for_image ( self , image ) : <EOL> """<STR_LIT>""" <EOL> size = getattr ( image , '<STR_LIT>' , None ) <EOL> if not size : <EOL> msg = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> image_id = image . id ) ) <EOL> self . fixture_log . warning ( msg ) <EOL> return max ( size , self . volumes . config . min_volume_from_image_size ) <EOL> def _compare_volume_image_metadata ( self , image , volume , key_list = None ) : <EOL> key_list = key_list or [ ] <EOL> comparable_keys = [ <EOL> key for key in image . metadata . keys ( ) if key in key_list ] <EOL> error_messages = [ ] <EOL> for key in comparable_keys : <EOL> if key not in volume . volume_image_metadata : <EOL> error_messages . append ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> key , image . id , volume . id_ ) ) <EOL> elif volume . volume_image_metadata [ key ] != image . metadata [ key ] : <EOL> error_messages . append ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> key , image . metadata [ key ] , image . id , <EOL> key , volume . volume_image_metadata [ key ] , volume . id_ ) ) <EOL> return error_messages <EOL> def assertImageMetadataWasCopiedToVolume ( <EOL> self , image , volume , key_list = None , msg = None ) : <EOL> errors = self . _compare_volume_image_metadata ( image , volume , key_list ) <EOL> if errors : <EOL> self . fail ( self . _formatMessage ( msg , "<STR_LIT:\n>" . join ( errors ) ) ) <EOL> def assertMinDiskSizeIsSet ( self , image , msg = None ) : <EOL> if getattr ( image , '<STR_LIT>' , <NUM_LIT:0> ) <= <NUM_LIT:0> : <EOL> stdmsg = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( image . id , image . name ) ) <EOL> self . fail ( self . _formatMessage ( msg , stdmsg ) ) <EOL> def check_if_minimum_disk_size_is_set ( self , image ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . assertMinDiskSizeIsSet ( image ) <EOL> except AssertionError : <EOL> return False <EOL> return True <EOL> def make_server_snapshot ( self , server , add_cleanup = True ) : <EOL> server_snapshot_name = random_string ( <EOL> prefix = "<STR_LIT>" . format ( server . name ) , size = <NUM_LIT:10> ) <EOL> create_img_resp = self . servers . client . create_image ( <EOL> server . id , name = server_snapshot_name ) <EOL> assert create_img_resp . ok , ( <EOL> "<STR_LIT>" . format ( <EOL> create_img_resp . status_code ) ) <EOL> self . images . behaviors . verify_server_snapshotting_progression ( server . id ) <EOL> list_imgs_resp = self . images . client . list_images ( ) <EOL> assert list_imgs_resp . ok , ( <EOL> "<STR_LIT>" . format ( <EOL> list_imgs_resp . status_code ) ) <EOL> assert list_imgs_resp . entity is not None , ( <EOL> "<STR_LIT>" . format ( <EOL> list_imgs_resp . status_code ) ) <EOL> image_list = list_imgs_resp . entity <EOL> server_snapshot = None <EOL> for img in image_list : <EOL> if img . name == server_snapshot_name : <EOL> server_snapshot = img <EOL> break <EOL> assert server_snapshot is not None , "<STR_LIT>" <EOL> if add_cleanup is True : <EOL> self . addCleanup ( <EOL> self . images . client . delete_image , server_snapshot . id ) <EOL> self . images . behaviors . wait_for_image_status ( <EOL> server_snapshot . id , '<STR_LIT>' , <NUM_LIT:10> , <NUM_LIT> ) <EOL> resp = self . images . client . get_image ( server_snapshot . id ) <EOL> assert resp . ok , ( "<STR_LIT>" ) <EOL> assert resp . entity is not None , ( <EOL> "<STR_LIT>" ) <EOL> return resp . entity <EOL> def create_bootable_volume_from_server_snapshot ( <EOL> self , image , flavor , volume_type ) : <EOL> server = self . new_server ( <EOL> name = None , image = image . id , flavor = flavor . id , add_cleanup = False ) <EOL> self . addCleanup ( self . servers . client . delete_server , server . id ) <EOL> server_snapshot = self . make_server_snapshot ( server ) <EOL> return self . create_volume_from_image_test ( volume_type , server_snapshot ) <EOL> def create_volume_from_image_test ( <EOL> self , volume_type , image , add_cleanup = True ) : <EOL> size = self . calculate_volume_size_for_image ( image ) <EOL> volume = self . volumes . behaviors . create_available_volume ( <EOL> size , volume_type . id_ , image_ref = image . id , <EOL> timeout = self . volumes . config . volume_create_max_timeout ) <EOL> if add_cleanup : <EOL> try : <EOL> self . addCleanup ( <EOL> self . volumes . behaviors . delete_volume_confirmed , volume . id_ ) <EOL> except : <EOL> raise Exception ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEquals ( <EOL> str ( size ) , str ( volume . size ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( size , volume . size ) ) <EOL> self . assertEquals ( <EOL> '<STR_LIT:true>' , volume . bootable , "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertImageMetadataWasCopiedToVolume ( image , volume ) <EOL> return volume <EOL> def create_bootable_volume_from_third_snapshot_of_server_test ( <EOL> self , image , flavor , volume_type ) : <EOL> server = self . new_server ( <EOL> name = None , image = image . id , flavor = flavor . id , add_cleanup = False ) <EOL> self . addCleanup ( self . servers . client . delete_server , server . id ) <EOL> self . make_server_snapshot ( server ) <EOL> self . servers . behaviors . wait_for_server_status ( <EOL> server . id , '<STR_LIT>' , timeout = <NUM_LIT> ) <EOL> self . make_server_snapshot ( server ) <EOL> self . servers . behaviors . wait_for_server_status ( <EOL> server . id , '<STR_LIT>' , timeout = <NUM_LIT> ) <EOL> server_snapshot_3 = self . make_server_snapshot ( server ) <EOL> self . servers . behaviors . wait_for_server_status ( <EOL> server . id , '<STR_LIT>' , timeout = <NUM_LIT> ) <EOL> self . create_volume_from_image_test ( volume_type , server_snapshot_3 ) </s>
