<s> """<STR_LIT>""" <EOL> import mock <EOL> from oslo_config import cfg <EOL> from nova import context <EOL> from nova import objects <EOL> from nova . scheduler import rpcapi as scheduler_rpcapi <EOL> from nova import test <EOL> CONF = cfg . CONF <EOL> class SchedulerRpcAPITestCase ( test . NoDBTestCase ) : <EOL> def _test_scheduler_api ( self , method , rpc_method , expected_args = None , <EOL> ** kwargs ) : <EOL> ctxt = context . RequestContext ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> rpcapi = scheduler_rpcapi . SchedulerAPI ( ) <EOL> self . assertIsNotNone ( rpcapi . client ) <EOL> self . assertEqual ( rpcapi . client . target . topic , CONF . scheduler_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> if expected_args : <EOL> expected_kwargs = expected_args <EOL> prepare_kwargs = { } <EOL> if expected_fanout : <EOL> prepare_kwargs [ '<STR_LIT>' ] = True <EOL> if expected_version : <EOL> prepare_kwargs [ '<STR_LIT:version>' ] = expected_version <EOL> orig_prepare = rpcapi . client . prepare <EOL> def fake_can_send_version ( version = None ) : <EOL> return orig_prepare ( version = version ) . can_send_version ( ) <EOL> @ mock . patch . object ( rpcapi . client , rpc_method , <EOL> return_value = expected_retval ) <EOL> @ mock . patch . object ( rpcapi . client , '<STR_LIT>' , <EOL> return_value = rpcapi . client ) <EOL> @ mock . patch . object ( rpcapi . client , '<STR_LIT>' , <EOL> side_effect = fake_can_send_version ) <EOL> def do_test ( mock_csv , mock_prepare , mock_rpc_method ) : <EOL> retval = getattr ( rpcapi , method ) ( ctxt , ** kwargs ) <EOL> self . assertEqual ( retval , expected_retval ) <EOL> mock_prepare . assert_called_once_with ( ** prepare_kwargs ) <EOL> mock_rpc_method . assert_called_once_with ( ctxt , method , <EOL> ** expected_kwargs ) <EOL> do_test ( ) <EOL> def test_select_destinations ( self ) : <EOL> fake_spec = objects . RequestSpec ( ) <EOL> self . _test_scheduler_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> spec_obj = fake_spec , <EOL> version = '<STR_LIT>' ) <EOL> @ mock . patch . object ( objects . RequestSpec , '<STR_LIT>' ) <EOL> @ mock . patch . object ( objects . RequestSpec , '<STR_LIT>' ) <EOL> def test_select_destinations_with_old_manager ( self , to_spec , to_props ) : <EOL> self . flags ( scheduler = '<STR_LIT>' , group = '<STR_LIT>' ) <EOL> to_spec . return_value = '<STR_LIT>' <EOL> to_props . return_value = '<STR_LIT>' <EOL> fake_spec = objects . RequestSpec ( ) <EOL> self . _test_scheduler_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> expected_args = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> spec_obj = fake_spec , <EOL> version = '<STR_LIT>' ) <EOL> def test_update_aggregates ( self ) : <EOL> self . _test_scheduler_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> aggregates = '<STR_LIT>' , <EOL> version = '<STR_LIT>' , <EOL> fanout = True ) <EOL> def test_delete_aggregate ( self ) : <EOL> self . _test_scheduler_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> aggregate = '<STR_LIT>' , <EOL> version = '<STR_LIT>' , <EOL> fanout = True ) <EOL> def test_update_instance_info ( self ) : <EOL> self . _test_scheduler_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> host_name = '<STR_LIT>' , <EOL> instance_info = '<STR_LIT>' , <EOL> fanout = True , <EOL> version = '<STR_LIT>' ) <EOL> def test_delete_instance_info ( self ) : <EOL> self . _test_scheduler_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> host_name = '<STR_LIT>' , <EOL> instance_uuid = '<STR_LIT>' , <EOL> fanout = True , <EOL> version = '<STR_LIT>' ) <EOL> def test_sync_instance_info ( self ) : <EOL> self . _test_scheduler_api ( '<STR_LIT>' , rpc_method = '<STR_LIT>' , <EOL> host_name = '<STR_LIT>' , <EOL> instance_uuids = [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> fanout = True , <EOL> version = '<STR_LIT>' ) </s>
