<s> import copy <EOL> import fixtures <EOL> import mock <EOL> import oslo_messaging as messaging <EOL> from oslo_serialization import jsonutils <EOL> import testtools <EOL> from nova import context <EOL> from nova import rpc <EOL> from nova import test <EOL> class RPCResetFixture ( fixtures . Fixture ) : <EOL> def _setUp ( self ) : <EOL> self . trans = copy . copy ( rpc . TRANSPORT ) <EOL> self . noti_trans = copy . copy ( rpc . NOTIFICATION_TRANSPORT ) <EOL> self . noti = copy . copy ( rpc . NOTIFIER ) <EOL> self . all_mods = copy . copy ( rpc . ALLOWED_EXMODS ) <EOL> self . ext_mods = copy . copy ( rpc . EXTRA_EXMODS ) <EOL> self . addCleanup ( self . _reset_everything ) <EOL> def _reset_everything ( self ) : <EOL> rpc . TRANSPORT = self . trans <EOL> rpc . NOTIFICATION_TRANSPORT = self . noti_trans <EOL> rpc . NOTIFIER = self . noti <EOL> rpc . ALLOWED_EXMODS = self . all_mods <EOL> rpc . EXTRA_EXMODS = self . ext_mods <EOL> class TestRPC ( testtools . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestRPC , self ) . setUp ( ) <EOL> self . useFixture ( RPCResetFixture ( ) ) <EOL> @ mock . patch . object ( rpc , '<STR_LIT>' ) <EOL> @ mock . patch . object ( rpc , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> def test_init_unversioned ( self , mock_notif , mock_noti_trans , mock_trans , <EOL> mock_ser , mock_exmods ) : <EOL> expected = [ { } , { '<STR_LIT>' : '<STR_LIT>' } ] <EOL> self . _test_init ( mock_notif , mock_noti_trans , mock_trans , mock_ser , <EOL> mock_exmods , '<STR_LIT>' , expected ) <EOL> @ mock . patch . object ( rpc , '<STR_LIT>' ) <EOL> @ mock . patch . object ( rpc , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> def test_init_both ( self , mock_notif , mock_noti_trans , mock_trans , <EOL> mock_ser , mock_exmods ) : <EOL> expected = [ { } , { '<STR_LIT>' : [ '<STR_LIT>' ] } ] <EOL> self . _test_init ( mock_notif , mock_noti_trans , mock_trans , mock_ser , <EOL> mock_exmods , '<STR_LIT>' , expected ) <EOL> @ mock . patch . object ( rpc , '<STR_LIT>' ) <EOL> @ mock . patch . object ( rpc , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> def test_init_versioned ( self , mock_notif , mock_noti_trans , mock_trans , <EOL> mock_ser , mock_exmods ) : <EOL> expected = [ { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : [ '<STR_LIT>' ] } ] <EOL> self . _test_init ( mock_notif , mock_noti_trans , mock_trans , mock_ser , <EOL> mock_exmods , '<STR_LIT>' , expected ) <EOL> def test_cleanup_transport_null ( self ) : <EOL> rpc . NOTIFICATION_TRANSPORT = mock . Mock ( ) <EOL> rpc . LEGACY_NOTIFIER = mock . Mock ( ) <EOL> rpc . NOTIFIER = mock . Mock ( ) <EOL> self . assertRaises ( AssertionError , rpc . cleanup ) <EOL> def test_cleanup_notification_transport_null ( self ) : <EOL> rpc . TRANSPORT = mock . Mock ( ) <EOL> rpc . NOTIFIER = mock . Mock ( ) <EOL> self . assertRaises ( AssertionError , rpc . cleanup ) <EOL> def test_cleanup_legacy_notifier_null ( self ) : <EOL> rpc . TRANSPORT = mock . Mock ( ) <EOL> rpc . NOTIFICATION_TRANSPORT = mock . Mock ( ) <EOL> rpc . NOTIFIER = mock . Mock ( ) <EOL> def test_cleanup_notifier_null ( self ) : <EOL> rpc . TRANSPORT = mock . Mock ( ) <EOL> rpc . LEGACY_NOTIFIER = mock . Mock ( ) <EOL> rpc . NOTIFICATION_TRANSPORT = mock . Mock ( ) <EOL> self . assertRaises ( AssertionError , rpc . cleanup ) <EOL> def test_cleanup ( self ) : <EOL> rpc . LEGACY_NOTIFIER = mock . Mock ( ) <EOL> rpc . NOTIFIER = mock . Mock ( ) <EOL> rpc . NOTIFICATION_TRANSPORT = mock . Mock ( ) <EOL> rpc . TRANSPORT = mock . Mock ( ) <EOL> trans_cleanup = mock . Mock ( ) <EOL> not_trans_cleanup = mock . Mock ( ) <EOL> rpc . TRANSPORT . cleanup = trans_cleanup <EOL> rpc . NOTIFICATION_TRANSPORT . cleanup = not_trans_cleanup <EOL> rpc . cleanup ( ) <EOL> trans_cleanup . assert_called_once_with ( ) <EOL> not_trans_cleanup . assert_called_once_with ( ) <EOL> self . assertIsNone ( rpc . TRANSPORT ) <EOL> self . assertIsNone ( rpc . NOTIFICATION_TRANSPORT ) <EOL> self . assertIsNone ( rpc . LEGACY_NOTIFIER ) <EOL> self . assertIsNone ( rpc . NOTIFIER ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> def test_set_defaults ( self , mock_set ) : <EOL> control_exchange = mock . Mock ( ) <EOL> rpc . set_defaults ( control_exchange ) <EOL> mock_set . assert_called_once_with ( control_exchange ) <EOL> def test_add_extra_exmods ( self ) : <EOL> rpc . EXTRA_EXMODS = [ ] <EOL> rpc . add_extra_exmods ( '<STR_LIT:foo>' , '<STR_LIT:bar>' ) <EOL> self . assertEqual ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] , rpc . EXTRA_EXMODS ) <EOL> def test_clear_extra_exmods ( self ) : <EOL> rpc . EXTRA_EXMODS = [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] <EOL> rpc . clear_extra_exmods ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , len ( rpc . EXTRA_EXMODS ) ) <EOL> def test_get_allowed_exmods ( self ) : <EOL> rpc . ALLOWED_EXMODS = [ '<STR_LIT:foo>' ] <EOL> rpc . EXTRA_EXMODS = [ '<STR_LIT:bar>' ] <EOL> exmods = rpc . get_allowed_exmods ( ) <EOL> self . assertEqual ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] , exmods ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> def test_get_transport_url ( self , mock_url ) : <EOL> conf = mock . Mock ( ) <EOL> rpc . CONF = conf <EOL> mock_url . parse . return_value = '<STR_LIT:foo>' <EOL> url = rpc . get_transport_url ( url_str = '<STR_LIT:bar>' ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , url ) <EOL> mock_url . parse . assert_called_once_with ( conf , '<STR_LIT:bar>' , <EOL> rpc . TRANSPORT_ALIASES ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> def test_get_transport_url_null ( self , mock_url ) : <EOL> conf = mock . Mock ( ) <EOL> rpc . CONF = conf <EOL> mock_url . parse . return_value = '<STR_LIT:foo>' <EOL> url = rpc . get_transport_url ( ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , url ) <EOL> mock_url . parse . assert_called_once_with ( conf , None , <EOL> rpc . TRANSPORT_ALIASES ) <EOL> @ mock . patch . object ( rpc , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> def test_get_client ( self , mock_client , mock_ser ) : <EOL> rpc . TRANSPORT = mock . Mock ( ) <EOL> tgt = mock . Mock ( ) <EOL> ser = mock . Mock ( ) <EOL> mock_client . return_value = '<STR_LIT>' <EOL> mock_ser . return_value = ser <EOL> client = rpc . get_client ( tgt , version_cap = '<STR_LIT:1.0>' , serializer = '<STR_LIT:foo>' ) <EOL> mock_ser . assert_called_once_with ( '<STR_LIT:foo>' ) <EOL> mock_client . assert_called_once_with ( rpc . TRANSPORT , <EOL> tgt , version_cap = '<STR_LIT:1.0>' , <EOL> serializer = ser ) <EOL> self . assertEqual ( '<STR_LIT>' , client ) <EOL> @ mock . patch . object ( rpc , '<STR_LIT>' ) <EOL> @ mock . patch . object ( messaging , '<STR_LIT>' ) <EOL> def test_get_server ( self , mock_get , mock_ser ) : <EOL> rpc . TRANSPORT = mock . Mock ( ) <EOL> ser = mock . Mock ( ) <EOL> tgt = mock . Mock ( ) <EOL> ends = mock . Mock ( ) <EOL> mock_ser . return_value = ser <EOL> mock_get . return_value = '<STR_LIT>' <EOL> server = rpc . get_server ( tgt , ends , serializer = '<STR_LIT:foo>' ) <EOL> mock_ser . assert_called_once_with ( '<STR_LIT:foo>' ) <EOL> mock_get . assert_called_once_with ( rpc . TRANSPORT , tgt , ends , <EOL> executor = '<STR_LIT>' , serializer = ser ) <EOL> self . assertEqual ( '<STR_LIT>' , server ) <EOL> def test_get_notifier ( self ) : <EOL> rpc . LEGACY_NOTIFIER = mock . Mock ( ) <EOL> mock_prep = mock . Mock ( ) <EOL> mock_prep . return_value = '<STR_LIT>' <EOL> rpc . LEGACY_NOTIFIER . prepare = mock_prep <EOL> notifier = rpc . get_notifier ( '<STR_LIT>' , publisher_id = '<STR_LIT:foo>' ) <EOL> mock_prep . assert_called_once_with ( publisher_id = '<STR_LIT:foo>' ) <EOL> self . assertIsInstance ( notifier , rpc . LegacyValidatingNotifier ) <EOL> self . assertEqual ( '<STR_LIT>' , notifier . notifier ) <EOL> def test_get_notifier_null_publisher ( self ) : <EOL> rpc . LEGACY_NOTIFIER = mock . Mock ( ) <EOL> mock_prep = mock . Mock ( ) <EOL> mock_prep . return_value = '<STR_LIT>' <EOL> rpc . LEGACY_NOTIFIER . prepare = mock_prep <EOL> notifier = rpc . get_notifier ( '<STR_LIT>' , host = '<STR_LIT:bar>' ) <EOL> mock_prep . assert_called_once_with ( publisher_id = '<STR_LIT>' ) <EOL> self . assertIsInstance ( notifier , rpc . LegacyValidatingNotifier ) <EOL> self . assertEqual ( '<STR_LIT>' , notifier . notifier ) <EOL> def test_get_versioned_notifier ( self ) : <EOL> rpc . NOTIFIER = mock . Mock ( ) <EOL> mock_prep = mock . Mock ( ) <EOL> mock_prep . return_value = '<STR_LIT>' <EOL> rpc . NOTIFIER . prepare = mock_prep <EOL> notifier = rpc . get_versioned_notifier ( '<STR_LIT>' ) <EOL> mock_prep . assert_called_once_with ( publisher_id = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , notifier ) <EOL> def _test_init ( self , mock_notif , mock_noti_trans , mock_trans , mock_ser , <EOL> mock_exmods , notif_format , expected_driver_topic_kwargs ) : <EOL> legacy_notifier = mock . Mock ( ) <EOL> notifier = mock . Mock ( ) <EOL> notif_transport = mock . Mock ( ) <EOL> transport = mock . Mock ( ) <EOL> serializer = mock . Mock ( ) <EOL> conf = mock . Mock ( ) <EOL> conf . notification_format = notif_format <EOL> mock_exmods . return_value = [ '<STR_LIT:foo>' ] <EOL> mock_trans . return_value = transport <EOL> mock_noti_trans . return_value = notif_transport <EOL> mock_ser . return_value = serializer <EOL> mock_notif . side_effect = [ legacy_notifier , notifier ] <EOL> rpc . init ( conf ) <EOL> mock_exmods . assert_called_once_with ( ) <EOL> mock_trans . assert_called_once_with ( conf , <EOL> allowed_remote_exmods = [ '<STR_LIT:foo>' ] , <EOL> aliases = rpc . TRANSPORT_ALIASES ) <EOL> self . assertIsNotNone ( rpc . TRANSPORT ) <EOL> self . assertIsNotNone ( rpc . LEGACY_NOTIFIER ) <EOL> self . assertIsNotNone ( rpc . NOTIFIER ) <EOL> self . assertEqual ( legacy_notifier , rpc . LEGACY_NOTIFIER ) <EOL> self . assertEqual ( notifier , rpc . NOTIFIER ) <EOL> expected_calls = [ ] <EOL> for kwargs in expected_driver_topic_kwargs : <EOL> expected_kwargs = { '<STR_LIT>' : serializer } <EOL> expected_kwargs . update ( kwargs ) <EOL> expected_calls . append ( ( ( notif_transport , ) , expected_kwargs ) ) <EOL> self . assertEqual ( expected_calls , mock_notif . call_args_list , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class TestJsonPayloadSerializer ( test . NoDBTestCase ) : <EOL> def test_serialize_entity ( self ) : <EOL> with mock . patch . object ( jsonutils , '<STR_LIT>' ) as mock_prim : <EOL> rpc . JsonPayloadSerializer . serialize_entity ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> mock_prim . assert_called_once_with ( '<STR_LIT>' , convert_instances = True ) <EOL> class TestRequestContextSerializer ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestRequestContextSerializer , self ) . setUp ( ) <EOL> self . mock_base = mock . Mock ( ) <EOL> self . ser = rpc . RequestContextSerializer ( self . mock_base ) <EOL> self . ser_null = rpc . RequestContextSerializer ( None ) <EOL> def test_serialize_entity ( self ) : <EOL> self . mock_base . serialize_entity . return_value = '<STR_LIT:foo>' <EOL> ser_ent = self . ser . serialize_entity ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . mock_base . serialize_entity . assert_called_once_with ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , ser_ent ) <EOL> def test_serialize_entity_null_base ( self ) : <EOL> ser_ent = self . ser_null . serialize_entity ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , ser_ent ) <EOL> def test_deserialize_entity ( self ) : <EOL> self . mock_base . deserialize_entity . return_value = '<STR_LIT:foo>' <EOL> deser_ent = self . ser . deserialize_entity ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . mock_base . deserialize_entity . assert_called_once_with ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT:foo>' , deser_ent ) <EOL> def test_deserialize_entity_null_base ( self ) : <EOL> deser_ent = self . ser_null . deserialize_entity ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , deser_ent ) <EOL> def test_serialize_context ( self ) : <EOL> context = mock . Mock ( ) <EOL> self . ser . serialize_context ( context ) <EOL> context . to_dict . assert_called_once_with ( ) <EOL> @ mock . patch . object ( context , '<STR_LIT>' ) <EOL> def test_deserialize_context ( self , mock_req ) : <EOL> self . ser . deserialize_context ( '<STR_LIT>' ) <EOL> mock_req . from_dict . assert_called_once_with ( '<STR_LIT>' ) </s>
