<s> """<STR_LIT>""" <EOL> import base64 <EOL> import unittest <EOL> from cafe . drivers . unittest . decorators import tags <EOL> from cloudcafe . common . tools . datagen import rand_name <EOL> from cloudcafe . compute . common . types import ComputeHypervisors , NovaServerStatusTypes <EOL> from cloudcafe . compute . config import ComputeConfig <EOL> from cloudroast . compute . fixtures import ServerFromImageFixture <EOL> from cloudcafe . compute . servers_api . config import ServersConfig <EOL> compute_config = ComputeConfig ( ) <EOL> hypervisor = compute_config . hypervisor . lower ( ) <EOL> @ unittest . skipIf ( <EOL> hypervisor in [ ComputeHypervisors . LXC_LIBVIRT , ComputeHypervisors . IRONIC ] , <EOL> '<STR_LIT>' ) <EOL> class RebuildServerTests ( object ) : <EOL> compute_config = ComputeConfig ( ) <EOL> hypervisor = compute_config . hypervisor . lower ( ) <EOL> servers_config = ServersConfig ( ) <EOL> file_injection_enabled = servers_config . personality_file_injection_enabled <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT>' ) <EOL> def test_verify_rebuild_server_response ( self ) : <EOL> """<STR_LIT>""" <EOL> rebuilt_server = self . rebuilt_server_response . entity <EOL> self . assertEqual ( rebuilt_server . name , self . name , <EOL> msg = "<STR_LIT>" ) <EOL> self . assertEqual ( rebuilt_server . image . id , self . image_ref_alt , <EOL> msg = "<STR_LIT>" ) <EOL> self . assertEqual ( rebuilt_server . flavor . id , self . flavor_ref , <EOL> msg = "<STR_LIT>" ) <EOL> self . assertEqual ( rebuilt_server . id , self . server . id , <EOL> msg = "<STR_LIT>" ) <EOL> self . assertEqual ( rebuilt_server . links . bookmark , <EOL> self . server . links . bookmark , <EOL> msg = "<STR_LIT>" ) <EOL> self . assertEqual ( rebuilt_server . metadata . get ( '<STR_LIT:key>' ) , '<STR_LIT:value>' ) <EOL> self . assertEqual ( rebuilt_server . created , self . server . created , <EOL> msg = "<STR_LIT>" ) <EOL> self . assertTrue ( rebuilt_server . updated != self . server . updated , <EOL> msg = "<STR_LIT>" ) <EOL> self . assertEquals ( rebuilt_server . addresses , self . server . addresses , <EOL> msg = "<STR_LIT>" ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT>' ) <EOL> def test_rebuilt_server_addresses ( self ) : <EOL> """<STR_LIT>""" <EOL> rebuilt_server = self . rebuilt_server_response . entity <EOL> addresses = rebuilt_server . addresses <EOL> for name , ip_addresses in self . expected_networks . iteritems ( ) : <EOL> network = addresses . get_by_name ( name ) <EOL> if ip_addresses . get ( '<STR_LIT>' ) : <EOL> self . assertIsNotNone ( <EOL> network . ipv4 , <EOL> msg = '<STR_LIT>' <EOL> . format ( name = name ) ) <EOL> else : <EOL> self . assertIsNone ( <EOL> network . ipv4 , <EOL> msg = '<STR_LIT>' <EOL> . format ( name = name ) ) <EOL> if ip_addresses . get ( '<STR_LIT>' ) : <EOL> self . assertIsNotNone ( <EOL> network . ipv6 , <EOL> msg = '<STR_LIT>' <EOL> . format ( name = name ) ) <EOL> else : <EOL> self . assertIsNone ( <EOL> network . ipv6 , <EOL> msg = '<STR_LIT>' <EOL> . format ( name = name ) ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT>' ) <EOL> def test_address_not_changed_on_rebuild ( self ) : <EOL> """<STR_LIT>""" <EOL> rebuilt_server = self . rebuilt_server_response . entity <EOL> addresses = rebuilt_server . addresses <EOL> self . assertEqual ( addresses , self . server . addresses ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> def test_can_log_into_server_after_rebuild ( self ) : <EOL> """<STR_LIT>""" <EOL> remote_instance = self . server_behaviors . get_remote_instance_client ( <EOL> self . server , config = self . servers_config , password = self . password , <EOL> key = self . key . private_key ) <EOL> self . assertTrue ( remote_instance . can_authenticate ( ) ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> def test_rebuilt_server_vcpus ( self ) : <EOL> """<STR_LIT>""" <EOL> remote_client = self . server_behaviors . get_remote_instance_client ( <EOL> self . server , config = self . servers_config , password = self . password , <EOL> key = self . key . private_key ) <EOL> server_actual_vcpus = remote_client . get_number_of_cpus ( ) <EOL> self . assertEqual ( <EOL> server_actual_vcpus , self . flavor . vcpus , <EOL> msg = "<STR_LIT>" . format ( <EOL> self . flavor . vcpus , server_actual_vcpus ) ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> def test_rebuilt_server_ephemeral_disk ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . flavor . ephemeral_disk == <NUM_LIT:0> : <EOL> return <EOL> remote_client = self . server_behaviors . get_remote_instance_client ( <EOL> self . server , self . servers_config , password = self . password , <EOL> key = self . key . private_key ) <EOL> disks = remote_client . get_all_disks ( ) <EOL> disks . pop ( self . servers_config . instance_disk_path , None ) <EOL> self . _verify_ephemeral_disk_size ( <EOL> disks = disks , flavor = self . flavor , <EOL> split_ephemeral_disk_enabled = self . split_ephemeral_disk_enabled , <EOL> ephemeral_disk_max_size = self . ephemeral_disk_max_size ) <EOL> for disk in disks : <EOL> self . _format_disk ( remote_client = remote_client , disk = disk , <EOL> disk_format = self . disk_format_type ) <EOL> mount_point = remote_client . generate_mountpoint ( ) <EOL> self . _mount_disk ( remote_client = remote_client , disk = disk , <EOL> mount_point = mount_point ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> def test_server_ram_after_rebuild ( self ) : <EOL> """<STR_LIT>""" <EOL> remote_instance = self . server_behaviors . get_remote_instance_client ( <EOL> self . server , self . servers_config , password = self . password , <EOL> key = self . key . private_key ) <EOL> lower_limit = int ( self . flavor . ram ) - ( int ( self . flavor . ram ) * <NUM_LIT> ) <EOL> server_ram_size = int ( remote_instance . get_allocated_ram ( ) ) <EOL> self . assertTrue ( ( int ( self . flavor . ram ) == server_ram_size <EOL> or lower_limit <= server_ram_size ) , <EOL> msg = '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> ( self . flavor . ram , server_ram_size ) ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> @ unittest . skipUnless ( file_injection_enabled , "<STR_LIT>" ) <EOL> def test_personality_file_created_on_rebuild ( self ) : <EOL> """<STR_LIT>""" <EOL> remote_client = self . server_behaviors . get_remote_instance_client ( <EOL> self . server , self . servers_config , password = self . password , <EOL> key = self . key . private_key ) <EOL> self . assertTrue ( remote_client . is_file_present ( self . rebuilt_file_path ) ) <EOL> self . assertEqual ( <EOL> remote_client . get_file_details ( self . rebuilt_file_path ) . content , <EOL> self . rebuilt_file_contents ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT>' ) <EOL> def test_server_metadata_set_on_rebuild ( self ) : <EOL> """<STR_LIT>""" <EOL> rebuilt_server = self . rebuilt_server_response . entity <EOL> self . assertIn ( '<STR_LIT:key>' , rebuilt_server . metadata ) <EOL> self . assertEqual ( rebuilt_server . metadata . get ( '<STR_LIT:key>' ) , '<STR_LIT:value>' ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT>' ) <EOL> def test_rebuilt_server_instance_actions ( self ) : <EOL> """<STR_LIT>""" <EOL> actions = self . servers_client . get_instance_actions ( <EOL> self . server . id ) . entity <EOL> self . assertTrue ( any ( a . action == '<STR_LIT>' for a in actions ) ) <EOL> filtered_actions = [ a for a in actions if a . action == '<STR_LIT>' ] <EOL> self . assertEquals ( len ( filtered_actions ) , <NUM_LIT:1> ) <EOL> rebuild_action = filtered_actions [ <NUM_LIT:0> ] <EOL> self . validate_instance_action ( <EOL> rebuild_action , self . server . id , self . compute . user . user_id , <EOL> self . compute . user . project_id , <EOL> self . rebuilt_server_response . headers [ '<STR_LIT>' ] ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> @ unittest . skipUnless ( <EOL> hypervisor == ComputeHypervisors . XEN_SERVER , <EOL> '<STR_LIT>' ) <EOL> def test_rebuilt_server_xenstore_metadata ( self ) : <EOL> """<STR_LIT>""" <EOL> remote_client = self . server_behaviors . get_remote_instance_client ( <EOL> self . server , self . servers_config , password = self . password , <EOL> key = self . key . private_key ) <EOL> xen_meta = remote_client . get_xen_user_metadata ( ) <EOL> for key , value in self . metadata . iteritems ( ) : <EOL> self . assertIn ( key , xen_meta ) <EOL> self . assertEqual ( xen_meta [ key ] , value ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> @ unittest . skipUnless ( <EOL> hypervisor == ComputeHypervisors . XEN_SERVER , <EOL> '<STR_LIT>' ) <EOL> def test_xenstore_disk_config_on_rebuild ( self ) : <EOL> """<STR_LIT>""" <EOL> rebuilt_server = self . rebuilt_server_response . entity <EOL> remote_client = self . server_behaviors . get_remote_instance_client ( <EOL> self . server , self . servers_config , key = self . key . private_key , <EOL> password = self . password ) <EOL> auto_config_enabled = remote_client . get_xenstore_disk_config_value ( ) <EOL> actual_disk_config = rebuilt_server . disk_config <EOL> self . assertEqual ( auto_config_enabled , <EOL> actual_disk_config . lower ( ) == '<STR_LIT>' ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> @ unittest . skipUnless ( file_injection_enabled , "<STR_LIT>" ) <EOL> def test_file_system_cleared_on_rebuild ( self ) : <EOL> """<STR_LIT>""" <EOL> remote_client = self . server_behaviors . get_remote_instance_client ( <EOL> self . server , self . servers_config , password = self . password , <EOL> key = self . key . private_key ) <EOL> self . assertFalse ( remote_client . is_file_present ( self . file_path ) ) <EOL> @ unittest . skip ( "<STR_LIT>" ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> def test_distro_after_rebuild ( self ) : <EOL> """<STR_LIT>""" <EOL> remote_client = self . server_behaviors . get_remote_instance_client ( <EOL> self . server , self . servers_config , password = self . password , <EOL> key = self . key . private_key ) <EOL> distro_after_rebuild = remote_client . get_distribution_and_version ( ) <EOL> if ( self . distro_before_rebuild and <EOL> distro_after_rebuild and <EOL> self . image_ref != self . image_ref_alt ) : <EOL> self . assertNotEqual ( self . distro_before_rebuild , <EOL> distro_after_rebuild ) <EOL> else : <EOL> self . assertEqual ( self . distro_before_rebuild , distro_after_rebuild ) <EOL> class RebuildBaseFixture ( object ) : <EOL> servers_config = ServersConfig ( ) <EOL> file_injection_enabled = servers_config . personality_file_injection_enabled <EOL> @ classmethod <EOL> def rebuild_and_await ( cls ) : <EOL> """<STR_LIT>""" <EOL> cls . metadata = { '<STR_LIT:key>' : '<STR_LIT:value>' } <EOL> cls . name = rand_name ( '<STR_LIT>' ) <EOL> personality = cls . server_behaviors . get_default_injected_files ( ) <EOL> if cls . file_injection_enabled : <EOL> separator = cls . images_config . primary_image_path_separator <EOL> cls . rebuilt_file_path = separator . join ( <EOL> [ cls . servers_config . default_file_path , '<STR_LIT>' ] ) <EOL> cls . rebuilt_file_contents = '<STR_LIT>' <EOL> if personality is None : <EOL> personality = [ ] <EOL> personality . extend ( [ { '<STR_LIT:path>' : cls . rebuilt_file_path , <EOL> '<STR_LIT>' : base64 . b64encode ( <EOL> cls . rebuilt_file_contents ) } ] ) <EOL> cls . password = '<STR_LIT>' <EOL> security_groups = None <EOL> if cls . security_groups_config . default_security_group : <EOL> security_groups = [ <EOL> { "<STR_LIT:name>" : cls . security_groups_config . default_security_group } ] <EOL> cls . rebuilt_server_response = cls . servers_client . rebuild ( <EOL> cls . server . id , cls . image_ref_alt , name = cls . name , <EOL> metadata = cls . metadata , personality = personality , <EOL> admin_pass = cls . password , key_name = cls . key . name , <EOL> security_groups = security_groups ) <EOL> cls . server_behaviors . wait_for_server_status ( <EOL> cls . server . id , NovaServerStatusTypes . ACTIVE ) <EOL> class ServerFromImageRebuildTests ( ServerFromImageFixture , <EOL> RebuildServerTests , <EOL> RebuildBaseFixture ) : <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> """<STR_LIT>""" <EOL> super ( ServerFromImageRebuildTests , cls ) . setUpClass ( ) <EOL> cls . key = cls . keypairs_client . create_keypair ( rand_name ( "<STR_LIT:key>" ) ) . entity <EOL> cls . resources . add ( cls . key . name , <EOL> cls . keypairs_client . delete_keypair ) <EOL> personality = cls . server_behaviors . get_default_injected_files ( ) <EOL> if cls . file_injection_enabled : <EOL> separator = cls . images_config . primary_image_path_separator <EOL> cls . file_path = separator . join ( <EOL> [ cls . servers_config . default_file_path , '<STR_LIT>' ] ) <EOL> cls . file_contents = '<STR_LIT>' <EOL> if personality is None : <EOL> personality = [ ] <EOL> personality . extend ( [ { '<STR_LIT:path>' : cls . file_path , <EOL> '<STR_LIT>' : base64 . b64encode ( <EOL> cls . file_contents ) } ] ) <EOL> cls . create_resp = cls . server_behaviors . create_active_server ( <EOL> personality = personality , key_name = cls . key . name ) <EOL> cls . server = cls . create_resp . entity <EOL> cls . resources . add ( cls . server . id , cls . servers_client . delete_server ) <EOL> cls . flavor = cls . flavors_client . get_flavor_details ( <EOL> cls . flavor_ref ) . entity <EOL> cls . rebuild_and_await ( ) <EOL> @ tags ( type = '<STR_LIT>' , net = '<STR_LIT:yes>' ) <EOL> def test_rebuilt_server_disk_size ( self ) : <EOL> """<STR_LIT>""" <EOL> remote_client = self . server_behaviors . get_remote_instance_client ( <EOL> self . server , self . servers_config , password = self . password , <EOL> key = self . key . private_key ) <EOL> disk_size = remote_client . get_disk_size ( <EOL> self . servers_config . instance_disk_path ) <EOL> self . assertEqual ( disk_size , self . flavor . disk , <EOL> msg = "<STR_LIT>" . format ( <EOL> self . flavor . disk , disk_size ) ) </s>
