<s> """<STR_LIT>""" <EOL> from IPy import IP <EOL> import os <EOL> import time <EOL> from cafe . engine . config import EngineConfig <EOL> from cloudcafe . common . tools . datagen import rand_name <EOL> from cloudcafe . compute . common . types import InstanceAuthStrategies <EOL> PING_PACKET_LOSS_REGEX = '<STR_LIT>' <EOL> class Instance ( object ) : <EOL> def __init__ ( self , entity , isolated_ips , remote_client = None ) : <EOL> self . entity = entity <EOL> for network_name , ip in isolated_ips . items ( ) : <EOL> setattr ( self , network_name , ip ) <EOL> self . remote_client = remote_client <EOL> class ScenarioMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> NAMES_PREFIX = '<STR_LIT>' <EOL> @ classmethod <EOL> def _create_network_with_subnet ( cls , name , cidr , allocation_pools = None , <EOL> gateway_ip = None ) : <EOL> create_name = '<STR_LIT>' . format ( rand_name ( cls . NAMES_PREFIX ) , name ) <EOL> network = cls . net . behaviors . networks_behaviors . create_network ( <EOL> name = create_name , <EOL> use_exact_name = True ) . response . entity <EOL> cls . delete_networks . append ( network . id ) <EOL> subnet = cls . net . behaviors . subnets_behaviors . create_subnet ( <EOL> network . id , name = create_name , ip_version = cls . ip_version , <EOL> cidr = cidr , allocation_pools = allocation_pools , <EOL> gateway_ip = gateway_ip , use_exact_name = True ) . response . entity <EOL> cls . delete_subnets . append ( subnet . id ) <EOL> return network , subnet <EOL> @ classmethod <EOL> def _create_server ( cls , name , isolated_networks_to_connect = None , <EOL> public_and_service = True , security_groups = None , <EOL> scheduler_hints = None ) : <EOL> isolated_networks_to_connect = isolated_networks_to_connect or [ ] <EOL> security_groups = security_groups or [ ] <EOL> networks = [ { '<STR_LIT>' : net . id } for net in isolated_networks_to_connect ] <EOL> if public_and_service : <EOL> networks . append ( { '<STR_LIT>' : cls . public_network_id } ) <EOL> networks . append ( { '<STR_LIT>' : cls . service_network_id } ) <EOL> nova_secgroups = [ { "<STR_LIT:name>" : secgroup . name } for secgroup in <EOL> security_groups ] <EOL> server = cls . compute . servers . behaviors . create_active_server ( <EOL> name = '<STR_LIT>' . format ( rand_name ( cls . NAMES_PREFIX ) , name ) , <EOL> key_name = cls . keypair . name , networks = networks , <EOL> security_groups = nova_secgroups , <EOL> scheduler_hints = scheduler_hints ) . entity <EOL> cls . delete_servers . append ( server . id ) <EOL> isolated_ips = cls . _get_server_isolated_ips ( <EOL> server , isolated_networks_to_connect ) <EOL> return Instance ( server , isolated_ips ) <EOL> @ classmethod <EOL> def _get_server_isolated_ips ( cls , server , isolated_networks ) : <EOL> ips = { } <EOL> for net in isolated_networks : <EOL> ips [ net . name ] = getattr ( server . addresses , <EOL> net . name ) . addresses [ <NUM_LIT:0> ] . addr <EOL> return ips <EOL> @ classmethod <EOL> def _create_keypair ( cls ) : <EOL> name = rand_name ( cls . NAMES_PREFIX ) <EOL> cls . keypair = cls . compute . keypairs . client . create_keypair ( name ) . entity <EOL> cls . resources . add ( name , cls . compute . keypairs . client . delete_keypair ) <EOL> @ classmethod <EOL> def _get_remote_client ( cls , server ) : <EOL> ip_address = cls . compute . servers . behaviors . get_public_ip_address ( <EOL> server . entity ) <EOL> remote_client = cls . compute . servers . behaviors . get_remote_instance_client ( <EOL> server . entity , <EOL> ip_address = ip_address , username = '<STR_LIT:root>' , <EOL> key = cls . keypair . private_key , <EOL> auth_strategy = InstanceAuthStrategies . KEY ) <EOL> return remote_client <EOL> @ classmethod <EOL> def _transfer_private_key_to_vm ( cls , ssh_client , private_key , <EOL> remote_file_path ) : <EOL> pkey_file_path = os . path . join ( EngineConfig ( ) . temp_directory , '<STR_LIT>' ) <EOL> with open ( pkey_file_path , "<STR_LIT:w>" ) as private_key_file : <EOL> private_key_file . write ( private_key ) <EOL> ssh_client . transfer_file_to ( pkey_file_path , remote_file_path ) <EOL> error = ssh_client . execute_command ( <EOL> '<STR_LIT>' . format ( remote_file_path ) ) . stderr <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> assert not error , msg <EOL> @ classmethod <EOL> def _create_empty_security_group ( cls , name ) : <EOL> """<STR_LIT>""" <EOL> sg_name = '<STR_LIT>' . format ( rand_name ( cls . NAMES_PREFIX ) , name ) <EOL> sg_desc = sg_name + "<STR_LIT>" <EOL> resp = cls . security_groups . client . create_security_group ( <EOL> name = sg_name , description = sg_desc ) <EOL> msg = "<STR_LIT>" <EOL> msg = msg . format ( resp . status_code ) <EOL> assert resp . status_code == <NUM_LIT> , msg <EOL> secgroup = resp . entity <EOL> cls . delete_secgroups . append ( secgroup . id ) <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> msg = msg . format ( secgroup . name ) <EOL> assert secgroup . name == sg_name , msg <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> assert secgroup . description == sg_desc , msg <EOL> return secgroup <EOL> @ classmethod <EOL> def _get_port_id_owned_by_instance_on_net ( cls , instance_id , net_id ) : <EOL> """<STR_LIT>""" <EOL> resp = cls . net . ports . client . list_ports ( device_id = instance_id , <EOL> network_id = net_id ) <EOL> msg = "<STR_LIT>" <EOL> msg = msg . format ( resp . status_code ) <EOL> assert resp . status_code == <NUM_LIT:200> , msg <EOL> ports = resp . entity <EOL> msg = ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> msg = msg . format ( len ( ports ) ) <EOL> assert len ( ports ) == <NUM_LIT:1> , msg <EOL> return ports [ <NUM_LIT:0> ] . id <EOL> def _update_port_secgroups ( self , port_id , security_groups_ids ) : <EOL> resp = self . net . ports . client . update_port ( <EOL> port_id , security_groups = security_groups_ids ) <EOL> msg = "<STR_LIT>" <EOL> msg = msg . format ( resp . status_code ) <EOL> self . assertEqual ( resp . status_code , <NUM_LIT:200> , msg ) <EOL> msg = "<STR_LIT>" <EOL> msg = msg . format ( len ( resp . entity . security_groups ) ) <EOL> self . assertEqual ( len ( resp . entity . security_groups ) , <EOL> len ( security_groups_ids ) , msg ) <EOL> msg = "<STR_LIT>" <EOL> for secgroup_id in security_groups_ids : <EOL> self . assertIn ( secgroup_id , resp . entity . security_groups , <EOL> msg . format ( secgroup_id ) ) <EOL> time . sleep ( self . security_groups . config . data_plane_delay ) <EOL> def _next_sequential_cidr ( self , cidr ) : <EOL> """<STR_LIT>""" <EOL> next_cidr_1st_ip = cidr [ - <NUM_LIT:1> ] . ip + <NUM_LIT:1> <EOL> return IP ( '<STR_LIT>' . format ( str ( next_cidr_1st_ip ) , <EOL> str ( cidr . prefixlen ( ) ) ) ) <EOL> def _execute_ssh_command ( self , ssh_client , cmd ) : <EOL> response = ssh_client . execute_command ( cmd ) <EOL> if ( response . stderr and <EOL> ( '<STR_LIT>' not in response . stderr or <EOL> '<STR_LIT>' not in response . stderr ) ) : <EOL> msg = '<STR_LIT>' <EOL> msg = msg . format ( response . stderr ) <EOL> self . fail ( msg ) <EOL> return response . stdout </s>
