<s> from oslo_config import cfg <EOL> from oslo_serialization import jsonutils <EOL> from six . moves import range <EOL> from nova . compute import api as compute_api <EOL> from nova . compute import manager as compute_manager <EOL> import nova . context <EOL> from nova import db <EOL> from nova import exception <EOL> from nova . network import manager as network_manager <EOL> from nova . network import model as network_model <EOL> from nova . network import rpcapi as network_rpcapi <EOL> from nova import objects <EOL> from nova . objects import base as obj_base <EOL> from nova . objects import network as network_obj <EOL> from nova . objects import virtual_interface as vif_obj <EOL> from nova . tests . unit . objects import test_fixed_ip <EOL> from nova . tests . unit . objects import test_instance_info_cache <EOL> from nova . tests . unit . objects import test_pci_device <EOL> from nova . tests import uuidsentinel as uuids <EOL> HOST = "<STR_LIT>" <EOL> CONF = cfg . CONF <EOL> CONF . import_opt ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> class FakeModel ( dict ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> self . update ( kwargs ) <EOL> class FakeNetworkManager ( network_manager . NetworkManager ) : <EOL> """<STR_LIT>""" <EOL> class FakeDB ( object ) : <EOL> vifs = [ { '<STR_LIT:id>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : uuids . instance_1 , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : uuids . vifs_1 , <EOL> '<STR_LIT:address>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : uuids . instance_2 , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : uuids . vifs_2 , <EOL> '<STR_LIT:address>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : uuids . instance_1 , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : uuids . vifs_3 , <EOL> '<STR_LIT:address>' : '<STR_LIT>' } ] <EOL> floating_ips = [ dict ( address = '<STR_LIT>' , <EOL> fixed_ip_id = <NUM_LIT:100> ) , <EOL> dict ( address = '<STR_LIT>' , <EOL> fixed_ip_id = <NUM_LIT:200> ) , <EOL> dict ( address = '<STR_LIT>' , <EOL> fixed_ip_id = <NUM_LIT> ) ] <EOL> fixed_ips = [ dict ( test_fixed_ip . fake_fixed_ip , <EOL> id = <NUM_LIT:100> , <EOL> address = '<STR_LIT>' , <EOL> virtual_interface_id = <NUM_LIT:0> ) , <EOL> dict ( test_fixed_ip . fake_fixed_ip , <EOL> id = <NUM_LIT:200> , <EOL> address = '<STR_LIT>' , <EOL> virtual_interface_id = <NUM_LIT:1> ) , <EOL> dict ( test_fixed_ip . fake_fixed_ip , <EOL> id = <NUM_LIT> , <EOL> address = '<STR_LIT>' , <EOL> virtual_interface_id = <NUM_LIT:2> ) ] <EOL> def fixed_ip_get_by_instance ( self , context , instance_uuid ) : <EOL> return [ dict ( address = '<STR_LIT>' ) , dict ( address = '<STR_LIT>' ) , <EOL> dict ( address = '<STR_LIT>' ) ] <EOL> def network_get_by_cidr ( self , context , cidr ) : <EOL> raise exception . NetworkNotFoundForCidr ( cidr = cidr ) <EOL> def network_create_safe ( self , context , net ) : <EOL> fakenet = dict ( net ) <EOL> fakenet [ '<STR_LIT:id>' ] = <NUM_LIT> <EOL> return fakenet <EOL> def network_get ( self , context , network_id , project_only = "<STR_LIT>" ) : <EOL> return { '<STR_LIT>' : '<STR_LIT>' % network_id } <EOL> def network_get_by_uuid ( self , context , network_uuid ) : <EOL> raise exception . NetworkNotFoundForUUID ( uuid = network_uuid ) <EOL> def network_get_all ( self , context ) : <EOL> raise exception . NoNetworksFound ( ) <EOL> def network_get_all_by_uuids ( self , context , project_only = "<STR_LIT>" ) : <EOL> raise exception . NoNetworksFound ( ) <EOL> def network_disassociate ( self , context , network_id ) : <EOL> return True <EOL> def virtual_interface_get_all ( self , context ) : <EOL> return self . vifs <EOL> def fixed_ips_by_virtual_interface ( self , context , vif_id ) : <EOL> return [ ip for ip in self . fixed_ips <EOL> if ip [ '<STR_LIT>' ] == vif_id ] <EOL> def fixed_ip_disassociate ( self , context , address ) : <EOL> return True <EOL> def __init__ ( self , stubs = None ) : <EOL> self . db = self . FakeDB ( ) <EOL> if stubs : <EOL> stubs . Set ( vif_obj , '<STR_LIT>' , self . db ) <EOL> self . deallocate_called = None <EOL> self . deallocate_fixed_ip_calls = [ ] <EOL> self . network_rpcapi = network_rpcapi . NetworkAPI ( ) <EOL> def deallocate_fixed_ip ( self , context , address = None , host = None , <EOL> instance = None ) : <EOL> self . deallocate_fixed_ip_calls . append ( ( context , address , host ) ) <EOL> self . deallocate_called = address <EOL> def _create_fixed_ips ( self , context , network_id , fixed_cidr = None , <EOL> extra_reserved = None , bottom_reserved = <NUM_LIT:0> , <EOL> top_reserved = <NUM_LIT:0> ) : <EOL> pass <EOL> def get_instance_nw_info ( context , instance_id , rxtx_factor , <EOL> host , instance_uuid = None , ** kwargs ) : <EOL> pass <EOL> def fake_network ( network_id , ipv6 = None ) : <EOL> if ipv6 is None : <EOL> ipv6 = CONF . use_ipv6 <EOL> fake_network = { '<STR_LIT:id>' : network_id , <EOL> '<STR_LIT>' : getattr ( uuids , '<STR_LIT>' % network_id ) , <EOL> '<STR_LIT:label>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT:host>' : None , <EOL> '<STR_LIT>' : uuids . project , <EOL> '<STR_LIT>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : network_id * <NUM_LIT:10> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' % network_id , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False } <EOL> if ipv6 : <EOL> fake_network [ '<STR_LIT>' ] = '<STR_LIT>' % network_id <EOL> fake_network [ '<STR_LIT>' ] = '<STR_LIT>' % network_id <EOL> fake_network [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if CONF . flat_injected : <EOL> fake_network [ '<STR_LIT>' ] = True <EOL> return fake_network <EOL> def fake_network_obj ( context , network_id = <NUM_LIT:1> , ipv6 = None ) : <EOL> return network_obj . Network . _from_db_object ( <EOL> context , network_obj . Network ( ) , fake_network ( network_id , ipv6 ) ) <EOL> def fake_vif ( x ) : <EOL> return { '<STR_LIT:id>' : x , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT:address>' : '<STR_LIT>' % x , <EOL> '<STR_LIT>' : getattr ( uuids , '<STR_LIT>' % x ) , <EOL> '<STR_LIT>' : x , <EOL> '<STR_LIT>' : uuids . vifs_1 } <EOL> def floating_ip_ids ( ) : <EOL> for i in range ( <NUM_LIT:1> , <NUM_LIT:100> ) : <EOL> yield i <EOL> def fixed_ip_ids ( ) : <EOL> for i in range ( <NUM_LIT:1> , <NUM_LIT:100> ) : <EOL> yield i <EOL> floating_ip_id = floating_ip_ids ( ) <EOL> fixed_ip_id = fixed_ip_ids ( ) <EOL> def next_fixed_ip ( network_id , num_floating_ips = <NUM_LIT:0> ) : <EOL> next_id = next ( fixed_ip_id ) <EOL> f_ips = [ FakeModel ( ** next_floating_ip ( next_id ) ) <EOL> for i in range ( num_floating_ips ) ] <EOL> return { '<STR_LIT:id>' : next_id , <EOL> '<STR_LIT>' : network_id , <EOL> '<STR_LIT:address>' : '<STR_LIT>' % ( network_id , ( next_id + <NUM_LIT> ) ) , <EOL> '<STR_LIT>' : uuids . fixed_ip , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT:host>' : HOST , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : fake_network ( network_id ) , <EOL> '<STR_LIT>' : fake_vif ( network_id ) , <EOL> '<STR_LIT>' : network_id , <EOL> '<STR_LIT>' : f_ips } <EOL> def next_floating_ip ( fixed_ip_id ) : <EOL> next_id = next ( floating_ip_id ) <EOL> return { '<STR_LIT:id>' : next_id , <EOL> '<STR_LIT:address>' : '<STR_LIT>' % ( next_id + <NUM_LIT> ) , <EOL> '<STR_LIT>' : fixed_ip_id , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : False } <EOL> def ipv4_like ( ip , match_string ) : <EOL> ip = ip . split ( '<STR_LIT:.>' ) <EOL> match_octets = match_string . split ( '<STR_LIT:.>' ) <EOL> for i , octet in enumerate ( match_octets ) : <EOL> if octet == '<STR_LIT:*>' : <EOL> continue <EOL> if octet != ip [ i ] : <EOL> return False <EOL> return True <EOL> def fake_get_instance_nw_info ( test , num_networks = <NUM_LIT:1> , ips_per_vif = <NUM_LIT:2> , <EOL> floating_ips_per_fixed_ip = <NUM_LIT:0> ) : <EOL> network = network_manager . FlatManager ( host = HOST ) <EOL> network . db = db <EOL> global floating_ip_id , fixed_ip_id , fixed_ips <EOL> floating_ip_id = floating_ip_ids ( ) <EOL> fixed_ip_id = fixed_ip_ids ( ) <EOL> fixed_ips = [ ] <EOL> def fixed_ips_fake ( * args , ** kwargs ) : <EOL> global fixed_ips <EOL> ips = [ next_fixed_ip ( i , floating_ips_per_fixed_ip ) <EOL> for i in range ( <NUM_LIT:1> , num_networks + <NUM_LIT:1> ) <EOL> for j in range ( ips_per_vif ) ] <EOL> fixed_ips = ips <EOL> return ips <EOL> def update_cache_fake ( * args , ** kwargs ) : <EOL> fake_info_cache = { <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : uuids . vifs_1 , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> return fake_info_cache <EOL> test . stub_out ( '<STR_LIT>' , fixed_ips_fake ) <EOL> test . stub_out ( '<STR_LIT>' , update_cache_fake ) <EOL> class FakeContext ( nova . context . RequestContext ) : <EOL> def is_admin ( self ) : <EOL> return True <EOL> nw_model = network . get_instance_nw_info ( <EOL> FakeContext ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> <NUM_LIT:0> , <NUM_LIT:3> , None ) <EOL> return nw_model <EOL> def stub_out_nw_api_get_instance_nw_info ( test , func = None , <EOL> num_networks = <NUM_LIT:1> , <EOL> ips_per_vif = <NUM_LIT:1> , <EOL> floating_ips_per_fixed_ip = <NUM_LIT:0> ) : <EOL> def get_instance_nw_info ( self , context , instance , conductor_api = None ) : <EOL> return fake_get_instance_nw_info ( test , num_networks = num_networks , <EOL> ips_per_vif = ips_per_vif , <EOL> floating_ips_per_fixed_ip = floating_ips_per_fixed_ip ) <EOL> if func is None : <EOL> func = get_instance_nw_info <EOL> test . stub_out ( '<STR_LIT>' , func ) <EOL> def stub_out_network_cleanup ( test ) : <EOL> test . stub_out ( '<STR_LIT>' , <EOL> lambda * args , ** kwargs : None ) <EOL> _real_functions = { } <EOL> def set_stub_network_methods ( test ) : <EOL> global _real_functions <EOL> cm = compute_manager . ComputeManager <EOL> if not _real_functions : <EOL> _real_functions = { <EOL> '<STR_LIT>' : cm . _allocate_network , <EOL> '<STR_LIT>' : cm . _deallocate_network } <EOL> def fake_networkinfo ( * args , ** kwargs ) : <EOL> return network_model . NetworkInfo ( ) <EOL> def fake_async_networkinfo ( * args , ** kwargs ) : <EOL> return network_model . NetworkInfoAsyncWrapper ( fake_networkinfo ) <EOL> test . stub_out ( '<STR_LIT>' , <EOL> fake_async_networkinfo ) <EOL> test . stub_out ( '<STR_LIT>' , <EOL> lambda * args , ** kwargs : None ) <EOL> def unset_stub_network_methods ( test ) : <EOL> global _real_functions <EOL> if _real_functions : <EOL> for name in _real_functions : <EOL> test . stub_out ( '<STR_LIT>' + name , <EOL> _real_functions [ name ] ) <EOL> def stub_compute_with_ips ( stubs ) : <EOL> orig_get = compute_api . API . get <EOL> orig_get_all = compute_api . API . get_all <EOL> orig_create = compute_api . API . create <EOL> def fake_get ( * args , ** kwargs ) : <EOL> return _get_instances_with_cached_ips ( orig_get , * args , ** kwargs ) <EOL> def fake_get_all ( * args , ** kwargs ) : <EOL> return _get_instances_with_cached_ips ( orig_get_all , * args , ** kwargs ) <EOL> def fake_create ( * args , ** kwargs ) : <EOL> return _create_instances_with_cached_ips ( orig_create , * args , ** kwargs ) <EOL> def fake_pci_device_get_by_addr ( context , node_id , dev_addr ) : <EOL> return test_pci_device . fake_db_dev <EOL> stubs . Set ( db , '<STR_LIT>' , fake_pci_device_get_by_addr ) <EOL> stubs . Set ( compute_api . API , '<STR_LIT>' , fake_get ) <EOL> stubs . Set ( compute_api . API , '<STR_LIT>' , fake_get_all ) <EOL> stubs . Set ( compute_api . API , '<STR_LIT>' , fake_create ) <EOL> def _get_fake_cache ( ) : <EOL> def _ip ( ip , fixed = True , floats = None ) : <EOL> ip_dict = { '<STR_LIT:address>' : ip , '<STR_LIT:type>' : '<STR_LIT>' } <EOL> if not fixed : <EOL> ip_dict [ '<STR_LIT:type>' ] = '<STR_LIT>' <EOL> if fixed and floats : <EOL> ip_dict [ '<STR_LIT>' ] = [ _ip ( f , fixed = False ) for f in floats ] <EOL> return ip_dict <EOL> info = [ { '<STR_LIT:address>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : <NUM_LIT:1> , <EOL> '<STR_LIT:label>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ _ip ( '<STR_LIT>' ) ] } ] } } ] <EOL> if CONF . use_ipv6 : <EOL> ipv6_addr = '<STR_LIT>' <EOL> info [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ '<STR_LIT>' ] . append ( { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ _ip ( ipv6_addr ) ] } ) <EOL> return jsonutils . dumps ( info ) <EOL> def _get_instances_with_cached_ips ( orig_func , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> instances = orig_func ( * args , ** kwargs ) <EOL> context = args [ <NUM_LIT:0> ] <EOL> fake_device = objects . PciDevice . get_by_dev_addr ( context , <NUM_LIT:1> , '<STR_LIT:a>' ) <EOL> def _info_cache_for ( instance ) : <EOL> info_cache = dict ( test_instance_info_cache . fake_info_cache , <EOL> network_info = _get_fake_cache ( ) , <EOL> instance_uuid = instance [ '<STR_LIT>' ] ) <EOL> if isinstance ( instance , obj_base . NovaObject ) : <EOL> _info_cache = objects . InstanceInfoCache ( context ) <EOL> objects . InstanceInfoCache . _from_db_object ( context , _info_cache , <EOL> info_cache ) <EOL> info_cache = _info_cache <EOL> instance [ '<STR_LIT>' ] = info_cache <EOL> if isinstance ( instances , ( list , obj_base . ObjectListBase ) ) : <EOL> for instance in instances : <EOL> _info_cache_for ( instance ) <EOL> fake_device . claim ( instance ) <EOL> fake_device . allocate ( instance ) <EOL> else : <EOL> _info_cache_for ( instances ) <EOL> fake_device . claim ( instances ) <EOL> fake_device . allocate ( instances ) <EOL> return instances <EOL> def _create_instances_with_cached_ips ( orig_func , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> instances , reservation_id = orig_func ( * args , ** kwargs ) <EOL> fake_cache = _get_fake_cache ( ) <EOL> for instance in instances : <EOL> instance [ '<STR_LIT>' ] [ '<STR_LIT>' ] = fake_cache <EOL> db . instance_info_cache_update ( args [ <NUM_LIT:1> ] , instance [ '<STR_LIT>' ] , <EOL> { '<STR_LIT>' : fake_cache } ) <EOL> return ( instances , reservation_id ) </s>
