<s> import mock <EOL> from oslo_config import cfg <EOL> from oslo_utils import uuidutils <EOL> from neutron . services . metering . agents import metering_agent <EOL> from neutron . tests import base <EOL> from neutron . tests import fake_notifier <EOL> _uuid = uuidutils . generate_uuid <EOL> TENANT_ID = _uuid ( ) <EOL> LABEL_ID = _uuid ( ) <EOL> ROUTERS = [ { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : TENANT_ID , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : LABEL_ID } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) } ] <EOL> ROUTERS_WITH_RULE = [ { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : TENANT_ID , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : { } , <EOL> '<STR_LIT:id>' : LABEL_ID } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) } ] <EOL> class TestMeteringOperations ( base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestMeteringOperations , self ) . setUp ( ) <EOL> cfg . CONF . register_opts ( metering_agent . MeteringAgent . Opts ) <EOL> self . noop_driver = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , self . noop_driver ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . setup_notification_driver ( ) <EOL> metering_rpc = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . metering_rpc_patch = mock . patch ( metering_rpc , return_value = [ ] ) <EOL> self . metering_rpc_patch . start ( ) <EOL> self . driver_patch = mock . patch ( self . noop_driver , spec = True ) <EOL> self . driver_patch . start ( ) <EOL> loopingcall_patch = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> loopingcall_patch . start ( ) <EOL> self . agent = metering_agent . MeteringAgent ( '<STR_LIT>' , cfg . CONF ) <EOL> self . driver = self . agent . metering_driver <EOL> def test_add_metering_label ( self ) : <EOL> self . agent . add_metering_label ( None , ROUTERS ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . driver . add_metering_label . call_count ) <EOL> def test_remove_metering_label ( self ) : <EOL> self . agent . remove_metering_label ( None , ROUTERS ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . driver . remove_metering_label . call_count ) <EOL> def test_update_metering_label_rule ( self ) : <EOL> self . agent . update_metering_label_rules ( None , ROUTERS ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . driver . update_metering_label_rules . call_count ) <EOL> def test_add_metering_label_rule ( self ) : <EOL> self . agent . add_metering_label_rule ( None , ROUTERS_WITH_RULE ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . driver . add_metering_label_rule . call_count ) <EOL> def test_remove_metering_label_rule ( self ) : <EOL> self . agent . remove_metering_label_rule ( None , ROUTERS_WITH_RULE ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . driver . remove_metering_label_rule . call_count ) <EOL> def test_routers_updated ( self ) : <EOL> self . agent . routers_updated ( None , ROUTERS ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . driver . update_routers . call_count ) <EOL> def test_get_traffic_counters ( self ) : <EOL> self . agent . _get_traffic_counters ( None , ROUTERS ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . driver . get_traffic_counters . call_count ) <EOL> def test_notification_report ( self ) : <EOL> self . agent . routers_updated ( None , ROUTERS ) <EOL> self . driver . get_traffic_counters . return_value = { LABEL_ID : <EOL> { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> } } <EOL> self . agent . _metering_loop ( ) <EOL> self . assertNotEqual ( len ( fake_notifier . NOTIFICATIONS ) , <NUM_LIT:0> ) <EOL> for n in fake_notifier . NOTIFICATIONS : <EOL> if n [ '<STR_LIT>' ] == '<STR_LIT>' : <EOL> break <EOL> self . assertEqual ( '<STR_LIT>' , n [ '<STR_LIT>' ] ) <EOL> payload = n [ '<STR_LIT>' ] <EOL> self . assertEqual ( TENANT_ID , payload [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( LABEL_ID , payload [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT> , payload [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT> , payload [ '<STR_LIT>' ] ) <EOL> def test_router_deleted ( self ) : <EOL> label_id = _uuid ( ) <EOL> self . driver . get_traffic_counters = mock . MagicMock ( ) <EOL> self . driver . get_traffic_counters . return_value = { label_id : <EOL> { '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : <NUM_LIT> } } <EOL> self . agent . _add_metering_info = mock . MagicMock ( ) <EOL> self . agent . routers_updated ( None , ROUTERS ) <EOL> self . agent . router_deleted ( None , ROUTERS [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . agent . _add_metering_info . call_count ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . driver . remove_router . call_count ) <EOL> self . agent . _add_metering_info . assert_called_with ( label_id , <NUM_LIT> , <NUM_LIT> ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_purge_metering_info ( self , current_timestamp , is_empty , <EOL> mock_time ) : <EOL> mock_time . return_value = current_timestamp <EOL> self . agent . metering_infos = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:1> } } <EOL> self . config ( report_interval = <NUM_LIT:1> ) <EOL> self . agent . _purge_metering_info ( ) <EOL> self . assertEqual ( <NUM_LIT:0> if is_empty else <NUM_LIT:1> , len ( self . agent . metering_infos ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_time . call_count ) <EOL> def test_purge_metering_info ( self ) : <EOL> self . _test_purge_metering_info ( <NUM_LIT:2> , False ) <EOL> def test_purge_metering_info_delete ( self ) : <EOL> self . _test_purge_metering_info ( <NUM_LIT:3> , True ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_add_metering_info ( self , expected_info , current_timestamp , <EOL> mock_time ) : <EOL> mock_time . return_value = current_timestamp <EOL> actual_info = self . agent . _add_metering_info ( '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( self . agent . metering_infos ) ) <EOL> self . assertEqual ( expected_info , actual_info ) <EOL> self . assertEqual ( expected_info , <EOL> self . agent . metering_infos [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , mock_time . call_count ) <EOL> def test_add_metering_info_create ( self ) : <EOL> expected_info = { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:time>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> } <EOL> self . _test_add_metering_info ( expected_info , <NUM_LIT:1> ) <EOL> def test_add_metering_info_update ( self ) : <EOL> expected_info = { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT:time>' : <NUM_LIT:0> , '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:1> } <EOL> self . agent . metering_infos = { '<STR_LIT>' : expected_info } <EOL> expected_info . update ( { '<STR_LIT>' : <NUM_LIT:2> , '<STR_LIT>' : <NUM_LIT:2> , '<STR_LIT:time>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:2> } ) <EOL> self . _test_add_metering_info ( expected_info , <NUM_LIT:2> ) <EOL> class TestMeteringDriver ( base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestMeteringDriver , self ) . setUp ( ) <EOL> cfg . CONF . register_opts ( metering_agent . MeteringAgent . Opts ) <EOL> self . noop_driver = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , self . noop_driver ) <EOL> self . agent = metering_agent . MeteringAgent ( '<STR_LIT>' , cfg . CONF ) <EOL> self . driver = mock . Mock ( ) <EOL> self . agent . metering_driver = self . driver <EOL> def test_add_metering_label_with_bad_driver_impl ( self ) : <EOL> del self . driver . add_metering_label <EOL> with mock . patch . object ( metering_agent , '<STR_LIT>' ) as log : <EOL> self . agent . add_metering_label ( None , ROUTERS ) <EOL> log . exception . assert_called_with ( mock . ANY , <EOL> { '<STR_LIT>' : self . noop_driver , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_add_metering_label_runtime_error ( self ) : <EOL> self . driver . add_metering_label . side_effect = RuntimeError <EOL> with mock . patch . object ( metering_agent , '<STR_LIT>' ) as log : <EOL> self . agent . add_metering_label ( None , ROUTERS ) <EOL> log . exception . assert_called_with ( mock . ANY , <EOL> { '<STR_LIT>' : self . noop_driver , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' } ) <EOL> def test_init_chain ( self ) : <EOL> with mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) as init : <EOL> metering_agent . MeteringAgent ( '<STR_LIT>' , cfg . CONF ) <EOL> init . assert_called_once_with ( cfg . CONF ) </s>
