<s> import logging <EOL> import os <EOL> import unittest <EOL> from keystoneclient import exceptions as ks_exceptions <EOL> from testconfig import config as config_ini <EOL> import cloudferry_devlab . tests . config as config <EOL> from cloudferry_devlab import generate_load <EOL> from cloudferry_devlab . tests import test_exceptions <EOL> import cloudferry_devlab . tests . utils as utils <EOL> def suppress_dependency_logging ( ) : <EOL> suppressed_logs = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> for l in suppressed_logs : <EOL> logging . getLogger ( l ) . setLevel ( logging . WARNING ) <EOL> def get_option_from_config_ini ( option , section = '<STR_LIT>' ) : <EOL> return config_ini . get ( section , { } ) . get ( option , '<STR_LIT:False>' ) <EOL> class FunctionalTest ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( FunctionalTest , self ) . setUp ( ) <EOL> suppress_dependency_logging ( ) <EOL> if not config_ini : <EOL> raise test_exceptions . ConfFileError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . src_cloud = generate_load . Prerequisites ( <EOL> cloud_prefix = '<STR_LIT>' , <EOL> configuration_ini = config_ini , <EOL> config = config ) <EOL> self . dst_cloud = generate_load . Prerequisites ( <EOL> cloud_prefix = '<STR_LIT>' , <EOL> configuration_ini = config_ini , <EOL> config = config ) <EOL> self . migration_utils = utils . MigrationUtils ( config ) <EOL> self . config_ini_path = config_ini [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . cloudferry_dir = config_ini [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . filtering_utils = utils . FilteringUtils ( <EOL> os . path . join ( self . cloudferry_dir , get_option_from_config_ini ( <EOL> '<STR_LIT>' ) ) ) <EOL> def filter_networks ( self ) : <EOL> networks = [ i [ '<STR_LIT:name>' ] for i in config . networks ] <EOL> for i in config . tenants : <EOL> if '<STR_LIT>' in i and not i . get ( '<STR_LIT>' ) : <EOL> for j in i [ '<STR_LIT>' ] : <EOL> networks . append ( j [ '<STR_LIT:name>' ] ) <EOL> return self . _get_neutron_resources ( '<STR_LIT>' , networks ) <EOL> def filter_subnets ( self ) : <EOL> subnets = [ ] <EOL> admin_tenant_id = self . src_cloud . get_tenant_id ( self . src_cloud . tenant ) <EOL> for net in config . networks : <EOL> if not net . get ( '<STR_LIT>' ) : <EOL> continue <EOL> for subnet in net [ '<STR_LIT>' ] : <EOL> subnet [ '<STR_LIT>' ] = admin_tenant_id <EOL> subnets . append ( subnet ) <EOL> subnets = [ i for net in config . networks if net . get ( '<STR_LIT>' ) <EOL> for i in net [ '<STR_LIT>' ] ] <EOL> for tenant in config . tenants : <EOL> if '<STR_LIT>' not in tenant or tenant . get ( '<STR_LIT>' ) : <EOL> continue <EOL> for network in tenant [ '<STR_LIT>' ] : <EOL> if '<STR_LIT>' not in network : <EOL> continue <EOL> for subnet in network [ '<STR_LIT>' ] : <EOL> subnet [ '<STR_LIT>' ] = self . src_cloud . get_tenant_id ( <EOL> tenant [ '<STR_LIT:name>' ] ) <EOL> subnets . append ( subnet ) <EOL> env_subnets = self . src_cloud . neutronclient . list_subnets ( ) [ '<STR_LIT>' ] <EOL> filtered_subnets = { '<STR_LIT>' : [ ] } <EOL> for env_subnet in env_subnets : <EOL> for subnet in subnets : <EOL> same_cidr = env_subnet [ '<STR_LIT>' ] == subnet [ '<STR_LIT>' ] <EOL> same_tenant = env_subnet [ '<STR_LIT>' ] == subnet [ '<STR_LIT>' ] <EOL> if same_cidr and same_tenant : <EOL> filtered_subnets [ '<STR_LIT>' ] . append ( env_subnet ) <EOL> return filtered_subnets <EOL> def filter_routers ( self ) : <EOL> routers = [ i [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] for i in config . routers ] <EOL> for tenant in config . tenants : <EOL> if tenant . get ( '<STR_LIT>' ) : <EOL> for router in tenant . get ( '<STR_LIT>' ) : <EOL> routers . append ( router [ '<STR_LIT>' ] [ '<STR_LIT:name>' ] ) <EOL> return self . _get_neutron_resources ( '<STR_LIT>' , routers ) <EOL> def filter_floatingips ( self ) : <EOL> def get_fips ( _user ) : <EOL> self . src_cloud . switch_user ( user = _user [ '<STR_LIT:name>' ] , <EOL> tenant = _user [ '<STR_LIT>' ] , <EOL> password = _user [ '<STR_LIT:password>' ] ) <EOL> _client = self . src_cloud . neutronclient <EOL> return [ _fip [ '<STR_LIT>' ] <EOL> for _fip in _client . list_floatingips ( ) [ '<STR_LIT>' ] ] <EOL> for tenant in config . tenants : <EOL> fips = [ fip for user in config . users <EOL> if tenant [ '<STR_LIT:name>' ] == user . get ( '<STR_LIT>' ) and <EOL> user [ '<STR_LIT>' ] and not user . get ( '<STR_LIT>' ) <EOL> for fip in get_fips ( user ) ] <EOL> return set ( fips ) <EOL> def filter_users ( self ) : <EOL> users = [ ] <EOL> for user in config . users : <EOL> if user . get ( '<STR_LIT>' ) : <EOL> continue <EOL> if self . src_cloud . tenant_exists ( user . get ( '<STR_LIT>' ) ) or self . src_cloud . user_has_not_primary_tenants ( user [ '<STR_LIT:name>' ] ) : <EOL> users . append ( user [ '<STR_LIT:name>' ] ) <EOL> return self . _get_keystone_resources ( '<STR_LIT>' , users ) <EOL> def filter_tenants ( self ) : <EOL> tenants = [ i [ '<STR_LIT:name>' ] for i in config . tenants ] <EOL> return self . _get_keystone_resources ( '<STR_LIT>' , tenants ) <EOL> def filter_roles ( self ) : <EOL> roles = [ i [ '<STR_LIT:name>' ] for i in config . roles ] <EOL> return self . _get_keystone_resources ( '<STR_LIT>' , roles ) <EOL> def get_src_vm_objects_specified_in_config ( self ) : <EOL> vms = self . migration_utils . get_all_vms_from_config ( ) <EOL> vms_names = [ vm [ '<STR_LIT:name>' ] for vm in vms if not vm . get ( '<STR_LIT>' ) ] <EOL> opts = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:1> } } <EOL> return [ i for i in self . src_cloud . novaclient . servers . list ( ** opts ) <EOL> if i . name in vms_names ] <EOL> def filter_flavors ( self , filter_only_private = False ) : <EOL> flavors = [ ] <EOL> if filter_only_private : <EOL> nova_args = { '<STR_LIT>' : None } <EOL> else : <EOL> nova_args = None <EOL> all_flavors = config . flavors <EOL> for tenant in config . tenants : <EOL> if tenant . get ( '<STR_LIT>' ) : <EOL> all_flavors += [ flavor for flavor in tenant [ '<STR_LIT>' ] ] <EOL> for flavor in all_flavors : <EOL> if filter_only_private : <EOL> if flavor . get ( '<STR_LIT>' ) is False : <EOL> flavors . append ( flavor [ '<STR_LIT:name>' ] ) <EOL> elif '<STR_LIT>' not in flavor or flavor . get ( '<STR_LIT>' ) : <EOL> flavors . append ( flavor [ '<STR_LIT:name>' ] ) <EOL> return self . _get_nova_resources ( '<STR_LIT>' , flavors , nova_args ) <EOL> def filter_keypairs ( self ) : <EOL> return self . src_cloud . get_users_keypairs ( ) <EOL> def filter_security_groups ( self ) : <EOL> sgs = [ sg [ '<STR_LIT:name>' ] for i in config . tenants if '<STR_LIT>' in i <EOL> for sg in i [ '<STR_LIT>' ] ] <EOL> return self . _get_neutron_resources ( '<STR_LIT>' , sgs ) <EOL> def filter_images ( self , exclude_images_with_fields = None ) : <EOL> if exclude_images_with_fields is None : <EOL> exclude_images_with_fields = { } <EOL> if exclude_images_with_fields . get ( '<STR_LIT>' ) is None : <EOL> exclude_images_with_fields [ '<STR_LIT>' ] = True <EOL> if exclude_images_with_fields . get ( '<STR_LIT>' ) is None : <EOL> exclude_images_with_fields [ '<STR_LIT>' ] = True <EOL> def _image_exclude_filter ( images ) : <EOL> filtered_images_name = [ ] <EOL> for image in images : <EOL> for key , value in exclude_images_with_fields . iteritems ( ) : <EOL> if key in image and image [ key ] == value : <EOL> break <EOL> else : <EOL> filtered_images_name . append ( image [ '<STR_LIT:name>' ] ) <EOL> return filtered_images_name <EOL> all_images = self . migration_utils . get_all_images_from_config ( ) <EOL> filtered_images = _image_exclude_filter ( all_images ) <EOL> image_list = self . src_cloud . glanceclient . images . list ( is_public = None ) <EOL> return [ i for i in image_list if i . name in filtered_images ] <EOL> def filter_volumes ( self ) : <EOL> volumes = config . cinder_volumes <EOL> for tenant in config . tenants : <EOL> if '<STR_LIT>' in tenant and not tenant . get ( '<STR_LIT>' ) : <EOL> volumes . extend ( tenant [ '<STR_LIT>' ] ) <EOL> volumes . extend ( config . cinder_volumes_from_images ) <EOL> volumes_names = [ volume . get ( '<STR_LIT>' ) for volume in volumes ] <EOL> opts = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:1> } } <EOL> return [ i for i in self . src_cloud . cinderclient . volumes . list ( ** opts ) <EOL> if i . display_name in volumes_names ] <EOL> def filter_health_monitors ( self ) : <EOL> hm = self . src_cloud . neutronclient . list_health_monitors ( ) <EOL> final_hm = [ m for m in hm [ '<STR_LIT>' ] <EOL> if self . src_cloud . tenant_exists ( tenant_id = m [ '<STR_LIT>' ] ) ] <EOL> return { '<STR_LIT>' : final_hm } <EOL> def filter_pools ( self ) : <EOL> pools = self . src_cloud . neutronclient . list_pools ( ) [ '<STR_LIT>' ] <EOL> final_p = [ p for p in pools <EOL> if self . src_cloud . tenant_exists ( tenant_id = p [ '<STR_LIT>' ] ) ] <EOL> return { '<STR_LIT>' : final_p } <EOL> def filter_lbaas_members ( self ) : <EOL> members = self . src_cloud . neutronclient . list_members ( ) [ '<STR_LIT>' ] <EOL> final_m = [ m for m in members <EOL> if self . src_cloud . tenant_exists ( tenant_id = m [ '<STR_LIT>' ] ) ] <EOL> return { '<STR_LIT>' : final_m } <EOL> def filter_vips ( self ) : <EOL> vips = self . src_cloud . neutronclient . list_vips ( ) [ '<STR_LIT>' ] <EOL> final_v = [ vip for vip in vips <EOL> if self . src_cloud . tenant_exists ( tenant_id = vip [ '<STR_LIT>' ] ) ] <EOL> return { '<STR_LIT>' : final_v } <EOL> def _get_neutron_resources ( self , res , names ) : <EOL> _list = getattr ( self . src_cloud . neutronclient , '<STR_LIT>' + res ) ( ) <EOL> return { res : [ i for i in _list [ res ] if i [ '<STR_LIT:name>' ] in names ] } <EOL> def _get_nova_resources ( self , res , names , args = None ) : <EOL> client = getattr ( self . src_cloud . novaclient , res ) <EOL> if args : <EOL> return [ i for i in client . list ( ** args ) if i . name in names ] <EOL> else : <EOL> return [ i for i in client . list ( ) if i . name in names ] <EOL> def _get_keystone_resources ( self , res , names ) : <EOL> client = getattr ( self . src_cloud . keystoneclient , res ) <EOL> return [ i for i in client . list ( ) <EOL> if i . name in names ] <EOL> def get_vms_with_fip_associated ( self ) : <EOL> vms = config . vms <EOL> for tenant in config . tenants : <EOL> if '<STR_LIT>' in tenant : <EOL> vms . extend ( tenant [ '<STR_LIT>' ] ) <EOL> return [ vm [ '<STR_LIT:name>' ] for vm in vms if vm . get ( '<STR_LIT>' ) ] <EOL> def tenant_exists ( self , keystone_client , tenant_id ) : <EOL> try : <EOL> keystone_client . tenants . get ( tenant_id ) <EOL> except ks_exceptions . NotFound : <EOL> return False <EOL> return True <EOL> def set_hash_for_vms ( self , vm_list ) : <EOL> for vm in vm_list : <EOL> nics = None <EOL> for net in vm . addresses : <EOL> nics = [ ( net , ip [ '<STR_LIT>' ] ) for ip in vm . addresses . get ( <EOL> net ) if ip [ '<STR_LIT>' ] == '<STR_LIT>' ] <EOL> setattr ( vm , '<STR_LIT>' , ( vm . name , nics ) ) </s>
