<s> import mock <EOL> from oslo_utils import uuidutils <EOL> from neutron . api . v2 import attributes as attr <EOL> from neutron import context <EOL> from neutron . db import agents_db <EOL> from neutron . db import l3_agentschedulers_db <EOL> from neutron . db . metering import metering_rpc <EOL> from neutron . extensions import l3 as ext_l3 <EOL> from neutron . extensions import metering as ext_metering <EOL> from neutron import manager <EOL> from neutron . plugins . common import constants <EOL> from neutron . tests . common import helpers <EOL> from neutron . tests import tools <EOL> from neutron . tests . unit . db . metering import test_metering_db <EOL> from neutron . tests . unit . db import test_db_base_plugin_v2 <EOL> from neutron . tests . unit . extensions import test_l3 <EOL> _uuid = uuidutils . generate_uuid <EOL> METERING_SERVICE_PLUGIN_KLASS = ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> class MeteringTestExtensionManager ( object ) : <EOL> def get_resources ( self ) : <EOL> attr . RESOURCE_ATTRIBUTE_MAP . update ( ext_metering . RESOURCE_ATTRIBUTE_MAP ) <EOL> attr . RESOURCE_ATTRIBUTE_MAP . update ( ext_l3 . RESOURCE_ATTRIBUTE_MAP ) <EOL> l3_res = ext_l3 . L3 . get_resources ( ) <EOL> metering_res = ext_metering . Metering . get_resources ( ) <EOL> return l3_res + metering_res <EOL> def get_actions ( self ) : <EOL> return [ ] <EOL> def get_request_extensions ( self ) : <EOL> return [ ] <EOL> class TestMeteringPlugin ( test_db_base_plugin_v2 . NeutronDbPluginV2TestCase , <EOL> test_l3 . L3NatTestCaseMixin , <EOL> test_metering_db . MeteringPluginDbTestCaseMixin ) : <EOL> resource_prefix_map = dict ( <EOL> ( k . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) , "<STR_LIT>" ) <EOL> for k in ext_metering . RESOURCE_ATTRIBUTE_MAP . keys ( ) <EOL> ) <EOL> def setUp ( self ) : <EOL> plugin = '<STR_LIT>' <EOL> service_plugins = { '<STR_LIT>' : <EOL> METERING_SERVICE_PLUGIN_KLASS } <EOL> ext_mgr = MeteringTestExtensionManager ( ) <EOL> super ( TestMeteringPlugin , self ) . setUp ( plugin = plugin , ext_mgr = ext_mgr , <EOL> service_plugins = service_plugins ) <EOL> self . uuid = '<STR_LIT>' <EOL> uuid = '<STR_LIT>' <EOL> self . uuid_patch = mock . patch ( uuid , return_value = self . uuid ) <EOL> self . mock_uuid = self . uuid_patch . start ( ) <EOL> self . tenant_id = '<STR_LIT>' <EOL> self . ctx = context . Context ( '<STR_LIT>' , self . tenant_id , is_admin = True ) <EOL> self . context_patch = mock . patch ( '<STR_LIT>' , <EOL> return_value = self . ctx ) <EOL> self . mock_context = self . context_patch . start ( ) <EOL> self . topic = '<STR_LIT>' <EOL> add = ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> self . add_patch = mock . patch ( add ) <EOL> self . mock_add = self . add_patch . start ( ) <EOL> remove = ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> self . remove_patch = mock . patch ( remove ) <EOL> self . mock_remove = self . remove_patch . start ( ) <EOL> update = ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> self . update_patch = mock . patch ( update ) <EOL> self . mock_update = self . update_patch . start ( ) <EOL> add_rule = ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> self . add_rule_patch = mock . patch ( add_rule ) <EOL> self . mock_add_rule = self . add_rule_patch . start ( ) <EOL> remove_rule = ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> self . remove_rule_patch = mock . patch ( remove_rule ) <EOL> self . mock_remove_rule = self . remove_rule_patch . start ( ) <EOL> def test_add_metering_label_rpc_call ( self ) : <EOL> second_uuid = '<STR_LIT>' <EOL> expected = [ { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . tenant_id , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : self . uuid } ] , <EOL> '<STR_LIT:id>' : self . uuid } ] <EOL> tenant_id_2 = '<STR_LIT>' <EOL> self . mock_uuid . return_value = second_uuid <EOL> with self . router ( name = '<STR_LIT>' , tenant_id = tenant_id_2 , <EOL> set_context = True ) : <EOL> self . mock_uuid . return_value = self . uuid <EOL> with self . router ( name = '<STR_LIT>' , tenant_id = self . tenant_id , <EOL> set_context = True ) : <EOL> with self . metering_label ( tenant_id = self . tenant_id , <EOL> set_context = True ) : <EOL> self . mock_add . assert_called_with ( self . ctx , expected ) <EOL> def test_add_metering_label_shared_rpc_call ( self ) : <EOL> second_uuid = '<STR_LIT>' <EOL> expected = [ { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . tenant_id , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : self . uuid } , <EOL> { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : second_uuid } ] , <EOL> '<STR_LIT:id>' : self . uuid } ] <EOL> tenant_id_2 = '<STR_LIT>' <EOL> with self . router ( name = '<STR_LIT>' , tenant_id = self . tenant_id , <EOL> set_context = True ) : <EOL> with self . metering_label ( tenant_id = self . tenant_id , <EOL> set_context = True ) : <EOL> self . mock_uuid . return_value = second_uuid <EOL> with self . metering_label ( tenant_id = tenant_id_2 , shared = True , <EOL> set_context = True ) : <EOL> self . mock_add . assert_called_with ( self . ctx , expected ) <EOL> def test_remove_metering_label_rpc_call ( self ) : <EOL> expected = [ { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . tenant_id , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : self . uuid } ] , <EOL> '<STR_LIT:id>' : self . uuid } ] <EOL> with self . router ( tenant_id = self . tenant_id , set_context = True ) : <EOL> with self . metering_label ( tenant_id = self . tenant_id , <EOL> set_context = True ) as label : <EOL> self . mock_add . assert_called_with ( self . ctx , expected ) <EOL> self . _delete ( '<STR_LIT>' , <EOL> label [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> self . mock_remove . assert_called_with ( self . ctx , expected ) <EOL> def test_remove_one_metering_label_rpc_call ( self ) : <EOL> second_uuid = '<STR_LIT>' <EOL> expected_add = [ { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . tenant_id , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : self . uuid } , <EOL> { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : second_uuid } ] , <EOL> '<STR_LIT:id>' : self . uuid } ] <EOL> expected_remove = [ { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . tenant_id , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : second_uuid } ] , <EOL> '<STR_LIT:id>' : self . uuid } ] <EOL> with self . router ( tenant_id = self . tenant_id , set_context = True ) : <EOL> with self . metering_label ( tenant_id = self . tenant_id , <EOL> set_context = True ) : <EOL> self . mock_uuid . return_value = second_uuid <EOL> with self . metering_label ( tenant_id = self . tenant_id , <EOL> set_context = True ) as label : <EOL> self . mock_add . assert_called_with ( self . ctx , expected_add ) <EOL> self . _delete ( '<STR_LIT>' , <EOL> label [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> self . mock_remove . assert_called_with ( self . ctx , expected_remove ) <EOL> def test_add_and_remove_metering_label_rule_rpc_call ( self ) : <EOL> second_uuid = '<STR_LIT>' <EOL> expected_add = [ { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . tenant_id , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . uuid , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT:id>' : second_uuid } , <EOL> '<STR_LIT:id>' : self . uuid } ] , <EOL> '<STR_LIT:id>' : self . uuid } ] <EOL> expected_del = [ { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . tenant_id , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : self . uuid , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT:id>' : second_uuid } , <EOL> '<STR_LIT:id>' : self . uuid } ] , <EOL> '<STR_LIT:id>' : self . uuid } ] <EOL> with self . router ( tenant_id = self . tenant_id , set_context = True ) : <EOL> with self . metering_label ( tenant_id = self . tenant_id , <EOL> set_context = True ) as label : <EOL> l = label [ '<STR_LIT>' ] <EOL> self . mock_uuid . return_value = second_uuid <EOL> with self . metering_label_rule ( l [ '<STR_LIT:id>' ] ) : <EOL> self . mock_add_rule . assert_called_with ( self . ctx , <EOL> expected_add ) <EOL> self . _delete ( '<STR_LIT>' , second_uuid ) <EOL> self . mock_remove_rule . assert_called_with ( self . ctx , <EOL> expected_del ) <EOL> def test_delete_metering_label_does_not_clear_router_tenant_id ( self ) : <EOL> tenant_id = '<STR_LIT>' <EOL> with self . metering_label ( tenant_id = tenant_id ) as metering_label : <EOL> with self . router ( tenant_id = tenant_id , set_context = True ) as r : <EOL> router = self . _show ( '<STR_LIT>' , r [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( tenant_id , router [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> metering_label_id = metering_label [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> self . _delete ( '<STR_LIT>' , metering_label_id , <NUM_LIT> ) <EOL> router = self . _show ( '<STR_LIT>' , r [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( tenant_id , router [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> class TestMeteringPluginL3AgentScheduler ( <EOL> l3_agentschedulers_db . L3AgentSchedulerDbMixin , <EOL> test_db_base_plugin_v2 . NeutronDbPluginV2TestCase , <EOL> test_l3 . L3NatTestCaseMixin , <EOL> test_metering_db . MeteringPluginDbTestCaseMixin ) : <EOL> resource_prefix_map = dict ( <EOL> ( k . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) , "<STR_LIT>" ) <EOL> for k in ext_metering . RESOURCE_ATTRIBUTE_MAP . keys ( ) <EOL> ) <EOL> def setUp ( self , plugin_str = None , service_plugins = None , scheduler = None ) : <EOL> if not plugin_str : <EOL> plugin_str = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if not service_plugins : <EOL> service_plugins = { '<STR_LIT>' : <EOL> METERING_SERVICE_PLUGIN_KLASS } <EOL> if not scheduler : <EOL> scheduler = plugin_str <EOL> ext_mgr = MeteringTestExtensionManager ( ) <EOL> super ( TestMeteringPluginL3AgentScheduler , <EOL> self ) . setUp ( plugin = plugin_str , ext_mgr = ext_mgr , <EOL> service_plugins = service_plugins ) <EOL> self . uuid = '<STR_LIT>' <EOL> uuid = '<STR_LIT>' <EOL> self . uuid_patch = mock . patch ( uuid , return_value = self . uuid ) <EOL> self . mock_uuid = self . uuid_patch . start ( ) <EOL> self . tenant_id = '<STR_LIT>' <EOL> self . ctx = context . Context ( '<STR_LIT>' , self . tenant_id , is_admin = True ) <EOL> self . context_patch = mock . patch ( '<STR_LIT>' , <EOL> return_value = self . ctx ) <EOL> self . mock_context = self . context_patch . start ( ) <EOL> self . l3routers_patch = mock . patch ( scheduler + <EOL> '<STR_LIT>' ) <EOL> self . l3routers_mock = self . l3routers_patch . start ( ) <EOL> self . topic = '<STR_LIT>' <EOL> add = ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> self . add_patch = mock . patch ( add ) <EOL> self . mock_add = self . add_patch . start ( ) <EOL> remove = ( '<STR_LIT>' + <EOL> '<STR_LIT>' + <EOL> '<STR_LIT>' ) <EOL> self . remove_patch = mock . patch ( remove ) <EOL> self . mock_remove = self . remove_patch . start ( ) <EOL> def test_add_metering_label_rpc_call ( self ) : <EOL> second_uuid = '<STR_LIT>' <EOL> expected = [ { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . tenant_id , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : second_uuid } ] , <EOL> '<STR_LIT:id>' : self . uuid } , <EOL> { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : self . tenant_id , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : second_uuid } ] , <EOL> '<STR_LIT:id>' : second_uuid } ] <EOL> agent1 = agents_db . Agent ( host = '<STR_LIT>' ) <EOL> agent2 = agents_db . Agent ( host = '<STR_LIT>' ) <EOL> agents = { self . uuid : agent1 , <EOL> second_uuid : agent2 } <EOL> def side_effect ( context , routers , admin_state_up , active ) : <EOL> return [ agents [ routers [ <NUM_LIT:0> ] ] ] <EOL> self . l3routers_mock . side_effect = side_effect <EOL> with self . router ( name = '<STR_LIT>' , tenant_id = self . tenant_id , <EOL> set_context = True ) : <EOL> self . mock_uuid . return_value = second_uuid <EOL> with self . router ( name = '<STR_LIT>' , tenant_id = self . tenant_id , <EOL> set_context = True ) : <EOL> with self . metering_label ( tenant_id = self . tenant_id , <EOL> set_context = True ) : <EOL> self . mock_add . assert_called_with ( <EOL> self . ctx , tools . UnorderedList ( expected ) ) <EOL> class TestMeteringPluginL3AgentSchedulerServicePlugin ( <EOL> TestMeteringPluginL3AgentScheduler ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> l3_plugin = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> service_plugins = { '<STR_LIT>' : <EOL> METERING_SERVICE_PLUGIN_KLASS , <EOL> '<STR_LIT>' : l3_plugin } <EOL> plugin_str = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> super ( TestMeteringPluginL3AgentSchedulerServicePlugin , self ) . setUp ( <EOL> plugin_str = plugin_str , service_plugins = service_plugins , <EOL> scheduler = l3_plugin ) <EOL> class TestMeteringPluginRpcFromL3Agent ( <EOL> test_db_base_plugin_v2 . NeutronDbPluginV2TestCase , <EOL> test_l3 . L3NatTestCaseMixin , <EOL> test_metering_db . MeteringPluginDbTestCaseMixin ) : <EOL> resource_prefix_map = dict ( <EOL> ( k . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) , "<STR_LIT>" ) <EOL> for k in ext_metering . RESOURCE_ATTRIBUTE_MAP <EOL> ) <EOL> def setUp ( self ) : <EOL> service_plugins = { '<STR_LIT>' : <EOL> METERING_SERVICE_PLUGIN_KLASS } <EOL> plugin = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> ext_mgr = MeteringTestExtensionManager ( ) <EOL> super ( TestMeteringPluginRpcFromL3Agent , <EOL> self ) . setUp ( plugin = plugin , service_plugins = service_plugins , <EOL> ext_mgr = ext_mgr ) <EOL> self . meter_plugin = manager . NeutronManager . get_service_plugins ( ) . get ( <EOL> constants . METERING ) <EOL> self . tenant_id = '<STR_LIT>' <EOL> self . tenant_id_1 = '<STR_LIT>' <EOL> self . tenant_id_2 = '<STR_LIT>' <EOL> self . adminContext = context . get_admin_context ( ) <EOL> helpers . register_l3_agent ( host = '<STR_LIT>' ) <EOL> def test_get_sync_data_metering ( self ) : <EOL> with self . subnet ( ) as subnet : <EOL> s = subnet [ '<STR_LIT>' ] <EOL> self . _set_net_external ( s [ '<STR_LIT>' ] ) <EOL> with self . router ( name = '<STR_LIT>' , subnet = subnet ) as router : <EOL> r = router [ '<STR_LIT>' ] <EOL> self . _add_external_gateway_to_router ( r [ '<STR_LIT:id>' ] , s [ '<STR_LIT>' ] ) <EOL> with self . metering_label ( tenant_id = r [ '<STR_LIT>' ] ) : <EOL> callbacks = metering_rpc . MeteringRpcCallbacks ( <EOL> self . meter_plugin ) <EOL> data = callbacks . get_sync_data_metering ( self . adminContext , <EOL> host = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , data [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] ) <EOL> helpers . register_l3_agent ( host = '<STR_LIT>' ) <EOL> data = callbacks . get_sync_data_metering ( self . adminContext , <EOL> host = '<STR_LIT>' ) <EOL> self . assertFalse ( data ) <EOL> self . _remove_external_gateway_from_router ( <EOL> r [ '<STR_LIT:id>' ] , s [ '<STR_LIT>' ] ) <EOL> def test_get_sync_data_metering_shared ( self ) : <EOL> with self . router ( name = '<STR_LIT>' , tenant_id = self . tenant_id_1 ) : <EOL> with self . router ( name = '<STR_LIT>' , tenant_id = self . tenant_id_2 ) : <EOL> with self . metering_label ( tenant_id = self . tenant_id , <EOL> shared = True ) : <EOL> callbacks = metering_rpc . MeteringRpcCallbacks ( <EOL> self . meter_plugin ) <EOL> data = callbacks . get_sync_data_metering ( self . adminContext ) <EOL> routers = [ router [ '<STR_LIT:name>' ] for router in data ] <EOL> self . assertIn ( '<STR_LIT>' , routers ) <EOL> self . assertIn ( '<STR_LIT>' , routers ) <EOL> def test_get_sync_data_metering_not_shared ( self ) : <EOL> with self . router ( name = '<STR_LIT>' , tenant_id = self . tenant_id_1 ) : <EOL> with self . router ( name = '<STR_LIT>' , tenant_id = self . tenant_id_2 ) : <EOL> with self . metering_label ( tenant_id = self . tenant_id ) : <EOL> callbacks = metering_rpc . MeteringRpcCallbacks ( <EOL> self . meter_plugin ) <EOL> data = callbacks . get_sync_data_metering ( self . adminContext ) <EOL> routers = [ router [ '<STR_LIT:name>' ] for router in data ] <EOL> self . assertEqual ( [ ] , routers ) </s>
