<s> """<STR_LIT>""" <EOL> import collections <EOL> import mock <EOL> from mox3 import mox <EOL> from oslo_config import cfg <EOL> from nova import context <EOL> from nova . network import rpcapi as network_rpcapi <EOL> from nova . objects import base as objects_base <EOL> from nova import test <EOL> from nova . tests . unit import fake_instance <EOL> from nova . tests . unit import fake_network <EOL> CONF = cfg . CONF <EOL> class NetworkRpcAPITestCase ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( NetworkRpcAPITestCase , self ) . setUp ( ) <EOL> self . flags ( multi_host = True ) <EOL> DefaultArg = collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT:value>' ] ) <EOL> def _test_network_api ( self , method , rpc_method , ** kwargs ) : <EOL> ctxt = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> rpcapi = network_rpcapi . NetworkAPI ( ) <EOL> self . assertIsNotNone ( rpcapi . client ) <EOL> self . assertEqual ( CONF . network_topic , rpcapi . client . target . topic ) <EOL> expected_retval = '<STR_LIT:foo>' if rpc_method == '<STR_LIT>' else None <EOL> expected_version = kwargs . pop ( '<STR_LIT:version>' , None ) <EOL> expected_fanout = kwargs . pop ( '<STR_LIT>' , None ) <EOL> expected_kwargs = kwargs . copy ( ) <EOL> for k , v in expected_kwargs . items ( ) : <EOL> if isinstance ( v , self . DefaultArg ) : <EOL> expected_kwargs [ k ] = v . value <EOL> kwargs . pop ( k ) <EOL> prepare_kwargs = { } <EOL> if expected_version : <EOL> prepare_kwargs [ '<STR_LIT:version>' ] = expected_version <EOL> if expected_fanout : <EOL> prepare_kwargs [ '<STR_LIT>' ] = True <EOL> if '<STR_LIT>' in expected_kwargs : <EOL> expected_kwargs [ '<STR_LIT:source>' ] = expected_kwargs . pop ( '<STR_LIT>' ) <EOL> expected_kwargs [ '<STR_LIT>' ] = expected_kwargs . pop ( '<STR_LIT>' ) <EOL> targeted_methods = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> ] <EOL> targeted_by_instance = [ '<STR_LIT>' ] <EOL> if method in targeted_methods and ( '<STR_LIT:host>' in expected_kwargs or <EOL> '<STR_LIT>' in expected_kwargs ) : <EOL> if method in targeted_by_instance : <EOL> host = expected_kwargs [ '<STR_LIT>' ] [ '<STR_LIT:host>' ] <EOL> else : <EOL> host = expected_kwargs [ '<STR_LIT:host>' ] <EOL> if method not in [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] : <EOL> expected_kwargs . pop ( '<STR_LIT:host>' ) <EOL> if CONF . multi_host : <EOL> prepare_kwargs [ '<STR_LIT>' ] = host <EOL> self . mox . StubOutWithMock ( rpcapi , '<STR_LIT>' ) <EOL> version_check = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> if method in version_check : <EOL> rpcapi . client . can_send_version ( mox . IgnoreArg ( ) ) . AndReturn ( True ) <EOL> if prepare_kwargs : <EOL> rpcapi . client . prepare ( ** prepare_kwargs ) . AndReturn ( rpcapi . client ) <EOL> rpc_method = getattr ( rpcapi . client , rpc_method ) <EOL> rpc_method ( ctxt , method , ** expected_kwargs ) . AndReturn ( '<STR_LIT:foo>' ) <EOL> self . mox . ReplayAll ( ) <EOL> retval = getattr ( rpcapi , method ) ( ctxt , ** kwargs ) <EOL> self . assertEqual ( expected_retval , retval ) <EOL> def test_create_networks ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> arg1 = '<STR_LIT>' , arg2 = '<STR_LIT>' ) <EOL> def test_delete_network ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> uuid = '<STR_LIT>' , fixed_range = '<STR_LIT>' ) <EOL> def test_allocate_for_instance ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> instance_id = '<STR_LIT>' , project_id = '<STR_LIT>' , host = '<STR_LIT>' , <EOL> rxtx_factor = '<STR_LIT>' , vpn = False , requested_networks = { } , <EOL> macs = [ ] , version = '<STR_LIT>' ) <EOL> def test_deallocate_for_instance ( self ) : <EOL> instance = fake_instance . fake_instance_obj ( context . get_admin_context ( ) ) <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> requested_networks = self . DefaultArg ( None ) , instance = instance , <EOL> version = '<STR_LIT>' ) <EOL> def test_deallocate_for_instance_with_expected_networks ( self ) : <EOL> instance = fake_instance . fake_instance_obj ( context . get_admin_context ( ) ) <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> instance = instance , requested_networks = { } , version = '<STR_LIT>' ) <EOL> def test_add_fixed_ip_to_instance ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> instance_id = '<STR_LIT>' , rxtx_factor = '<STR_LIT>' , <EOL> host = '<STR_LIT>' , network_id = '<STR_LIT>' , version = '<STR_LIT>' ) <EOL> def test_remove_fixed_ip_from_instance ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , <EOL> rpc_method = '<STR_LIT>' , instance_id = '<STR_LIT>' , <EOL> rxtx_factor = '<STR_LIT>' , host = '<STR_LIT>' , <EOL> address = '<STR_LIT>' , version = '<STR_LIT>' ) <EOL> def test_add_network_to_project ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> project_id = '<STR_LIT>' , network_uuid = '<STR_LIT>' ) <EOL> def test_get_instance_nw_info ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> instance_id = '<STR_LIT>' , rxtx_factor = '<STR_LIT>' , <EOL> host = '<STR_LIT>' , project_id = '<STR_LIT>' , version = '<STR_LIT>' ) <EOL> def test_validate_networks ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> networks = { } ) <EOL> def test_get_dns_domains ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' ) <EOL> def test_add_dns_entry ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> address = '<STR_LIT>' , name = '<STR_LIT:name>' , dns_type = '<STR_LIT:foo>' , domain = '<STR_LIT>' ) <EOL> def test_modify_dns_entry ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> address = '<STR_LIT>' , name = '<STR_LIT:name>' , domain = '<STR_LIT>' ) <EOL> def test_delete_dns_entry ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> name = '<STR_LIT:name>' , domain = '<STR_LIT>' ) <EOL> def test_delete_dns_domain ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> domain = '<STR_LIT>' ) <EOL> def test_get_dns_entries_by_address ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> address = '<STR_LIT>' , domain = '<STR_LIT>' ) <EOL> def test_get_dns_entries_by_name ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> name = '<STR_LIT>' , domain = '<STR_LIT>' ) <EOL> def test_create_private_dns_domain ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> domain = '<STR_LIT>' , av_zone = '<STR_LIT>' ) <EOL> def test_create_public_dns_domain ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> domain = '<STR_LIT>' , project = '<STR_LIT>' ) <EOL> def test_setup_networks_on_host ( self ) : <EOL> ctxt = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = fake_instance . fake_instance_obj ( ctxt ) <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> instance_id = instance . id , host = '<STR_LIT>' , teardown = False , <EOL> instance = instance , version = '<STR_LIT>' ) <EOL> def test_setup_networks_on_host_v1_0 ( self ) : <EOL> ctxt = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> instance = fake_instance . fake_instance_obj ( ctxt ) <EOL> host = '<STR_LIT>' <EOL> teardown = True <EOL> rpcapi = network_rpcapi . NetworkAPI ( ) <EOL> call_mock = mock . Mock ( ) <EOL> cctxt_mock = mock . Mock ( call = call_mock ) <EOL> with test . nested ( <EOL> mock . patch . object ( rpcapi . client , '<STR_LIT>' , <EOL> return_value = False ) , <EOL> mock . patch . object ( rpcapi . client , '<STR_LIT>' , <EOL> return_value = cctxt_mock ) <EOL> ) as ( <EOL> can_send_mock , prepare_mock <EOL> ) : <EOL> rpcapi . setup_networks_on_host ( ctxt , instance . id , host , teardown , <EOL> instance ) <EOL> can_send_mock . assert_called_once_with ( '<STR_LIT>' ) <EOL> prepare_mock . assert_called_once_with ( version = '<STR_LIT:1.0>' ) <EOL> call_mock . assert_called_once_with ( ctxt , '<STR_LIT>' , <EOL> host = host , teardown = teardown , <EOL> instance_id = instance . id ) <EOL> def test_lease_fixed_ip ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> host = '<STR_LIT>' , address = '<STR_LIT>' ) <EOL> def test_release_fixed_ip ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> host = '<STR_LIT>' , address = '<STR_LIT>' , mac = '<STR_LIT>' , <EOL> version = '<STR_LIT>' ) <EOL> def test_release_fixed_ip_no_mac_support ( self ) : <EOL> ctxt = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> address = '<STR_LIT>' <EOL> host = '<STR_LIT>' <EOL> mac = '<STR_LIT>' <EOL> rpcapi = network_rpcapi . NetworkAPI ( ) <EOL> cast_mock = mock . Mock ( ) <EOL> cctxt_mock = mock . Mock ( cast = cast_mock ) <EOL> with test . nested ( <EOL> mock . patch . object ( rpcapi . client , '<STR_LIT>' , <EOL> return_value = False ) , <EOL> mock . patch . object ( rpcapi . client , '<STR_LIT>' , <EOL> return_value = cctxt_mock ) <EOL> ) as ( <EOL> can_send_mock , prepare_mock <EOL> ) : <EOL> rpcapi . release_fixed_ip ( ctxt , address , host , mac ) <EOL> can_send_mock . assert_called_once_with ( '<STR_LIT>' ) <EOL> prepare_mock . assert_called_once_with ( server = host , version = '<STR_LIT:1.0>' ) <EOL> cast_mock . assert_called_once_with ( ctxt , '<STR_LIT>' , <EOL> address = address ) <EOL> def test_set_network_host ( self ) : <EOL> network = fake_network . fake_network_obj ( context . get_admin_context ( ) ) <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> network_ref = network , version = '<STR_LIT>' ) <EOL> def test_set_network_host_network_object_to_primitive ( self ) : <EOL> ctxt = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> network = fake_network . fake_network_obj ( ctxt ) <EOL> network_dict = objects_base . obj_to_primitive ( network ) <EOL> rpcapi = network_rpcapi . NetworkAPI ( ) <EOL> call_mock = mock . Mock ( ) <EOL> cctxt_mock = mock . Mock ( call = call_mock ) <EOL> with test . nested ( <EOL> mock . patch . object ( rpcapi . client , '<STR_LIT>' , <EOL> return_value = False ) , <EOL> mock . patch . object ( rpcapi . client , '<STR_LIT>' , <EOL> return_value = cctxt_mock ) <EOL> ) as ( <EOL> can_send_mock , prepare_mock <EOL> ) : <EOL> rpcapi . set_network_host ( ctxt , network ) <EOL> can_send_mock . assert_called_once_with ( '<STR_LIT>' ) <EOL> prepare_mock . assert_called_once_with ( version = '<STR_LIT:1.0>' ) <EOL> call_mock . assert_called_once_with ( ctxt , '<STR_LIT>' , <EOL> network_ref = network_dict ) <EOL> def test_rpc_setup_network_on_host ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> network_id = '<STR_LIT>' , teardown = False , host = '<STR_LIT>' ) <EOL> def test_rpc_allocate_fixed_ip ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> instance_id = '<STR_LIT>' , network_id = '<STR_LIT>' , address = '<STR_LIT>' , <EOL> vpn = True , host = '<STR_LIT>' ) <EOL> def test_deallocate_fixed_ip ( self ) : <EOL> instance = fake_instance . fake_db_instance ( ) <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> address = '<STR_LIT>' , host = '<STR_LIT>' , instance = instance , <EOL> version = '<STR_LIT>' ) <EOL> def test_update_dns ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , fanout = True , <EOL> network_ids = '<STR_LIT>' , version = '<STR_LIT>' ) <EOL> def test__associate_floating_ip ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> floating_address = '<STR_LIT>' , fixed_address = '<STR_LIT>' , <EOL> interface = '<STR_LIT>' , host = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' , version = '<STR_LIT>' ) <EOL> def test__disassociate_floating_ip ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> address = '<STR_LIT>' , interface = '<STR_LIT>' , <EOL> host = '<STR_LIT>' , instance_uuid = '<STR_LIT>' , version = '<STR_LIT>' ) <EOL> def test_migrate_instance_start ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' , <EOL> rxtx_factor = '<STR_LIT>' , <EOL> project_id = '<STR_LIT>' , <EOL> source_compute = '<STR_LIT>' , <EOL> dest_compute = '<STR_LIT>' , <EOL> floating_addresses = '<STR_LIT>' , <EOL> host = self . DefaultArg ( None ) , <EOL> version = '<STR_LIT>' ) <EOL> def test_migrate_instance_start_multi_host ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' , <EOL> rxtx_factor = '<STR_LIT>' , <EOL> project_id = '<STR_LIT>' , <EOL> source_compute = '<STR_LIT>' , <EOL> dest_compute = '<STR_LIT>' , <EOL> floating_addresses = '<STR_LIT>' , <EOL> host = '<STR_LIT>' , <EOL> version = '<STR_LIT>' ) <EOL> def test_migrate_instance_finish ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' , <EOL> rxtx_factor = '<STR_LIT>' , <EOL> project_id = '<STR_LIT>' , <EOL> source_compute = '<STR_LIT>' , <EOL> dest_compute = '<STR_LIT>' , <EOL> floating_addresses = '<STR_LIT>' , <EOL> host = self . DefaultArg ( None ) , <EOL> version = '<STR_LIT>' ) <EOL> def test_migrate_instance_finish_multi_host ( self ) : <EOL> self . _test_network_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' , <EOL> rxtx_factor = '<STR_LIT>' , <EOL> project_id = '<STR_LIT>' , <EOL> source_compute = '<STR_LIT>' , <EOL> dest_compute = '<STR_LIT>' , <EOL> floating_addresses = '<STR_LIT>' , <EOL> host = '<STR_LIT>' , <EOL> version = '<STR_LIT>' ) </s>
