<s> import copy <EOL> from itertools import chain as iter_chain <EOL> from itertools import combinations as iter_combinations <EOL> import eventlet <EOL> import mock <EOL> import netaddr <EOL> from neutron_lib import constants as l3_constants <EOL> from neutron_lib import exceptions as exc <EOL> from oslo_log import log <EOL> import oslo_messaging <EOL> from oslo_utils import timeutils <EOL> from oslo_utils import uuidutils <EOL> import six <EOL> from testtools import matchers <EOL> from neutron . agent . common import config as agent_config <EOL> from neutron . agent . l3 import agent as l3_agent <EOL> from neutron . agent . l3 import config as l3_config <EOL> from neutron . agent . l3 import dvr_edge_router as dvr_router <EOL> from neutron . agent . l3 import dvr_snat_ns <EOL> from neutron . agent . l3 import ha <EOL> from neutron . agent . l3 import legacy_router <EOL> from neutron . agent . l3 import link_local_allocator as lla <EOL> from neutron . agent . l3 import namespaces <EOL> from neutron . agent . l3 import router_info as l3router <EOL> from neutron . agent . l3 import router_processing_queue <EOL> from neutron . agent . linux import dibbler <EOL> from neutron . agent . linux import external_process <EOL> from neutron . agent . linux import interface <EOL> from neutron . agent . linux import iptables_manager <EOL> from neutron . agent . linux import pd <EOL> from neutron . agent . linux import ra <EOL> from neutron . agent . metadata import driver as metadata_driver <EOL> from neutron . agent import rpc as agent_rpc <EOL> from neutron . common import config as base_config <EOL> from neutron . common import constants as n_const <EOL> from neutron . common import exceptions as n_exc <EOL> from neutron . extensions import portbindings <EOL> from neutron . plugins . common import constants as p_const <EOL> from neutron . tests import base <EOL> from neutron . tests . common import l3_test_common <EOL> _uuid = uuidutils . generate_uuid <EOL> HOSTNAME = '<STR_LIT>' <EOL> FAKE_ID = _uuid ( ) <EOL> FAKE_ID_2 = _uuid ( ) <EOL> FIP_PRI = <NUM_LIT> <EOL> class BasicRouterOperationsFramework ( base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( BasicRouterOperationsFramework , self ) . setUp ( ) <EOL> mock . patch ( '<STR_LIT>' ) . start ( ) <EOL> self . conf = agent_config . setup_conf ( ) <EOL> self . conf . register_opts ( base_config . core_opts ) <EOL> log . register_options ( self . conf ) <EOL> self . conf . register_opts ( agent_config . AGENT_STATE_OPTS , '<STR_LIT>' ) <EOL> self . conf . register_opts ( l3_config . OPTS ) <EOL> self . conf . register_opts ( ha . OPTS ) <EOL> agent_config . register_interface_driver_opts_helper ( self . conf ) <EOL> agent_config . register_process_monitor_opts ( self . conf ) <EOL> agent_config . register_availability_zone_opts_helper ( self . conf ) <EOL> self . conf . register_opts ( interface . OPTS ) <EOL> self . conf . register_opts ( external_process . OPTS ) <EOL> self . conf . register_opts ( pd . OPTS ) <EOL> self . conf . register_opts ( ra . OPTS ) <EOL> self . conf . set_override ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . conf . set_override ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> self . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . device_exists_p = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> self . device_exists = self . device_exists_p . start ( ) <EOL> self . ensure_dir = mock . patch ( '<STR_LIT>' ) . start ( ) <EOL> mock . patch ( '<STR_LIT>' <EOL> '<STR_LIT>' ) . start ( ) <EOL> self . utils_exec_p = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> self . utils_exec = self . utils_exec_p . start ( ) <EOL> self . utils_replace_file_p = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> self . utils_replace_file = self . utils_replace_file_p . start ( ) <EOL> self . external_process_p = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> self . external_process = self . external_process_p . start ( ) <EOL> self . process_monitor = mock . patch ( <EOL> '<STR_LIT>' ) . start ( ) <EOL> self . send_adv_notif_p = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> self . send_adv_notif = self . send_adv_notif_p . start ( ) <EOL> self . dvr_cls_p = mock . patch ( '<STR_LIT>' ) <EOL> driver_cls = self . dvr_cls_p . start ( ) <EOL> self . mock_driver = mock . MagicMock ( ) <EOL> self . mock_driver . DEV_NAME_LEN = ( <EOL> interface . LinuxInterfaceDriver . DEV_NAME_LEN ) <EOL> driver_cls . return_value = self . mock_driver <EOL> self . ip_cls_p = mock . patch ( '<STR_LIT>' ) <EOL> ip_cls = self . ip_cls_p . start ( ) <EOL> self . mock_ip = mock . MagicMock ( ) <EOL> ip_cls . return_value = self . mock_ip <EOL> ip_rule = mock . patch ( '<STR_LIT>' ) . start ( ) <EOL> self . mock_rule = mock . MagicMock ( ) <EOL> ip_rule . return_value = self . mock_rule <EOL> ip_dev = mock . patch ( '<STR_LIT>' ) . start ( ) <EOL> self . mock_ip_dev = mock . MagicMock ( ) <EOL> ip_dev . return_value = self . mock_ip_dev <EOL> self . l3pluginApi_cls_p = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> l3pluginApi_cls = self . l3pluginApi_cls_p . start ( ) <EOL> self . plugin_api = mock . MagicMock ( ) <EOL> l3pluginApi_cls . return_value = self . plugin_api <EOL> self . looping_call_p = mock . patch ( <EOL> '<STR_LIT>' ) <EOL> self . looping_call_p . start ( ) <EOL> subnet_id_1 = _uuid ( ) <EOL> subnet_id_2 = _uuid ( ) <EOL> self . snat_ports = [ { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : subnet_id_1 } ] , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : <EOL> l3_constants . DEVICE_OWNER_ROUTER_SNAT , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : subnet_id_1 , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:16> } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , '<STR_LIT>' : _uuid ( ) } , <EOL> { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : subnet_id_2 } ] , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : <EOL> l3_constants . DEVICE_OWNER_ROUTER_SNAT , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : subnet_id_2 , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:16> } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , '<STR_LIT>' : _uuid ( ) } ] <EOL> self . ri_kwargs = { '<STR_LIT>' : self . conf , <EOL> '<STR_LIT>' : self . mock_driver } <EOL> def _process_router_instance_for_agent ( self , agent , ri , router ) : <EOL> ri . router = router <EOL> if not ri . radvd : <EOL> ri . radvd = ra . DaemonMonitor ( router [ '<STR_LIT:id>' ] , <EOL> ri . ns_name , <EOL> agent . process_monitor , <EOL> ri . get_internal_device_name , <EOL> self . conf ) <EOL> ri . process ( agent ) <EOL> class TestBasicRouterOperations ( BasicRouterOperationsFramework ) : <EOL> def test_init_ha_conf ( self ) : <EOL> with mock . patch ( '<STR_LIT>' , return_value = '<STR_LIT>' ) : <EOL> l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . ensure_dir . assert_called_once_with ( '<STR_LIT>' ) <EOL> def test_enqueue_state_change_router_not_found ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> non_existent_router = <NUM_LIT> <EOL> agent . enqueue_state_change ( non_existent_router , '<STR_LIT>' ) <EOL> def test_enqueue_state_change_metadata_disable ( self ) : <EOL> self . conf . set_override ( '<STR_LIT>' , False ) <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = mock . Mock ( ) <EOL> router_info = mock . MagicMock ( ) <EOL> agent . router_info [ router . id ] = router_info <EOL> agent . _update_metadata_proxy = mock . Mock ( ) <EOL> agent . enqueue_state_change ( router . id , '<STR_LIT>' ) <EOL> self . assertFalse ( agent . _update_metadata_proxy . call_count ) <EOL> def test_periodic_sync_routers_task_raise_exception ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . plugin_api . get_router_ids . return_value = [ '<STR_LIT>' ] <EOL> self . plugin_api . get_routers . side_effect = ValueError <EOL> self . assertRaises ( ValueError , <EOL> agent . periodic_sync_routers_task , <EOL> agent . context ) <EOL> self . assertTrue ( agent . fullsync ) <EOL> def test_l3_initial_report_state_done ( self ) : <EOL> with mock . patch . object ( l3_agent . L3NATAgentWithStateReport , <EOL> '<STR_LIT>' ) , mock . patch . object ( agent_rpc . PluginReportStateAPI , <EOL> '<STR_LIT>' ) as report_state , mock . patch . object ( eventlet , '<STR_LIT>' ) : <EOL> agent = l3_agent . L3NATAgentWithStateReport ( host = HOSTNAME , <EOL> conf = self . conf ) <EOL> self . assertTrue ( agent . agent_state [ '<STR_LIT>' ] ) <EOL> agent . after_start ( ) <EOL> report_state . assert_called_once_with ( agent . context , <EOL> agent . agent_state , <EOL> True ) <EOL> self . assertIsNone ( agent . agent_state . get ( '<STR_LIT>' ) ) <EOL> def test_report_state_revival_logic ( self ) : <EOL> with mock . patch . object ( agent_rpc . PluginReportStateAPI , <EOL> '<STR_LIT>' ) as report_state : <EOL> agent = l3_agent . L3NATAgentWithStateReport ( host = HOSTNAME , <EOL> conf = self . conf ) <EOL> report_state . return_value = n_const . AGENT_REVIVED <EOL> agent . _report_state ( ) <EOL> self . assertTrue ( agent . fullsync ) <EOL> agent . fullsync = False <EOL> report_state . return_value = n_const . AGENT_ALIVE <EOL> agent . _report_state ( ) <EOL> self . assertFalse ( agent . fullsync ) <EOL> def test_periodic_sync_routers_task_call_clean_stale_namespaces ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . plugin_api . get_routers . return_value = [ ] <EOL> agent . periodic_sync_routers_task ( agent . context ) <EOL> self . assertFalse ( agent . namespaces_manager . _clean_stale ) <EOL> def test_periodic_sync_routers_task_call_clean_stale_meta_proxies ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> stale_router_ids = [ _uuid ( ) , _uuid ( ) ] <EOL> active_routers = [ { '<STR_LIT:id>' : _uuid ( ) } , { '<STR_LIT:id>' : _uuid ( ) } ] <EOL> self . plugin_api . get_router_ids . return_value = [ r [ '<STR_LIT:id>' ] for r <EOL> in active_routers ] <EOL> self . plugin_api . get_routers . return_value = active_routers <EOL> namespace_list = [ namespaces . NS_PREFIX + r_id <EOL> for r_id in stale_router_ids ] <EOL> namespace_list += [ namespaces . NS_PREFIX + r [ '<STR_LIT:id>' ] <EOL> for r in active_routers ] <EOL> self . mock_ip . get_namespaces . return_value = namespace_list <EOL> driver = metadata_driver . MetadataDriver <EOL> with mock . patch . object ( <EOL> driver , '<STR_LIT>' ) as destroy_proxy : <EOL> agent . periodic_sync_routers_task ( agent . context ) <EOL> expected_calls = [ mock . call ( mock . ANY , r_id , agent . conf ) <EOL> for r_id in stale_router_ids ] <EOL> self . assertEqual ( len ( stale_router_ids ) , destroy_proxy . call_count ) <EOL> destroy_proxy . assert_has_calls ( expected_calls , any_order = True ) <EOL> def test_router_info_create ( self ) : <EOL> id = _uuid ( ) <EOL> ri = l3router . RouterInfo ( id , { } , ** self . ri_kwargs ) <EOL> self . assertTrue ( ri . ns_name . endswith ( id ) ) <EOL> def test_router_info_create_with_router ( self ) : <EOL> ns_id = _uuid ( ) <EOL> subnet_id = _uuid ( ) <EOL> ex_gw_port = { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : subnet_id } ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT:id>' : subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] } <EOL> router = { <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : ex_gw_port } <EOL> ri = l3router . RouterInfo ( ns_id , router , ** self . ri_kwargs ) <EOL> self . assertTrue ( ri . ns_name . endswith ( ns_id ) ) <EOL> self . assertEqual ( router , ri . router ) <EOL> def test_agent_create ( self ) : <EOL> l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> def _test_internal_network_action ( self , action ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> router_id = router [ '<STR_LIT:id>' ] <EOL> ri = l3router . RouterInfo ( router_id , router , ** self . ri_kwargs ) <EOL> port = { '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } ] } <EOL> interface_name = ri . get_internal_device_name ( port [ '<STR_LIT:id>' ] ) <EOL> if action == '<STR_LIT>' : <EOL> self . device_exists . return_value = False <EOL> ri . internal_network_added ( port ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . mock_driver . plug . call_count ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . mock_driver . init_router_port . call_count ) <EOL> self . send_adv_notif . assert_called_once_with ( ri . ns_name , <EOL> interface_name , <EOL> '<STR_LIT>' , mock . ANY ) <EOL> elif action == '<STR_LIT>' : <EOL> self . device_exists . return_value = True <EOL> ri . internal_network_removed ( port ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . mock_driver . unplug . call_count ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" % action ) <EOL> @ staticmethod <EOL> def _fixed_ip_cidr ( fixed_ip ) : <EOL> return '<STR_LIT>' % ( fixed_ip [ '<STR_LIT>' ] , fixed_ip [ '<STR_LIT>' ] ) <EOL> def _test_internal_network_action_dist ( self , action ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> router_id = router [ '<STR_LIT:id>' ] <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = dvr_router . DvrEdgeRouter ( <EOL> agent , HOSTNAME , router_id , router , ** self . ri_kwargs ) <EOL> subnet_id = _uuid ( ) <EOL> port = { '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT:id>' : subnet_id } ] } <EOL> ri . router [ '<STR_LIT>' ] = HOSTNAME <EOL> agent . host = HOSTNAME <EOL> agent . conf . agent_mode = '<STR_LIT>' <EOL> sn_port = { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _uuid ( ) } ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> ex_gw_port = { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : _uuid ( ) } ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> portbindings . HOST_ID : HOSTNAME , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> ri . snat_ports = sn_port <EOL> ri . ex_gw_port = ex_gw_port <EOL> ri . snat_namespace = mock . Mock ( ) <EOL> if action == '<STR_LIT>' : <EOL> self . device_exists . return_value = False <EOL> ri . get_snat_port_for_internal_port = mock . Mock ( <EOL> return_value = sn_port ) <EOL> ri . _snat_redirect_add = mock . Mock ( ) <EOL> ri . _set_subnet_arp_info = mock . Mock ( ) <EOL> ri . _internal_network_added = mock . Mock ( ) <EOL> ri . _set_subnet_arp_info = mock . Mock ( ) <EOL> ri . internal_network_added ( port ) <EOL> self . assertEqual ( <NUM_LIT:1> , ri . _snat_redirect_add . call_count ) <EOL> self . assertEqual ( <NUM_LIT:2> , ri . _internal_network_added . call_count ) <EOL> ri . _set_subnet_arp_info . assert_called_once_with ( subnet_id ) <EOL> ri . _internal_network_added . assert_called_with ( <EOL> dvr_snat_ns . SnatNamespace . get_snat_ns_name ( ri . router [ '<STR_LIT:id>' ] ) , <EOL> sn_port [ '<STR_LIT>' ] , <EOL> sn_port [ '<STR_LIT:id>' ] , <EOL> sn_port [ '<STR_LIT>' ] , <EOL> sn_port [ '<STR_LIT>' ] , <EOL> ri . _get_snat_int_device_name ( sn_port [ '<STR_LIT:id>' ] ) , <EOL> dvr_snat_ns . SNAT_INT_DEV_PREFIX , <EOL> mtu = None ) <EOL> elif action == '<STR_LIT>' : <EOL> self . device_exists . return_value = False <EOL> ri . get_snat_port_for_internal_port = mock . Mock ( <EOL> return_value = sn_port ) <EOL> ri . _delete_arp_cache_for_internal_port = mock . Mock ( ) <EOL> ri . _snat_redirect_modify = mock . Mock ( ) <EOL> ri . internal_network_removed ( port ) <EOL> self . assertEqual ( <EOL> <NUM_LIT:1> , ri . _delete_arp_cache_for_internal_port . call_count ) <EOL> ri . _snat_redirect_modify . assert_called_with ( <EOL> sn_port , port , <EOL> ri . get_internal_device_name ( port [ '<STR_LIT:id>' ] ) , <EOL> is_add = False ) <EOL> def test_agent_add_internal_network ( self ) : <EOL> self . _test_internal_network_action ( '<STR_LIT>' ) <EOL> def test_agent_add_internal_network_dist ( self ) : <EOL> self . _test_internal_network_action_dist ( '<STR_LIT>' ) <EOL> def test_agent_remove_internal_network ( self ) : <EOL> self . _test_internal_network_action ( '<STR_LIT>' ) <EOL> def test_agent_remove_internal_network_dist ( self ) : <EOL> self . _test_internal_network_action_dist ( '<STR_LIT>' ) <EOL> def _add_external_gateway ( self , ri , router , ex_gw_port , interface_name , <EOL> use_fake_fip = False , <EOL> no_subnet = False , no_sub_gw = None , <EOL> dual_stack = False ) : <EOL> self . device_exists . return_value = False <EOL> if no_sub_gw is None : <EOL> no_sub_gw = [ ] <EOL> if use_fake_fip : <EOL> fake_fip = { '<STR_LIT>' : [ { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _uuid ( ) } ] } <EOL> router [ l3_constants . FLOATINGIP_KEY ] = fake_fip [ '<STR_LIT>' ] <EOL> ri . external_gateway_added ( ex_gw_port , interface_name ) <EOL> if not router . get ( '<STR_LIT>' ) : <EOL> self . assertEqual ( <NUM_LIT:1> , self . mock_driver . plug . call_count ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . mock_driver . init_router_port . call_count ) <EOL> if no_subnet and not dual_stack : <EOL> self . assertEqual ( <NUM_LIT:0> , self . send_adv_notif . call_count ) <EOL> ip_cidrs = [ ] <EOL> kwargs = { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : '<STR_LIT>' + router [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : True } <EOL> else : <EOL> exp_arp_calls = [ mock . call ( ri . ns_name , interface_name , <EOL> '<STR_LIT>' , mock . ANY ) ] <EOL> if dual_stack and not no_sub_gw : <EOL> exp_arp_calls += [ mock . call ( ri . ns_name , interface_name , <EOL> '<STR_LIT>' , <EOL> mock . ANY ) ] <EOL> self . send_adv_notif . assert_has_calls ( exp_arp_calls ) <EOL> ip_cidrs = [ '<STR_LIT>' ] <EOL> if dual_stack : <EOL> if not no_sub_gw : <EOL> ip_cidrs . append ( '<STR_LIT>' ) <EOL> kwargs = { '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' + router [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT>' : True } <EOL> self . mock_driver . init_router_port . assert_called_with ( <EOL> interface_name , ip_cidrs , ** kwargs ) <EOL> else : <EOL> ri . _create_dvr_gateway . assert_called_once_with ( <EOL> ex_gw_port , interface_name ) <EOL> def _test_external_gateway_action ( self , action , router , dual_stack = False ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ex_net_id = _uuid ( ) <EOL> sn_port = self . snat_ports [ <NUM_LIT:1> ] <EOL> if router . get ( '<STR_LIT>' ) : <EOL> agent . conf . agent_mode = '<STR_LIT>' <EOL> agent . host = HOSTNAME <EOL> ri = dvr_router . DvrEdgeRouter ( agent , <EOL> HOSTNAME , <EOL> router [ '<STR_LIT:id>' ] , <EOL> router , <EOL> ** self . ri_kwargs ) <EOL> ri . _create_dvr_gateway = mock . Mock ( ) <EOL> ri . get_snat_interfaces = mock . Mock ( return_value = self . snat_ports ) <EOL> ri . snat_ports = self . snat_ports <EOL> ri . _create_snat_namespace ( ) <EOL> ri . fip_ns = agent . get_fip_ns ( ex_net_id ) <EOL> ri . internal_ports = self . snat_ports <EOL> else : <EOL> ri = l3router . RouterInfo ( <EOL> router [ '<STR_LIT:id>' ] , router , <EOL> ** self . ri_kwargs ) <EOL> ri . use_ipv6 = False <EOL> subnet_id = _uuid ( ) <EOL> fixed_ips = [ { '<STR_LIT>' : subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> } ] <EOL> subnets = [ { '<STR_LIT:id>' : subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> if dual_stack : <EOL> ri . use_ipv6 = True <EOL> subnet_id_v6 = _uuid ( ) <EOL> fixed_ips . append ( { '<STR_LIT>' : subnet_id_v6 , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:64> } ) <EOL> subnets . append ( { '<STR_LIT:id>' : subnet_id_v6 , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> ex_gw_port = { '<STR_LIT>' : fixed_ips , <EOL> '<STR_LIT>' : subnets , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : ex_net_id , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> ex_gw_port_no_sub = { '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : ex_net_id , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> interface_name = ri . get_external_device_name ( ex_gw_port [ '<STR_LIT:id>' ] ) <EOL> if action == '<STR_LIT>' : <EOL> self . _add_external_gateway ( ri , router , ex_gw_port , interface_name , <EOL> use_fake_fip = True , <EOL> dual_stack = dual_stack ) <EOL> elif action == '<STR_LIT>' : <EOL> ri . use_ipv6 = True <EOL> self . _add_external_gateway ( ri , router , ex_gw_port_no_sub , <EOL> interface_name , <EOL> no_subnet = True ) <EOL> elif action == '<STR_LIT>' : <EOL> ri . use_ipv6 = True <EOL> self . conf . set_override ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> if dual_stack : <EOL> use_fake_fip = True <EOL> del ex_gw_port [ '<STR_LIT>' ] [ - <NUM_LIT:1> ] <EOL> del ex_gw_port [ '<STR_LIT>' ] [ - <NUM_LIT:1> ] <EOL> else : <EOL> use_fake_fip = False <EOL> ex_gw_port = ex_gw_port_no_sub <EOL> self . _add_external_gateway ( ri , router , ex_gw_port , <EOL> interface_name , no_subnet = True , <EOL> no_sub_gw = '<STR_LIT>' , <EOL> use_fake_fip = use_fake_fip , <EOL> dual_stack = dual_stack ) <EOL> elif action == '<STR_LIT>' : <EOL> self . device_exists . return_value = True <EOL> ri . get_snat_port_for_internal_port = mock . Mock ( <EOL> return_value = sn_port ) <EOL> ri . _snat_redirect_remove = mock . Mock ( ) <EOL> ri . external_gateway_removed ( ex_gw_port , interface_name ) <EOL> if not router . get ( '<STR_LIT>' ) : <EOL> self . mock_driver . unplug . assert_called_once_with ( <EOL> interface_name , <EOL> bridge = agent . conf . external_network_bridge , <EOL> namespace = mock . ANY , <EOL> prefix = mock . ANY ) <EOL> else : <EOL> ri . _snat_redirect_remove . assert_called_with ( <EOL> sn_port , sn_port , <EOL> ri . get_internal_device_name ( sn_port [ '<STR_LIT:id>' ] ) ) <EOL> ri . get_snat_port_for_internal_port . assert_called_with ( <EOL> mock . ANY , ri . snat_ports ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" % action ) <EOL> def _test_external_gateway_updated ( self , dual_stack = False ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . use_ipv6 = False <EOL> interface_name , ex_gw_port = l3_test_common . prepare_ext_gw_test ( <EOL> self , ri , dual_stack = dual_stack ) <EOL> fake_fip = { '<STR_LIT>' : [ { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _uuid ( ) } ] } <EOL> router [ l3_constants . FLOATINGIP_KEY ] = fake_fip [ '<STR_LIT>' ] <EOL> ri . external_gateway_updated ( ex_gw_port , interface_name ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . mock_driver . plug . call_count ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . mock_driver . init_router_port . call_count ) <EOL> exp_arp_calls = [ mock . call ( ri . ns_name , interface_name , <EOL> '<STR_LIT>' , mock . ANY ) ] <EOL> if dual_stack : <EOL> ri . use_ipv6 = True <EOL> exp_arp_calls += [ mock . call ( ri . ns_name , interface_name , <EOL> '<STR_LIT>' , mock . ANY ) ] <EOL> self . send_adv_notif . assert_has_calls ( exp_arp_calls ) <EOL> ip_cidrs = [ '<STR_LIT>' ] <EOL> gateway_ips = [ '<STR_LIT>' ] <EOL> if dual_stack : <EOL> ip_cidrs . append ( '<STR_LIT>' ) <EOL> gateway_ips . append ( '<STR_LIT>' ) <EOL> kwargs = { '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' + router [ '<STR_LIT:id>' ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT>' : True } <EOL> self . mock_driver . init_router_port . assert_called_with ( interface_name , <EOL> ip_cidrs , <EOL> ** kwargs ) <EOL> def test_external_gateway_updated ( self ) : <EOL> self . _test_external_gateway_updated ( ) <EOL> def test_external_gateway_updated_dual_stack ( self ) : <EOL> self . _test_external_gateway_updated ( dual_stack = True ) <EOL> def _test_ext_gw_updated_dvr_edge_router ( self , host_match , <EOL> snat_hosted_before = True ) : <EOL> """<STR_LIT>""" <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> ri = dvr_router . DvrEdgeRouter ( mock . Mock ( ) , <EOL> HOSTNAME , <EOL> router [ '<STR_LIT:id>' ] , <EOL> router , <EOL> ** self . ri_kwargs ) <EOL> if snat_hosted_before : <EOL> ri . _create_snat_namespace ( ) <EOL> snat_ns_name = ri . snat_namespace . name <EOL> else : <EOL> self . assertIsNone ( ri . snat_namespace ) <EOL> interface_name , ex_gw_port = l3_test_common . prepare_ext_gw_test ( self , <EOL> ri ) <EOL> ri . _external_gateway_added = mock . Mock ( ) <EOL> router [ '<STR_LIT>' ] = ri . host if host_match else ( ri . host + '<STR_LIT:foo>' ) <EOL> ri . external_gateway_updated ( ex_gw_port , interface_name ) <EOL> if not host_match : <EOL> self . assertFalse ( ri . _external_gateway_added . called ) <EOL> if snat_hosted_before : <EOL> self . mock_driver . unplug . assert_called_with ( <EOL> interface_name , <EOL> bridge = self . conf . external_network_bridge , <EOL> namespace = snat_ns_name , <EOL> prefix = l3_agent . EXTERNAL_DEV_PREFIX ) <EOL> self . assertIsNone ( ri . snat_namespace ) <EOL> else : <EOL> if not snat_hosted_before : <EOL> self . assertIsNotNone ( ri . snat_namespace ) <EOL> self . assertTrue ( ri . _external_gateway_added . called ) <EOL> def test_ext_gw_updated_dvr_edge_router ( self ) : <EOL> self . _test_ext_gw_updated_dvr_edge_router ( host_match = True ) <EOL> def test_ext_gw_updated_dvr_edge_router_host_mismatch ( self ) : <EOL> self . _test_ext_gw_updated_dvr_edge_router ( host_match = False ) <EOL> def test_ext_gw_updated_dvr_dvr_edge_router_snat_rescheduled ( self ) : <EOL> self . _test_ext_gw_updated_dvr_edge_router ( host_match = True , <EOL> snat_hosted_before = False ) <EOL> def test_agent_add_external_gateway ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , router ) <EOL> def test_agent_add_external_gateway_dual_stack ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , router , dual_stack = True ) <EOL> def test_agent_add_external_gateway_dist ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> router [ '<STR_LIT>' ] = True <EOL> router [ '<STR_LIT>' ] = HOSTNAME <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , router ) <EOL> def test_agent_add_external_gateway_dist_dual_stack ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> router [ '<STR_LIT>' ] = True <EOL> router [ '<STR_LIT>' ] = HOSTNAME <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , router , dual_stack = True ) <EOL> def test_agent_add_external_gateway_no_subnet ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> , <EOL> v6_ext_gw_with_sub = False ) <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , router ) <EOL> def test_agent_add_external_gateway_no_subnet_with_ipv6_gw ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> , <EOL> v6_ext_gw_with_sub = False ) <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , router ) <EOL> def test_agent_add_external_gateway_dual_stack_no_subnet_w_ipv6_gw ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> , <EOL> v6_ext_gw_with_sub = False ) <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , <EOL> router , dual_stack = True ) <EOL> def test_agent_remove_external_gateway ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , router ) <EOL> def test_agent_remove_external_gateway_dual_stack ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , router , dual_stack = True ) <EOL> def test_agent_remove_external_gateway_dist ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> router [ '<STR_LIT>' ] = True <EOL> router [ '<STR_LIT>' ] = HOSTNAME <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , router ) <EOL> def test_agent_remove_external_gateway_dist_dual_stack ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> router [ '<STR_LIT>' ] = True <EOL> router [ '<STR_LIT>' ] = HOSTNAME <EOL> self . _test_external_gateway_action ( '<STR_LIT>' , router , dual_stack = True ) <EOL> def _verify_snat_mangle_rules ( self , nat_rules , mangle_rules , router , <EOL> negate = False ) : <EOL> interfaces = router [ l3_constants . INTERFACE_KEY ] <EOL> source_cidrs = [ ] <EOL> for iface in interfaces : <EOL> for subnet in iface [ '<STR_LIT>' ] : <EOL> prefix = subnet [ '<STR_LIT>' ] . split ( '<STR_LIT:/>' ) [ <NUM_LIT:1> ] <EOL> source_cidr = "<STR_LIT>" % ( iface [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> prefix ) <EOL> source_cidrs . append ( source_cidr ) <EOL> source_nat_ip = router [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> interface_name = ( '<STR_LIT>' % router [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) [ : <NUM_LIT> ] <EOL> expected_rules = [ <EOL> '<STR_LIT>' % <EOL> ( interface_name , interface_name ) , <EOL> '<STR_LIT>' % ( interface_name , source_nat_ip ) , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> ( n_const . ROUTER_MARK_MASK , source_nat_ip ) ] <EOL> for r in nat_rules : <EOL> if negate : <EOL> self . assertNotIn ( r . rule , expected_rules ) <EOL> else : <EOL> self . assertIn ( r . rule , expected_rules ) <EOL> expected_rules = [ <EOL> '<STR_LIT>' % <EOL> ( interface_name , n_const . ROUTER_MARK_MASK ) , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> ( interface_name , <EOL> l3router . ADDRESS_SCOPE_MARK_MASK , <EOL> l3router . ADDRESS_SCOPE_MARK_MASK , <EOL> l3router . ADDRESS_SCOPE_MARK_MASK ) ] <EOL> for r in mangle_rules : <EOL> if negate : <EOL> self . assertNotIn ( r . rule , expected_rules ) <EOL> else : <EOL> self . assertIn ( r . rule , expected_rules ) <EOL> def test_get_snat_port_for_internal_port ( self ) : <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:4> ) <EOL> ri = dvr_router . DvrEdgeRouter ( mock . sentinel . agent , <EOL> HOSTNAME , <EOL> router [ '<STR_LIT:id>' ] , <EOL> router , <EOL> ** self . ri_kwargs ) <EOL> test_port = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : l3_test_common . get_subnet_id ( <EOL> router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] ) , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] } <EOL> internal_ports = ri . router . get ( l3_constants . INTERFACE_KEY , [ ] ) <EOL> with mock . patch . object ( ri , '<STR_LIT>' ) as get_interfaces : <EOL> get_interfaces . return_value = [ test_port ] <EOL> res_port = ri . get_snat_port_for_internal_port ( internal_ports [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( test_port , res_port ) <EOL> test_port [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = <NUM_LIT> <EOL> res_ip = ri . get_snat_port_for_internal_port ( internal_ports [ <NUM_LIT:0> ] ) <EOL> self . assertNotEqual ( test_port , res_ip ) <EOL> self . assertIsNone ( res_ip ) <EOL> def test_process_cent_router ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> self . _test_process_router ( ri , agent ) <EOL> def test_process_dist_router ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = dvr_router . DvrEdgeRouter ( agent , <EOL> HOSTNAME , <EOL> router [ '<STR_LIT:id>' ] , <EOL> router , <EOL> ** self . ri_kwargs ) <EOL> subnet_id = l3_test_common . get_subnet_id ( <EOL> router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] ) <EOL> ri . router [ '<STR_LIT>' ] = True <EOL> ri . router [ '<STR_LIT>' ] = [ { <EOL> '<STR_LIT>' : [ { '<STR_LIT>' : subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] } ] <EOL> ri . router [ '<STR_LIT>' ] = None <EOL> self . _test_process_router ( ri , agent ) <EOL> def _test_process_router ( self , ri , agent ) : <EOL> router = ri . router <EOL> agent . host = HOSTNAME <EOL> fake_fip_id = '<STR_LIT>' <EOL> ri . create_dvr_fip_interfaces = mock . Mock ( ) <EOL> ri . process_floating_ip_addresses = mock . Mock ( ) <EOL> ri . process_floating_ip_nat_rules = mock . Mock ( ) <EOL> ri . process_floating_ip_addresses . return_value = { <EOL> fake_fip_id : '<STR_LIT>' } <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> ri . external_gateway_updated = mock . Mock ( ) <EOL> ri . process_address_scope = mock . Mock ( ) <EOL> fake_floatingips1 = { '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : fake_fip_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT:host>' : HOSTNAME } ] } <EOL> ri . process ( agent ) <EOL> ri . process_floating_ip_addresses . assert_called_with ( mock . ANY ) <EOL> ri . process_floating_ip_addresses . reset_mock ( ) <EOL> ri . process_floating_ip_nat_rules . assert_called_with ( ) <EOL> ri . process_floating_ip_nat_rules . reset_mock ( ) <EOL> ri . external_gateway_added . reset_mock ( ) <EOL> fake_floatingips2 = copy . deepcopy ( fake_floatingips1 ) <EOL> fake_floatingips2 [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> router [ l3_constants . FLOATINGIP_KEY ] = fake_floatingips2 [ '<STR_LIT>' ] <EOL> ri . process ( agent ) <EOL> ri . process_floating_ip_addresses . assert_called_with ( mock . ANY ) <EOL> ri . process_floating_ip_addresses . reset_mock ( ) <EOL> ri . process_floating_ip_nat_rules . assert_called_with ( ) <EOL> ri . process_floating_ip_nat_rules . reset_mock ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , ri . external_gateway_added . call_count ) <EOL> self . assertEqual ( <NUM_LIT:0> , ri . external_gateway_updated . call_count ) <EOL> ri . external_gateway_added . reset_mock ( ) <EOL> ri . external_gateway_updated . reset_mock ( ) <EOL> new_gw_port = copy . deepcopy ( ri . router [ '<STR_LIT>' ] ) <EOL> ri . router [ '<STR_LIT>' ] = new_gw_port <EOL> old_ip = ( netaddr . IPAddress ( ri . router [ '<STR_LIT>' ] <EOL> [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) ) <EOL> ri . router [ '<STR_LIT>' ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = str ( old_ip + <NUM_LIT:1> ) <EOL> ri . process ( agent ) <EOL> ri . process_floating_ip_addresses . reset_mock ( ) <EOL> ri . process_floating_ip_nat_rules . reset_mock ( ) <EOL> self . assertEqual ( <NUM_LIT:0> , ri . external_gateway_added . call_count ) <EOL> self . assertEqual ( <NUM_LIT:1> , ri . external_gateway_updated . call_count ) <EOL> del router [ l3_constants . FLOATINGIP_KEY ] <EOL> ri . process ( agent ) <EOL> ri . process_floating_ip_addresses . assert_called_with ( mock . ANY ) <EOL> ri . process_floating_ip_addresses . reset_mock ( ) <EOL> ri . process_floating_ip_nat_rules . assert_called_with ( ) <EOL> ri . process_floating_ip_nat_rules . reset_mock ( ) <EOL> del router [ l3_constants . INTERFACE_KEY ] <EOL> del router [ '<STR_LIT>' ] <EOL> ri . process ( agent ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . send_adv_notif . call_count ) <EOL> distributed = ri . router . get ( '<STR_LIT>' , False ) <EOL> self . assertEqual ( distributed , ri . process_floating_ip_addresses . called ) <EOL> self . assertEqual ( distributed , ri . process_floating_ip_nat_rules . called ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def _test_process_floating_ip_addresses_add ( self , ri , agent , IPDevice ) : <EOL> floating_ips = ri . get_floating_ips ( ) <EOL> fip_id = floating_ips [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> IPDevice . return_value = device = mock . Mock ( ) <EOL> device . addr . list . return_value = [ ] <EOL> ri . iptables_manager . ipv4 [ '<STR_LIT>' ] = mock . MagicMock ( ) <EOL> ex_gw_port = { '<STR_LIT:id>' : _uuid ( ) , '<STR_LIT>' : mock . sentinel . ext_net_id } <EOL> ri . add_floating_ip = mock . Mock ( <EOL> return_value = l3_constants . FLOATINGIP_STATUS_ACTIVE ) <EOL> with mock . patch . object ( lla . LinkLocalAllocator , '<STR_LIT>' ) : <EOL> if ri . router [ '<STR_LIT>' ] : <EOL> ri . fip_ns = agent . get_fip_ns ( ex_gw_port [ '<STR_LIT>' ] ) <EOL> ri . create_dvr_fip_interfaces ( ex_gw_port ) <EOL> fip_statuses = ri . process_floating_ip_addresses ( <EOL> mock . sentinel . interface_name ) <EOL> self . assertEqual ( { fip_id : l3_constants . FLOATINGIP_STATUS_ACTIVE } , <EOL> fip_statuses ) <EOL> ri . add_floating_ip . assert_called_once_with ( <EOL> floating_ips [ <NUM_LIT:0> ] , mock . sentinel . interface_name , device ) <EOL> @ mock . patch . object ( lla . LinkLocalAllocator , '<STR_LIT>' ) <EOL> def test_create_dvr_fip_interfaces_if_fipnamespace_exist ( self , lla_write ) : <EOL> fake_network_id = _uuid ( ) <EOL> subnet_id = _uuid ( ) <EOL> fake_floatingips = { '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT:host>' : HOSTNAME } ] } <EOL> agent_gateway_port = ( <EOL> [ { '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : subnet_id } ] , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : fake_network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> ) <EOL> router = l3_test_common . prepare_router_data ( enable_snat = True ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = fake_floatingips [ '<STR_LIT>' ] <EOL> router [ n_const . FLOATINGIP_AGENT_INTF_KEY ] = agent_gateway_port <EOL> router [ '<STR_LIT>' ] = True <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = dvr_router . DvrEdgeRouter ( <EOL> agent , HOSTNAME , router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ext_gw_port = ri . router . get ( '<STR_LIT>' ) <EOL> ri . fip_ns = agent . get_fip_ns ( ext_gw_port [ '<STR_LIT>' ] ) <EOL> ri . dist_fip_count = <NUM_LIT:0> <EOL> agent . process_router_add = mock . Mock ( ) <EOL> ri . fip_ns . create_rtr_2_fip_link = mock . Mock ( ) <EOL> with mock . patch . object ( ri , '<STR_LIT>' ) as fips , mock . patch . object ( ri . fip_ns , <EOL> '<STR_LIT>' ) as create_fip , mock . patch . object ( ri , '<STR_LIT>' <EOL> ) as fip_gw_port : <EOL> fips . return_value = fake_floatingips <EOL> fip_gw_port . return_value = agent_gateway_port [ <NUM_LIT:0> ] <EOL> ri . create_dvr_fip_interfaces ( ext_gw_port ) <EOL> self . assertTrue ( fip_gw_port . called ) <EOL> self . assertTrue ( fips . called ) <EOL> self . assertTrue ( create_fip . called ) <EOL> self . assertEqual ( agent_gateway_port [ <NUM_LIT:0> ] , <EOL> ri . fip_ns . agent_gateway_port ) <EOL> ri . floating_ip_added_dist ( fips , "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:1> , ri . dist_fip_count ) <EOL> ri . floating_ip_removed_dist ( "<STR_LIT>" ) <EOL> self . assertEqual ( <NUM_LIT:0> , ri . dist_fip_count ) <EOL> ri . create_dvr_fip_interfaces ( ext_gw_port ) <EOL> self . assertTrue ( fip_gw_port . called ) <EOL> self . assertTrue ( fips . called ) <EOL> create_fip . assert_called_once_with ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , ri . fip_ns . create_rtr_2_fip_link . call_count ) <EOL> @ mock . patch . object ( lla . LinkLocalAllocator , '<STR_LIT>' ) <EOL> def test_create_dvr_fip_interfaces_for_late_binding ( self , lla_write ) : <EOL> fake_network_id = _uuid ( ) <EOL> fake_subnet_id = _uuid ( ) <EOL> fake_floatingips = { '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT:host>' : HOSTNAME } ] } <EOL> agent_gateway_port = ( <EOL> { '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : fake_subnet_id } ] , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : fake_subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : fake_network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> router = l3_test_common . prepare_router_data ( enable_snat = True ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = fake_floatingips [ '<STR_LIT>' ] <EOL> router [ n_const . FLOATINGIP_AGENT_INTF_KEY ] = [ ] <EOL> router [ '<STR_LIT>' ] = True <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = dvr_router . DvrEdgeRouter ( <EOL> agent , HOSTNAME , router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ext_gw_port = ri . router . get ( '<STR_LIT>' ) <EOL> ri . fip_ns = agent . get_fip_ns ( ext_gw_port [ '<STR_LIT>' ] ) <EOL> ri . dist_fip_count = <NUM_LIT:0> <EOL> ri . fip_ns . subscribe = mock . Mock ( ) <EOL> with mock . patch . object ( agent . plugin_rpc , <EOL> '<STR_LIT>' ) as fip_gw_port : <EOL> fip_gw_port . return_value = agent_gateway_port <EOL> ri . create_dvr_fip_interfaces ( ext_gw_port ) <EOL> self . assertTrue ( fip_gw_port . called ) <EOL> self . assertEqual ( agent_gateway_port , <EOL> ri . fip_ns . agent_gateway_port ) <EOL> @ mock . patch . object ( lla . LinkLocalAllocator , '<STR_LIT>' ) <EOL> def test_create_dvr_fip_interfaces ( self , lla_write ) : <EOL> fake_network_id = _uuid ( ) <EOL> subnet_id = _uuid ( ) <EOL> fake_floatingips = { '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT:host>' : HOSTNAME } ] } <EOL> agent_gateway_port = ( <EOL> [ { '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : subnet_id } ] , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : fake_network_id , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> ) <EOL> router = l3_test_common . prepare_router_data ( enable_snat = True ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = fake_floatingips [ '<STR_LIT>' ] <EOL> router [ n_const . FLOATINGIP_AGENT_INTF_KEY ] = agent_gateway_port <EOL> router [ '<STR_LIT>' ] = True <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = dvr_router . DvrEdgeRouter ( <EOL> agent , HOSTNAME , router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ext_gw_port = ri . router . get ( '<STR_LIT>' ) <EOL> ri . fip_ns = agent . get_fip_ns ( ext_gw_port [ '<STR_LIT>' ] ) <EOL> ri . dist_fip_count = <NUM_LIT:0> <EOL> ri . fip_ns . subscribe = mock . Mock ( ) <EOL> ri . fip_ns . agent_router_gateway = mock . Mock ( ) <EOL> agent . process_router_add = mock . Mock ( ) <EOL> with mock . patch . object ( ri , '<STR_LIT>' ) as fips , mock . patch . object ( ri , '<STR_LIT>' <EOL> ) as fip_gw_port : <EOL> fips . return_value = fake_floatingips <EOL> fip_gw_port . return_value = agent_gateway_port [ <NUM_LIT:0> ] <EOL> ri . create_dvr_fip_interfaces ( ext_gw_port ) <EOL> self . assertTrue ( fip_gw_port . called ) <EOL> self . assertTrue ( fips . called ) <EOL> self . assertEqual ( agent_gateway_port [ <NUM_LIT:0> ] , <EOL> ri . fip_ns . agent_gateway_port ) <EOL> self . assertTrue ( ri . rtr_fip_subnet ) <EOL> @ mock . patch . object ( lla . LinkLocalAllocator , '<STR_LIT>' ) <EOL> def test_create_dvr_fip_interfaces_for_restart_l3agent_case ( self , <EOL> lla_write ) : <EOL> fake_floatingips = { '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT:host>' : HOSTNAME } ] } <EOL> agent_gateway_port = ( <EOL> [ { '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> ) <EOL> router = l3_test_common . prepare_router_data ( enable_snat = True ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = fake_floatingips [ '<STR_LIT>' ] <EOL> router [ n_const . FLOATINGIP_AGENT_INTF_KEY ] = agent_gateway_port <EOL> router [ '<STR_LIT>' ] = True <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = dvr_router . DvrEdgeRouter ( <EOL> agent , HOSTNAME , router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ext_gw_port = ri . router . get ( '<STR_LIT>' ) <EOL> ri . fip_ns = agent . get_fip_ns ( ext_gw_port [ '<STR_LIT>' ] ) <EOL> ri . fip_ns . subscribe = mock . Mock ( return_value = True ) <EOL> ri . fip_ns . agent_router_gateway = mock . Mock ( ) <EOL> ri . rtr_fip_subnet = None <EOL> ri . dist_fip_count = <NUM_LIT:0> <EOL> with mock . patch . object ( ri , '<STR_LIT>' ) as fips , mock . patch . object ( ri , '<STR_LIT>' <EOL> ) as fip_gw_port : <EOL> fips . return_value = fake_floatingips <EOL> fip_gw_port . return_value = agent_gateway_port [ <NUM_LIT:0> ] <EOL> ri . create_dvr_fip_interfaces ( ext_gw_port ) <EOL> self . assertTrue ( fip_gw_port . called ) <EOL> self . assertTrue ( fips . called ) <EOL> self . assertEqual ( agent_gateway_port [ <NUM_LIT:0> ] , <EOL> ri . fip_ns . agent_gateway_port ) <EOL> self . assertTrue ( ri . rtr_fip_subnet ) <EOL> def test_process_router_cent_floating_ip_add ( self ) : <EOL> fake_floatingips = { '<STR_LIT>' : [ <EOL> { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT:host>' : HOSTNAME } ] } <EOL> router = l3_test_common . prepare_router_data ( enable_snat = True ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = fake_floatingips [ '<STR_LIT>' ] <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . iptables_manager . ipv4 [ '<STR_LIT>' ] = mock . MagicMock ( ) <EOL> ri . get_external_device_name = mock . Mock ( return_value = '<STR_LIT>' ) <EOL> self . _test_process_floating_ip_addresses_add ( ri , agent ) <EOL> def test_process_router_snat_disabled ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( enable_snat = True ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> ri . process ( agent ) <EOL> orig_nat_rules = ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules [ : ] <EOL> orig_mangle_rules = ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules [ : ] <EOL> router [ '<STR_LIT>' ] = False <EOL> ri . router = router <EOL> ri . process ( agent ) <EOL> nat_rules_delta = [ r for r in orig_nat_rules <EOL> if r not in ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( nat_rules_delta ) ) <EOL> mangle_rules_delta = [ <EOL> r for r in orig_mangle_rules <EOL> if r not in ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( mangle_rules_delta ) ) <EOL> self . _verify_snat_mangle_rules ( nat_rules_delta , mangle_rules_delta , <EOL> router ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . send_adv_notif . call_count ) <EOL> def test_process_router_snat_enabled ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( enable_snat = False ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> ri . process ( agent ) <EOL> orig_nat_rules = ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules [ : ] <EOL> orig_mangle_rules = ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules [ : ] <EOL> router [ '<STR_LIT>' ] = True <EOL> ri . router = router <EOL> ri . process ( agent ) <EOL> nat_rules_delta = [ r for r in ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules <EOL> if r not in orig_nat_rules ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( nat_rules_delta ) ) <EOL> mangle_rules_delta = [ <EOL> r for r in ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules <EOL> if r not in orig_mangle_rules ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( mangle_rules_delta ) ) <EOL> self . _verify_snat_mangle_rules ( nat_rules_delta , mangle_rules_delta , <EOL> router ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . send_adv_notif . call_count ) <EOL> def _test_update_routing_table ( self , is_snat_host = True ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> uuid = router [ '<STR_LIT:id>' ] <EOL> s_netns = '<STR_LIT>' + uuid <EOL> q_netns = '<STR_LIT>' + uuid <EOL> fake_route1 = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> calls = [ mock . call ( '<STR_LIT:replace>' , fake_route1 , q_netns ) ] <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = dvr_router . DvrEdgeRouter ( <EOL> agent , <EOL> HOSTNAME , <EOL> uuid , <EOL> router , <EOL> ** self . ri_kwargs ) <EOL> ri . _update_routing_table = mock . Mock ( ) <EOL> with mock . patch . object ( ri , '<STR_LIT>' ) as snat_host : <EOL> snat_host . return_value = is_snat_host <EOL> ri . update_routing_table ( '<STR_LIT:replace>' , fake_route1 ) <EOL> if is_snat_host : <EOL> ri . _update_routing_table ( '<STR_LIT:replace>' , fake_route1 , s_netns ) <EOL> calls += [ mock . call ( '<STR_LIT:replace>' , fake_route1 , s_netns ) ] <EOL> ri . _update_routing_table . assert_has_calls ( calls , any_order = True ) <EOL> def test_process_update_snat_routing_table ( self ) : <EOL> self . _test_update_routing_table ( ) <EOL> def test_process_not_update_snat_routing_table ( self ) : <EOL> self . _test_update_routing_table ( is_snat_host = False ) <EOL> def test_process_router_interface_added ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> ri . process ( agent ) <EOL> l3_test_common . router_append_interface ( router ) <EOL> ri . router = router <EOL> ri . process ( agent ) <EOL> self . assertEqual ( <NUM_LIT:2> , self . send_adv_notif . call_count ) <EOL> def _test_process_ipv6_only_or_dual_stack_gw ( self , dual_stack = False ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( ip_version = <NUM_LIT:6> , <EOL> dual_stack = dual_stack ) <EOL> gw_port = router [ '<STR_LIT>' ] <EOL> router [ '<STR_LIT>' ] = None <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> orig_nat_rules = ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules [ : ] <EOL> router [ '<STR_LIT>' ] = gw_port <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> p = ri . external_gateway_nat_fip_rules <EOL> s = ri . external_gateway_nat_snat_rules <EOL> attrs_to_mock = dict ( <EOL> [ ( a , mock . DEFAULT ) for a in <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] ] <EOL> ) <EOL> with mock . patch . multiple ( ri , ** attrs_to_mock ) as mocks : <EOL> mocks [ '<STR_LIT>' ] . side_effect = p <EOL> mocks [ '<STR_LIT>' ] . side_effect = s <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> new_nat_rules = ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules [ : ] <EOL> if dual_stack : <EOL> self . assertTrue ( <EOL> mocks [ '<STR_LIT>' ] . called ) <EOL> self . assertTrue ( <EOL> mocks [ '<STR_LIT>' ] . called ) <EOL> self . assertNotEqual ( orig_nat_rules , new_nat_rules ) <EOL> else : <EOL> self . assertFalse ( <EOL> mocks [ '<STR_LIT>' ] . called ) <EOL> self . assertFalse ( <EOL> mocks [ '<STR_LIT>' ] . called ) <EOL> self . assertEqual ( orig_nat_rules , new_nat_rules ) <EOL> def test_process_ipv6_only_gw ( self ) : <EOL> self . _test_process_ipv6_only_or_dual_stack_gw ( ) <EOL> def test_process_dual_stack_gw ( self ) : <EOL> self . _test_process_ipv6_only_or_dual_stack_gw ( dual_stack = True ) <EOL> def _process_router_ipv6_interface_added ( <EOL> self , router , ra_mode = None , addr_mode = None ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> ri . process ( agent ) <EOL> orig_nat_rules = ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules [ : ] <EOL> l3_test_common . router_append_interface ( router , count = <NUM_LIT:1> , <EOL> ip_version = <NUM_LIT:6> , ra_mode = ra_mode , <EOL> addr_mode = addr_mode ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> nat_rules_delta = [ r for r in ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules <EOL> if r not in orig_nat_rules ] <EOL> self . assertFalse ( nat_rules_delta ) <EOL> return ri <EOL> def _radvd_expected_call_external_process ( self , ri , enable = True ) : <EOL> expected_calls = [ mock . call ( uuid = ri . router [ '<STR_LIT:id>' ] , <EOL> service = '<STR_LIT>' , <EOL> default_cmd_callback = mock . ANY , <EOL> namespace = ri . ns_name , <EOL> conf = mock . ANY , <EOL> run_as_root = True ) ] <EOL> if enable : <EOL> expected_calls . append ( mock . call ( ) . enable ( reload_cfg = True ) ) <EOL> else : <EOL> expected_calls . append ( mock . call ( ) . disable ( ) ) <EOL> return expected_calls <EOL> def _process_router_ipv6_subnet_added ( self , router , <EOL> ipv6_subnet_modes = None , dns_nameservers = None , network_mtu = <NUM_LIT:0> ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> agent . external_gateway_added = mock . Mock ( ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> l3_test_common . router_append_subnet ( <EOL> router , <EOL> count = len ( ipv6_subnet_modes ) , <EOL> ip_version = <NUM_LIT:6> , <EOL> ipv6_subnet_modes = ipv6_subnet_modes , <EOL> dns_nameservers = dns_nameservers , <EOL> network_mtu = network_mtu ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> return ri <EOL> def _assert_ri_process_enabled ( self , ri ) : <EOL> """<STR_LIT>""" <EOL> expected_calls = self . _radvd_expected_call_external_process ( ri ) <EOL> self . assertEqual ( expected_calls , self . external_process . mock_calls ) <EOL> def _assert_ri_process_disabled ( self , ri ) : <EOL> """<STR_LIT>""" <EOL> expected_calls = self . _radvd_expected_call_external_process ( ri , False ) <EOL> self . assertEqual ( expected_calls , self . external_process . mock_calls ) <EOL> def test_process_router_ipv6_interface_added ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = self . _process_router_ipv6_interface_added ( router ) <EOL> self . _assert_ri_process_enabled ( ri ) <EOL> self . assertNotIn ( '<STR_LIT>' , self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) <EOL> def test_process_router_ipv6_slaac_interface_added ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = self . _process_router_ipv6_interface_added ( <EOL> router , ra_mode = n_const . IPV6_SLAAC ) <EOL> self . _assert_ri_process_enabled ( ri ) <EOL> radvd_config_str = self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> self . assertIn ( '<STR_LIT>' , radvd_config_str ) <EOL> self . assertIn ( '<STR_LIT>' , radvd_config_str ) <EOL> def test_process_router_ipv6_dhcpv6_stateful_interface_added ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = self . _process_router_ipv6_interface_added ( <EOL> router , ra_mode = n_const . DHCPV6_STATEFUL ) <EOL> self . _assert_ri_process_enabled ( ri ) <EOL> radvd_config_str = self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> self . assertIn ( '<STR_LIT>' , radvd_config_str ) <EOL> self . assertIn ( '<STR_LIT>' , radvd_config_str ) <EOL> def test_process_router_ipv6_subnets_added ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = self . _process_router_ipv6_subnet_added ( router , ipv6_subnet_modes = [ <EOL> { '<STR_LIT>' : n_const . IPV6_SLAAC , <EOL> '<STR_LIT>' : n_const . IPV6_SLAAC } , <EOL> { '<STR_LIT>' : n_const . DHCPV6_STATELESS , <EOL> '<STR_LIT>' : n_const . DHCPV6_STATELESS } , <EOL> { '<STR_LIT>' : n_const . DHCPV6_STATEFUL , <EOL> '<STR_LIT>' : n_const . DHCPV6_STATEFUL } ] ) <EOL> self . _assert_ri_process_enabled ( ri ) <EOL> radvd_config_str = self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> self . assertEqual ( <NUM_LIT:3> , radvd_config_str . count ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , radvd_config_str . count ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , radvd_config_str . count ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , radvd_config_str . count ( "<STR_LIT>" ) ) <EOL> def test_process_router_ipv6_subnets_added_to_existing_port ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> agent . external_gateway_added = mock . Mock ( ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> l3_test_common . router_append_subnet ( <EOL> router , count = <NUM_LIT:1> , <EOL> ip_version = <NUM_LIT:6> , ipv6_subnet_modes = [ <EOL> { '<STR_LIT>' : n_const . IPV6_SLAAC , <EOL> '<STR_LIT>' : n_const . IPV6_SLAAC } ] ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> self . _assert_ri_process_enabled ( ri ) <EOL> radvd_config = self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . split ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( ri . internal_ports [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( ri . internal_ports [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , radvd_config . count ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , radvd_config . count ( "<STR_LIT>" ) ) <EOL> self . external_process . reset_mock ( ) <EOL> self . utils_replace_file . reset_mock ( ) <EOL> interface_id = router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:1> ] [ '<STR_LIT:id>' ] <EOL> l3_test_common . router_append_subnet ( <EOL> router , count = <NUM_LIT:1> , <EOL> ip_version = <NUM_LIT:6> , <EOL> ipv6_subnet_modes = [ <EOL> { '<STR_LIT>' : n_const . IPV6_SLAAC , <EOL> '<STR_LIT>' : n_const . IPV6_SLAAC } ] , <EOL> interface_id = interface_id ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> self . _assert_ri_process_enabled ( ri ) <EOL> radvd_config = self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . split ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( ri . internal_ports [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( ri . internal_ports [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , radvd_config . count ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , radvd_config . count ( "<STR_LIT>" ) ) <EOL> def test_process_router_ipv6v4_interface_added ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> ri . process ( agent ) <EOL> l3_test_common . router_append_interface ( router , count = <NUM_LIT:1> , ip_version = <NUM_LIT:4> ) <EOL> l3_test_common . router_append_interface ( router , count = <NUM_LIT:1> , ip_version = <NUM_LIT:6> ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> self . _assert_ri_process_enabled ( ri ) <EOL> def test_process_router_interface_removed ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> ri . process ( agent ) <EOL> del router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:1> ] <EOL> ri . router = router <EOL> ri . process ( agent ) <EOL> self . assertEqual ( <NUM_LIT:2> , self . send_adv_notif . call_count ) <EOL> def test_process_router_ipv6_interface_removed ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> l3_test_common . router_append_interface ( router , count = <NUM_LIT:1> , ip_version = <NUM_LIT:6> ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> self . _assert_ri_process_enabled ( ri ) <EOL> self . external_process . reset_mock ( ) <EOL> self . process_monitor . reset_mock ( ) <EOL> del router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:1> ] <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> self . _assert_ri_process_disabled ( ri ) <EOL> def test_process_router_ipv6_subnet_removed ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> agent . external_gateway_added = mock . Mock ( ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> l3_test_common . router_append_subnet ( <EOL> router , count = <NUM_LIT:2> , ip_version = <NUM_LIT:6> , <EOL> ipv6_subnet_modes = ( [ { '<STR_LIT>' : n_const . IPV6_SLAAC , <EOL> '<STR_LIT>' : n_const . IPV6_SLAAC } ] <EOL> * <NUM_LIT:2> ) ) <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> self . _assert_ri_process_enabled ( ri ) <EOL> self . utils_replace_file . reset_mock ( ) <EOL> self . external_process . reset_mock ( ) <EOL> interfaces = copy . deepcopy ( router [ l3_constants . INTERFACE_KEY ] ) <EOL> del interfaces [ <NUM_LIT:1> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> del interfaces [ <NUM_LIT:1> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> router [ l3_constants . INTERFACE_KEY ] = interfaces <EOL> self . _process_router_instance_for_agent ( agent , ri , router ) <EOL> self . _assert_ri_process_enabled ( ri ) <EOL> radvd_config = self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] . split ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( ri . internal_ports [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( ri . internal_ports [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , radvd_config . count ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , radvd_config . count ( "<STR_LIT>" ) ) <EOL> def test_process_router_internal_network_added_unexpected_error ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> with mock . patch . object ( <EOL> ri , <EOL> '<STR_LIT>' ) as internal_network_added : <EOL> internal_network_added . side_effect = RuntimeError <EOL> self . assertRaises ( RuntimeError , ri . process , agent ) <EOL> self . assertNotIn ( <EOL> router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] , ri . internal_ports ) <EOL> internal_network_added . side_effect = None <EOL> ri . process ( agent ) <EOL> self . assertIn ( <EOL> router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] , ri . internal_ports ) <EOL> def test_process_router_internal_network_removed_unexpected_error ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> ri . process ( agent ) <EOL> with mock . patch . object ( <EOL> ri , <EOL> '<STR_LIT>' ) as internal_net_removed : <EOL> internal_net_removed . side_effect = RuntimeError <EOL> ri . internal_ports [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = False <EOL> self . assertRaises ( RuntimeError , ri . process , agent ) <EOL> self . assertIn ( <EOL> router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] , ri . internal_ports ) <EOL> internal_net_removed . side_effect = None <EOL> ri . process ( agent ) <EOL> self . assertNotIn ( <EOL> router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] , ri . internal_ports ) <EOL> def test_process_router_floatingip_nochange ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:1> ) <EOL> fip1 = { '<STR_LIT:id>' : _uuid ( ) , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] } <EOL> fip2 = copy . copy ( fip1 ) <EOL> fip2 . update ( { '<STR_LIT:id>' : _uuid ( ) , '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = [ fip1 , fip2 ] <EOL> ri = legacy_router . LegacyRouter ( router [ '<STR_LIT:id>' ] , router , <EOL> ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> with mock . patch . object ( <EOL> agent . plugin_rpc , '<STR_LIT>' <EOL> ) as mock_update_fip_status , mock . patch . object ( ri , '<STR_LIT>' ) as mock_get_cidrs : <EOL> mock_get_cidrs . return_value = set ( <EOL> [ fip1 [ '<STR_LIT>' ] + '<STR_LIT>' ] ) <EOL> ri . process ( agent ) <EOL> mock_update_fip_status . assert_called_once_with ( <EOL> mock . ANY , ri . router_id , { fip2 [ '<STR_LIT:id>' ] : '<STR_LIT>' } ) <EOL> def test_process_router_floatingip_status_update_if_processed ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:1> ) <EOL> fip1 = { '<STR_LIT:id>' : _uuid ( ) , '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] } <EOL> fip2 = copy . copy ( fip1 ) <EOL> fip2 . update ( { '<STR_LIT:id>' : _uuid ( ) , '<STR_LIT:status>' : '<STR_LIT>' , } ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = [ fip1 , fip2 ] <EOL> ri = legacy_router . LegacyRouter ( router [ '<STR_LIT:id>' ] , router , <EOL> ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> with mock . patch . object ( <EOL> agent . plugin_rpc , '<STR_LIT>' <EOL> ) as mock_update_fip_status , mock . patch . object ( ri , '<STR_LIT>' ) as mock_get_cidrs : <EOL> mock_get_cidrs . return_value = set ( ) <EOL> ri . process ( agent ) <EOL> mock_update_fip_status . assert_called_once_with ( <EOL> mock . ANY , ri . router_id , { fip1 [ '<STR_LIT:id>' ] : '<STR_LIT>' , <EOL> fip2 [ '<STR_LIT:id>' ] : '<STR_LIT>' } ) <EOL> def test_process_router_floatingip_disabled ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> with mock . patch . object ( <EOL> agent . plugin_rpc , <EOL> '<STR_LIT>' ) as mock_update_fip_status : <EOL> fip_id = _uuid ( ) <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:1> ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = [ <EOL> { '<STR_LIT:id>' : fip_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] } ] <EOL> ri = legacy_router . LegacyRouter ( router [ '<STR_LIT:id>' ] , <EOL> router , <EOL> ** self . ri_kwargs ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> ri . process ( agent ) <EOL> mock_update_fip_status . assert_called_once_with ( <EOL> mock . ANY , ri . router_id , <EOL> { fip_id : l3_constants . FLOATINGIP_STATUS_ACTIVE } ) <EOL> mock_update_fip_status . reset_mock ( ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = [ ] <EOL> ri . router = router <EOL> ri . process ( agent ) <EOL> mock_update_fip_status . assert_called_once_with ( <EOL> mock . ANY , ri . router_id , <EOL> { fip_id : l3_constants . FLOATINGIP_STATUS_DOWN } ) <EOL> def test_process_router_floatingip_exception ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> with mock . patch . object ( <EOL> agent . plugin_rpc , <EOL> '<STR_LIT>' ) as mock_update_fip_status : <EOL> fip_id = _uuid ( ) <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:1> ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = [ <EOL> { '<STR_LIT:id>' : fip_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] } ] <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . process_floating_ip_addresses = mock . Mock ( <EOL> side_effect = RuntimeError ) <EOL> ri . external_gateway_added = mock . Mock ( ) <EOL> ri . process ( agent ) <EOL> mock_update_fip_status . assert_called_once_with ( <EOL> mock . ANY , ri . router_id , <EOL> { fip_id : l3_constants . FLOATINGIP_STATUS_ERROR } ) <EOL> def test_process_external_iptables_exception ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> with mock . patch . object ( <EOL> agent . plugin_rpc , <EOL> '<STR_LIT>' ) as mock_update_fip_status : <EOL> fip_id = _uuid ( ) <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:1> ) <EOL> router [ l3_constants . FLOATINGIP_KEY ] = [ <EOL> { '<STR_LIT:id>' : fip_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] } ] <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . iptables_manager . _apply = mock . Mock ( side_effect = Exception ) <EOL> ri . process_external ( agent ) <EOL> mock_update_fip_status . assert_called_once_with ( <EOL> mock . ANY , ri . router_id , <EOL> { fip_id : l3_constants . FLOATINGIP_STATUS_ERROR } ) <EOL> self . assertEqual ( <NUM_LIT:1> , ri . iptables_manager . _apply . call_count ) <EOL> def test_handle_router_snat_rules_distributed_without_snat_manager ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = dvr_router . DvrEdgeRouter ( <EOL> agent , <EOL> HOSTNAME , <EOL> '<STR_LIT>' , <EOL> { } , <EOL> ** self . ri_kwargs ) <EOL> ri . iptables_manager = mock . MagicMock ( ) <EOL> ri . _is_this_snat_host = mock . Mock ( return_value = True ) <EOL> ri . get_ex_gw_port = mock . Mock ( return_value = None ) <EOL> ri . _handle_router_snat_rules ( None , mock . ANY ) <EOL> self . assertIsNone ( ri . snat_iptables_manager ) <EOL> self . assertFalse ( ri . iptables_manager . called ) <EOL> def test_handle_router_snat_rules_add_back_jump ( self ) : <EOL> ri = l3router . RouterInfo ( _uuid ( ) , { } , ** self . ri_kwargs ) <EOL> ri . iptables_manager = mock . MagicMock ( ) <EOL> port = { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } ] } <EOL> ri . _handle_router_snat_rules ( port , "<STR_LIT>" ) <EOL> nat = ri . iptables_manager . ipv4 [ '<STR_LIT>' ] <EOL> nat . empty_chain . assert_any_call ( '<STR_LIT>' ) <EOL> nat . add_rule . assert_any_call ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> for call in nat . mock_calls : <EOL> name , args , kwargs = call <EOL> if name == '<STR_LIT>' : <EOL> self . assertEqual ( ( '<STR_LIT>' , '<STR_LIT>' ) , args ) <EOL> self . assertEqual ( { } , kwargs ) <EOL> break <EOL> def test_handle_router_snat_rules_add_rules ( self ) : <EOL> ri = l3router . RouterInfo ( _uuid ( ) , { } , ** self . ri_kwargs ) <EOL> ex_gw_port = { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' } ] } <EOL> ri . router = { '<STR_LIT>' : False } <EOL> ri . _handle_router_snat_rules ( ex_gw_port , "<STR_LIT>" ) <EOL> nat_rules = list ( map ( str , ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules ) ) <EOL> wrap_name = ri . iptables_manager . wrap_name <EOL> jump_float_rule = "<STR_LIT>" % ( wrap_name , <EOL> wrap_name ) <EOL> snat_rule1 = ( "<STR_LIT>" ) % ( <EOL> wrap_name , ex_gw_port [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> snat_rule2 = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) % ( <EOL> wrap_name , n_const . ROUTER_MARK_MASK , <EOL> ex_gw_port [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertIn ( jump_float_rule , nat_rules ) <EOL> self . assertIn ( snat_rule1 , nat_rules ) <EOL> self . assertIn ( snat_rule2 , nat_rules ) <EOL> self . assertThat ( nat_rules . index ( jump_float_rule ) , <EOL> matchers . LessThan ( nat_rules . index ( snat_rule1 ) ) ) <EOL> mangle_rules = list ( map ( str , ri . iptables_manager . ipv4 [ '<STR_LIT>' ] . rules ) ) <EOL> mangle_rule = ( "<STR_LIT>" <EOL> "<STR_LIT>" % <EOL> ( wrap_name , n_const . ROUTER_MARK_MASK ) ) <EOL> self . assertIn ( mangle_rule , mangle_rules ) <EOL> def test_process_router_delete_stale_internal_devices ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> stale_devlist = [ l3_test_common . FakeDev ( '<STR_LIT>' ) , <EOL> l3_test_common . FakeDev ( '<STR_LIT>' ) ] <EOL> stale_devnames = [ dev . name for dev in stale_devlist ] <EOL> get_devices_return = [ ] <EOL> get_devices_return . extend ( stale_devlist ) <EOL> self . mock_ip . get_devices . return_value = get_devices_return <EOL> router = l3_test_common . prepare_router_data ( enable_snat = True , <EOL> num_internal_ports = <NUM_LIT:1> ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> internal_ports = ri . router . get ( l3_constants . INTERFACE_KEY , [ ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( internal_ports ) ) <EOL> internal_port = internal_ports [ <NUM_LIT:0> ] <EOL> with mock . patch . object ( ri , '<STR_LIT>' <EOL> ) as internal_network_removed , mock . patch . object ( ri , '<STR_LIT>' <EOL> ) as internal_network_added , mock . patch . object ( ri , '<STR_LIT>' <EOL> ) as external_gateway_removed , mock . patch . object ( ri , '<STR_LIT>' <EOL> ) as external_gateway_added : <EOL> ri . process ( agent ) <EOL> self . assertEqual ( <NUM_LIT:1> , external_gateway_added . call_count ) <EOL> self . assertFalse ( external_gateway_removed . called ) <EOL> self . assertFalse ( internal_network_removed . called ) <EOL> internal_network_added . assert_called_once_with ( internal_port ) <EOL> self . assertEqual ( len ( stale_devnames ) , <EOL> self . mock_driver . unplug . call_count ) <EOL> calls = [ mock . call ( stale_devname , <EOL> namespace = ri . ns_name , <EOL> prefix = l3_agent . INTERNAL_DEV_PREFIX ) <EOL> for stale_devname in stale_devnames ] <EOL> self . mock_driver . unplug . assert_has_calls ( calls , any_order = True ) <EOL> def test_process_router_delete_stale_external_devices ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> stale_devlist = [ l3_test_common . FakeDev ( '<STR_LIT>' ) ] <EOL> stale_devnames = [ dev . name for dev in stale_devlist ] <EOL> router = l3_test_common . prepare_router_data ( enable_snat = True , <EOL> num_internal_ports = <NUM_LIT:1> ) <EOL> del router [ '<STR_LIT>' ] <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> self . mock_ip . get_devices . return_value = stale_devlist <EOL> ri . process ( agent ) <EOL> self . mock_driver . unplug . assert_called_with ( <EOL> stale_devnames [ <NUM_LIT:0> ] , <EOL> bridge = "<STR_LIT>" , <EOL> namespace = ri . ns_name , <EOL> prefix = l3_agent . EXTERNAL_DEV_PREFIX ) <EOL> def test_router_deleted ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> agent . _queue = mock . Mock ( ) <EOL> agent . router_deleted ( None , FAKE_ID ) <EOL> self . assertEqual ( <NUM_LIT:1> , agent . _queue . add . call_count ) <EOL> def test_routers_updated ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> agent . _queue = mock . Mock ( ) <EOL> agent . routers_updated ( None , [ FAKE_ID ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , agent . _queue . add . call_count ) <EOL> def test_removed_from_agent ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> agent . _queue = mock . Mock ( ) <EOL> agent . router_removed_from_agent ( None , { '<STR_LIT>' : FAKE_ID } ) <EOL> self . assertEqual ( <NUM_LIT:1> , agent . _queue . add . call_count ) <EOL> def test_added_to_agent ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> agent . _queue = mock . Mock ( ) <EOL> agent . router_added_to_agent ( None , [ FAKE_ID ] ) <EOL> self . assertEqual ( <NUM_LIT:1> , agent . _queue . add . call_count ) <EOL> def test_destroy_namespace ( self ) : <EOL> namespace = '<STR_LIT>' <EOL> self . mock_ip . get_namespaces . return_value = [ namespace ] <EOL> self . mock_ip . get_devices . return_value = [ <EOL> l3_test_common . FakeDev ( '<STR_LIT>' ) , <EOL> l3_test_common . FakeDev ( '<STR_LIT>' ) ] <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ns = namespaces . RouterNamespace ( <EOL> '<STR_LIT:bar>' , self . conf , agent . driver , agent . use_ipv6 ) <EOL> ns . create ( ) <EOL> ns . delete ( ) <EOL> self . mock_driver . unplug . assert_called_once_with ( '<STR_LIT>' , <EOL> prefix = '<STR_LIT>' , <EOL> namespace = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . mock_ip . del_veth . assert_called_once_with ( '<STR_LIT>' ) <EOL> def test_destroy_router_namespace ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ns = namespaces . Namespace ( <EOL> '<STR_LIT>' , self . conf , agent . driver , agent . use_ipv6 ) <EOL> ns . create ( ) <EOL> ns . delete ( ) <EOL> self . mock_ip . netns . delete . assert_called_once_with ( "<STR_LIT>" ) <EOL> def _configure_metadata_proxy ( self , enableflag = True ) : <EOL> if not enableflag : <EOL> self . conf . set_override ( '<STR_LIT>' , False ) <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router_id = _uuid ( ) <EOL> router = { '<STR_LIT:id>' : router_id , <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : False } <EOL> driver = metadata_driver . MetadataDriver <EOL> with mock . patch . object ( <EOL> driver , '<STR_LIT>' ) as destroy_proxy : <EOL> with mock . patch . object ( <EOL> driver , '<STR_LIT>' ) as spawn_proxy : <EOL> agent . _process_added_router ( router ) <EOL> if enableflag : <EOL> spawn_proxy . assert_called_with ( <EOL> mock . ANY , <EOL> mock . ANY , <EOL> self . conf . metadata_port , <EOL> mock . ANY , <EOL> router_id = router_id <EOL> ) <EOL> else : <EOL> self . assertFalse ( spawn_proxy . call_count ) <EOL> agent . _router_removed ( router_id ) <EOL> if enableflag : <EOL> destroy_proxy . assert_called_with ( mock . ANY , <EOL> router_id , <EOL> mock . ANY ) <EOL> else : <EOL> self . assertFalse ( destroy_proxy . call_count ) <EOL> def test_enable_metadata_proxy ( self ) : <EOL> self . _configure_metadata_proxy ( ) <EOL> def test_disable_metadata_proxy_spawn ( self ) : <EOL> self . _configure_metadata_proxy ( enableflag = False ) <EOL> def test_router_id_specified_in_conf ( self ) : <EOL> self . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . _configure_metadata_proxy ( ) <EOL> def _test_process_routers_update_rpc_timeout ( self , ext_net_call = False , <EOL> ext_net_call_failed = False ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> agent . fullsync = False <EOL> agent . _process_router_if_compatible = mock . Mock ( ) <EOL> if ext_net_call_failed : <EOL> agent . _process_router_if_compatible . side_effect = ( <EOL> oslo_messaging . MessagingTimeout ) <EOL> agent . _queue = mock . Mock ( ) <EOL> agent . _resync_router = mock . Mock ( ) <EOL> update = mock . Mock ( ) <EOL> update . router = None <EOL> agent . _queue . each_update_to_next_router . side_effect = [ <EOL> [ ( None , update ) ] ] <EOL> agent . _process_router_update ( ) <EOL> self . assertFalse ( agent . fullsync ) <EOL> self . assertEqual ( ext_net_call , <EOL> agent . _process_router_if_compatible . called ) <EOL> agent . _resync_router . assert_called_with ( update ) <EOL> def test_process_routers_update_rpc_timeout_on_get_routers ( self ) : <EOL> self . plugin_api . get_routers . side_effect = ( <EOL> oslo_messaging . MessagingTimeout ) <EOL> self . _test_process_routers_update_rpc_timeout ( ) <EOL> def test_process_routers_update_resyncs_failed_router ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> agent . _process_router_if_compatible = mock . MagicMock ( ) <EOL> agent . _process_router_if_compatible . side_effect = RuntimeError ( ) <EOL> update = router_processing_queue . RouterUpdate ( <EOL> <NUM_LIT> , <EOL> router_processing_queue . PRIORITY_SYNC_ROUTERS_TASK , <EOL> router = mock . Mock ( ) , <EOL> timestamp = timeutils . utcnow ( ) ) <EOL> agent . _queue . add ( update ) <EOL> agent . _process_router_update ( ) <EOL> self . assertFalse ( agent . plugin_rpc . get_routers . called ) <EOL> agent . _process_router_update ( ) <EOL> self . assertTrue ( agent . plugin_rpc . get_routers . called ) <EOL> def test_process_routers_update_rpc_timeout_on_get_ext_net ( self ) : <EOL> self . _test_process_routers_update_rpc_timeout ( ext_net_call = True , <EOL> ext_net_call_failed = True ) <EOL> def _test_process_routers_update_router_deleted ( self , error = False ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> agent . _queue = mock . Mock ( ) <EOL> update = mock . Mock ( ) <EOL> update . router = None <EOL> update . action = <NUM_LIT:1> <EOL> router_info = mock . MagicMock ( ) <EOL> agent . router_info [ update . id ] = router_info <EOL> router_processor = mock . Mock ( ) <EOL> agent . _queue . each_update_to_next_router . side_effect = [ <EOL> [ ( router_processor , update ) ] ] <EOL> agent . _resync_router = mock . Mock ( ) <EOL> if error : <EOL> agent . _safe_router_removed = mock . Mock ( ) <EOL> agent . _safe_router_removed . return_value = False <EOL> agent . _process_router_update ( ) <EOL> if error : <EOL> self . assertFalse ( router_processor . fetched_and_processed . called ) <EOL> agent . _resync_router . assert_called_with ( update ) <EOL> else : <EOL> router_info . delete . assert_called_once_with ( agent ) <EOL> self . assertFalse ( agent . router_info ) <EOL> self . assertFalse ( agent . _resync_router . called ) <EOL> router_processor . fetched_and_processed . assert_called_once_with ( <EOL> update . timestamp ) <EOL> def test_process_routers_update_router_deleted_success ( self ) : <EOL> self . _test_process_routers_update_router_deleted ( ) <EOL> def test_process_routers_update_router_deleted_error ( self ) : <EOL> self . _test_process_routers_update_router_deleted ( True ) <EOL> def test_process_router_if_compatible_with_no_ext_net_in_conf ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . plugin_api . get_external_network_id . return_value = '<STR_LIT>' <EOL> router = { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> agent . _process_router_if_compatible ( router ) <EOL> self . assertIn ( router [ '<STR_LIT:id>' ] , agent . router_info ) <EOL> self . plugin_api . get_external_network_id . assert_called_with ( <EOL> agent . context ) <EOL> def test_process_router_if_compatible_with_cached_ext_net ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . plugin_api . get_external_network_id . return_value = '<STR_LIT>' <EOL> agent . target_ex_net_id = '<STR_LIT>' <EOL> router = { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> agent . _process_router_if_compatible ( router ) <EOL> self . assertIn ( router [ '<STR_LIT:id>' ] , agent . router_info ) <EOL> self . assertFalse ( self . plugin_api . get_external_network_id . called ) <EOL> def test_process_router_if_compatible_with_stale_cached_ext_net ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . plugin_api . get_external_network_id . return_value = '<STR_LIT>' <EOL> agent . target_ex_net_id = '<STR_LIT>' <EOL> router = { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> agent . _process_router_if_compatible ( router ) <EOL> self . assertIn ( router [ '<STR_LIT:id>' ] , agent . router_info ) <EOL> self . plugin_api . get_external_network_id . assert_called_with ( <EOL> agent . context ) <EOL> def test_process_router_if_compatible_w_no_ext_net_and_2_net_plugin ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> agent . router_info = { } <EOL> self . plugin_api . get_external_network_id . side_effect = ( <EOL> exc . TooManyExternalNetworks ( ) ) <EOL> self . assertRaises ( exc . TooManyExternalNetworks , <EOL> agent . _process_router_if_compatible , <EOL> router ) <EOL> self . assertNotIn ( router [ '<STR_LIT:id>' ] , agent . router_info ) <EOL> def test_process_router_if_compatible_with_ext_net_in_conf ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . plugin_api . get_external_network_id . return_value = '<STR_LIT>' <EOL> router = { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> agent . router_info = { } <EOL> self . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertRaises ( n_exc . RouterNotCompatibleWithAgent , <EOL> agent . _process_router_if_compatible , <EOL> router ) <EOL> self . assertNotIn ( router [ '<STR_LIT:id>' ] , agent . router_info ) <EOL> def test_process_router_if_compatible_with_no_bridge_no_ext_net ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . plugin_api . get_external_network_id . return_value = '<STR_LIT>' <EOL> router = { '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } } <EOL> agent . router_info = { } <EOL> self . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> agent . _process_router_if_compatible ( router ) <EOL> self . assertIn ( router [ '<STR_LIT:id>' ] , agent . router_info ) <EOL> def test_nonexistent_interface_driver ( self ) : <EOL> self . conf . set_override ( '<STR_LIT>' , None ) <EOL> self . assertRaises ( SystemExit , l3_agent . L3NATAgent , <EOL> HOSTNAME , self . conf ) <EOL> self . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertRaises ( SystemExit , l3_agent . L3NATAgent , <EOL> HOSTNAME , self . conf ) <EOL> @ mock . patch . object ( namespaces . RouterNamespace , '<STR_LIT>' ) <EOL> @ mock . patch . object ( dvr_snat_ns . SnatNamespace , '<STR_LIT>' ) <EOL> def _cleanup_namespace_test ( self , <EOL> stale_namespace_list , <EOL> router_list , <EOL> other_namespaces , <EOL> mock_snat_ns , <EOL> mock_router_ns ) : <EOL> good_namespace_list = [ namespaces . NS_PREFIX + r [ '<STR_LIT:id>' ] <EOL> for r in router_list ] <EOL> good_namespace_list += [ dvr_snat_ns . SNAT_NS_PREFIX + r [ '<STR_LIT:id>' ] <EOL> for r in router_list ] <EOL> self . mock_ip . get_namespaces . return_value = ( stale_namespace_list + <EOL> good_namespace_list + <EOL> other_namespaces ) <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . assertTrue ( agent . namespaces_manager . _clean_stale ) <EOL> pm = self . external_process . return_value <EOL> pm . reset_mock ( ) <EOL> with agent . namespaces_manager as ns_manager : <EOL> for r in router_list : <EOL> ns_manager . keep_router ( r [ '<STR_LIT:id>' ] ) <EOL> qrouters = [ n for n in stale_namespace_list <EOL> if n . startswith ( namespaces . NS_PREFIX ) ] <EOL> self . assertEqual ( len ( qrouters ) , mock_router_ns . call_count ) <EOL> self . assertEqual ( <EOL> len ( stale_namespace_list ) - len ( qrouters ) , <EOL> mock_snat_ns . call_count ) <EOL> self . assertFalse ( agent . namespaces_manager . _clean_stale ) <EOL> def test_cleanup_namespace ( self ) : <EOL> self . conf . set_override ( '<STR_LIT>' , None ) <EOL> stale_namespaces = [ namespaces . NS_PREFIX + '<STR_LIT:foo>' , <EOL> namespaces . NS_PREFIX + '<STR_LIT:bar>' , <EOL> dvr_snat_ns . SNAT_NS_PREFIX + '<STR_LIT:foo>' ] <EOL> other_namespaces = [ '<STR_LIT>' ] <EOL> self . _cleanup_namespace_test ( stale_namespaces , <EOL> [ ] , <EOL> other_namespaces ) <EOL> def test_cleanup_namespace_with_registered_router_ids ( self ) : <EOL> self . conf . set_override ( '<STR_LIT>' , None ) <EOL> stale_namespaces = [ namespaces . NS_PREFIX + '<STR_LIT>' , <EOL> namespaces . NS_PREFIX + '<STR_LIT>' , <EOL> dvr_snat_ns . SNAT_NS_PREFIX + '<STR_LIT>' ] <EOL> router_list = [ { '<STR_LIT:id>' : '<STR_LIT:foo>' , '<STR_LIT>' : False } , <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , '<STR_LIT>' : False } ] <EOL> other_namespaces = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . _cleanup_namespace_test ( stale_namespaces , <EOL> router_list , <EOL> other_namespaces ) <EOL> def test_cleanup_namespace_with_conf_router_id ( self ) : <EOL> self . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> stale_namespaces = [ namespaces . NS_PREFIX + '<STR_LIT>' , <EOL> namespaces . NS_PREFIX + '<STR_LIT>' , <EOL> namespaces . NS_PREFIX + self . conf . router_id ] <EOL> router_list = [ { '<STR_LIT:id>' : '<STR_LIT:foo>' , '<STR_LIT>' : False } , <EOL> { '<STR_LIT:id>' : '<STR_LIT>' , '<STR_LIT>' : False } ] <EOL> other_namespaces = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> self . _cleanup_namespace_test ( stale_namespaces , <EOL> router_list , <EOL> other_namespaces ) <EOL> def test_create_dvr_gateway ( self ) : <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = dvr_router . DvrEdgeRouter ( agent , <EOL> HOSTNAME , <EOL> router [ '<STR_LIT:id>' ] , <EOL> router , <EOL> ** self . ri_kwargs ) <EOL> port_id = _uuid ( ) <EOL> subnet_id = _uuid ( ) <EOL> dvr_gw_port = { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : subnet_id } ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT:id>' : subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT:id>' : port_id , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> interface_name = ri . _get_snat_int_device_name ( port_id ) <EOL> self . device_exists . return_value = False <EOL> with mock . patch . object ( ri , '<STR_LIT>' ) as get_interfaces : <EOL> get_interfaces . return_value = self . snat_ports <EOL> ri . _create_dvr_gateway ( dvr_gw_port , interface_name ) <EOL> self . assertEqual ( <NUM_LIT:3> , self . mock_driver . plug . call_count ) <EOL> self . assertEqual ( <NUM_LIT:3> , self . mock_driver . init_router_port . call_count ) <EOL> def test_process_address_scope ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> router [ '<STR_LIT>' ] = True <EOL> router [ '<STR_LIT>' ] = HOSTNAME <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> ri = dvr_router . DvrEdgeRouter ( agent , <EOL> HOSTNAME , <EOL> router [ '<STR_LIT:id>' ] , <EOL> router , <EOL> ** self . ri_kwargs ) <EOL> ri . get_ex_gw_port = mock . Mock ( return_value = None ) <EOL> ri . process_address_scope ( ) <EOL> with mock . patch . object ( ri , '<STR_LIT>' ) as mocked_func : <EOL> ri . snat_iptables_manager = iptables_manager . IptablesManager ( <EOL> namespace = mock . ANY , use_ipv6 = False ) <EOL> ri . snat_iptables_manager . defer_apply_off = mock . Mock ( ) <EOL> ri . process_address_scope ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , mocked_func . call_count ) <EOL> def test_get_service_plugin_list ( self ) : <EOL> service_plugins = [ p_const . L3_ROUTER_NAT ] <EOL> self . plugin_api . get_service_plugin_list . return_value = service_plugins <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . assertEqual ( service_plugins , agent . neutron_service_plugins ) <EOL> self . assertTrue ( self . plugin_api . get_service_plugin_list . called ) <EOL> def test_get_service_plugin_list_failed ( self ) : <EOL> raise_rpc = oslo_messaging . RemoteError ( ) <EOL> self . plugin_api . get_service_plugin_list . side_effect = raise_rpc <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . assertIsNone ( agent . neutron_service_plugins ) <EOL> self . assertTrue ( self . plugin_api . get_service_plugin_list . called ) <EOL> def test_get_service_plugin_list_retried ( self ) : <EOL> raise_timeout = oslo_messaging . MessagingTimeout ( ) <EOL> self . plugin_api . get_service_plugin_list . side_effect = ( <EOL> raise_timeout , raise_timeout , tuple ( ) <EOL> ) <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . assertEqual ( tuple ( ) , agent . neutron_service_plugins ) <EOL> def test_get_service_plugin_list_retried_max ( self ) : <EOL> raise_timeout = oslo_messaging . MessagingTimeout ( ) <EOL> self . plugin_api . get_service_plugin_list . side_effect = ( <EOL> ( raise_timeout , ) * <NUM_LIT:5> <EOL> ) <EOL> self . assertRaises ( oslo_messaging . MessagingTimeout , l3_agent . L3NATAgent , <EOL> HOSTNAME , self . conf ) <EOL> def test_external_gateway_removed_ext_gw_port_no_fip_ns ( self ) : <EOL> self . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> agent . conf . agent_mode = '<STR_LIT>' <EOL> router = l3_test_common . prepare_router_data ( num_internal_ports = <NUM_LIT:2> ) <EOL> router [ '<STR_LIT>' ] = HOSTNAME <EOL> self . mock_driver . unplug . reset_mock ( ) <EOL> external_net_id = router [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> ri = dvr_router . DvrEdgeRouter ( <EOL> agent , HOSTNAME , router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> ri . remove_floating_ip = mock . Mock ( ) <EOL> agent . _fetch_external_net_id = mock . Mock ( return_value = external_net_id ) <EOL> ri . ex_gw_port = ri . router [ '<STR_LIT>' ] <EOL> del ri . router [ '<STR_LIT>' ] <EOL> ri . fip_ns = None <EOL> nat = ri . iptables_manager . ipv4 [ '<STR_LIT>' ] <EOL> nat . clear_rules_by_tag = mock . Mock ( ) <EOL> nat . add_rule = mock . Mock ( ) <EOL> ri . snat_namespace = mock . Mock ( ) <EOL> ri . external_gateway_removed ( <EOL> ri . ex_gw_port , <EOL> ri . get_external_device_name ( ri . ex_gw_port [ '<STR_LIT:id>' ] ) ) <EOL> self . assertFalse ( ri . remove_floating_ip . called ) <EOL> def test_spawn_radvd ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ip_version = <NUM_LIT:6> ) <EOL> conffile = '<STR_LIT>' <EOL> pidfile = '<STR_LIT>' <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> self . external_process_p . stop ( ) <EOL> self . ip_cls_p . stop ( ) <EOL> get_conf_file_name = '<STR_LIT>' <EOL> get_pid_file_name = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> utils_execute = '<STR_LIT>' <EOL> mock . patch ( get_conf_file_name ) . start ( ) . return_value = conffile <EOL> mock . patch ( get_pid_file_name ) . start ( ) . return_value = pidfile <EOL> execute = mock . patch ( utils_execute ) . start ( ) <EOL> radvd = ra . DaemonMonitor ( <EOL> router [ '<STR_LIT:id>' ] , <EOL> namespaces . RouterNamespace . _get_ns_name ( router [ '<STR_LIT:id>' ] ) , <EOL> agent . process_monitor , <EOL> l3_test_common . FakeDev , <EOL> self . conf ) <EOL> radvd . enable ( router [ '<STR_LIT>' ] ) <EOL> cmd = execute . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> self . assertIn ( '<STR_LIT>' , cmd ) <EOL> _join = lambda * args : '<STR_LIT:U+0020>' . join ( args ) <EOL> cmd = _join ( * cmd ) <EOL> self . assertIn ( _join ( '<STR_LIT>' , conffile ) , cmd ) <EOL> self . assertIn ( _join ( '<STR_LIT>' , pidfile ) , cmd ) <EOL> self . assertIn ( _join ( '<STR_LIT>' , '<STR_LIT>' ) , cmd ) <EOL> def test_generate_radvd_mtu_conf ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ipv6_subnet_modes = [ { '<STR_LIT>' : n_const . IPV6_SLAAC , <EOL> '<STR_LIT>' : n_const . IPV6_SLAAC } ] <EOL> network_mtu = '<STR_LIT>' <EOL> ri = self . _process_router_ipv6_subnet_added ( router , <EOL> ipv6_subnet_modes , <EOL> None , <EOL> network_mtu ) <EOL> expected = "<STR_LIT>" <EOL> ri . agent_conf . set_override ( '<STR_LIT>' , False ) <EOL> ri . radvd . _generate_radvd_conf ( router [ l3_constants . INTERFACE_KEY ] ) <EOL> self . assertNotIn ( expected , self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) <EOL> ri . agent_conf . set_override ( '<STR_LIT>' , True ) <EOL> ri . radvd . _generate_radvd_conf ( router [ l3_constants . INTERFACE_KEY ] ) <EOL> self . assertIn ( expected , self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) <EOL> def test_generate_radvd_conf_other_and_managed_flag ( self ) : <EOL> expected = { n_const . IPV6_SLAAC : ( False , False ) , <EOL> n_const . DHCPV6_STATELESS : ( True , False ) , <EOL> n_const . DHCPV6_STATEFUL : ( False , True ) } <EOL> modes = [ n_const . IPV6_SLAAC , n_const . DHCPV6_STATELESS , <EOL> n_const . DHCPV6_STATEFUL ] <EOL> mode_combos = list ( iter_chain ( * [ [ list ( combo ) for combo in <EOL> iter_combinations ( modes , i ) ] for i in range ( <NUM_LIT:1> , len ( modes ) + <NUM_LIT:1> ) ] ) ) <EOL> for mode_list in mode_combos : <EOL> ipv6_subnet_modes = [ { '<STR_LIT>' : mode , '<STR_LIT>' : mode } <EOL> for mode in mode_list ] <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = self . _process_router_ipv6_subnet_added ( router , <EOL> ipv6_subnet_modes ) <EOL> ri . radvd . _generate_radvd_conf ( router [ l3_constants . INTERFACE_KEY ] ) <EOL> def assertFlag ( flag ) : <EOL> return ( self . assertIn if flag else self . assertNotIn ) <EOL> other_flag , managed_flag = ( <EOL> any ( expected [ mode ] [ <NUM_LIT:0> ] for mode in mode_list ) , <EOL> any ( expected [ mode ] [ <NUM_LIT:1> ] for mode in mode_list ) ) <EOL> assertFlag ( other_flag ) ( '<STR_LIT>' , <EOL> self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) <EOL> assertFlag ( managed_flag ) ( '<STR_LIT>' , <EOL> self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) <EOL> def test_generate_radvd_intervals ( self ) : <EOL> self . conf . set_override ( '<STR_LIT>' , <NUM_LIT> ) <EOL> self . conf . set_override ( '<STR_LIT>' , <NUM_LIT> ) <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ipv6_subnet_modes = [ { '<STR_LIT>' : n_const . IPV6_SLAAC , <EOL> '<STR_LIT>' : n_const . IPV6_SLAAC } ] <EOL> ri = self . _process_router_ipv6_subnet_added ( router , <EOL> ipv6_subnet_modes ) <EOL> ri . radvd . _generate_radvd_conf ( router [ l3_constants . INTERFACE_KEY ] ) <EOL> self . assertIn ( "<STR_LIT>" , <EOL> self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) <EOL> self . assertIn ( "<STR_LIT>" , <EOL> self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) <EOL> def test_generate_radvd_rdnss_conf ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ipv6_subnet_modes = [ { '<STR_LIT>' : n_const . IPV6_SLAAC , <EOL> '<STR_LIT>' : n_const . IPV6_SLAAC } ] <EOL> dns_list = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ri = self . _process_router_ipv6_subnet_added ( router , <EOL> ipv6_subnet_modes , <EOL> dns_nameservers = dns_list ) <EOL> ri . radvd . _generate_radvd_conf ( router [ l3_constants . INTERFACE_KEY ] ) <EOL> expected = "<STR_LIT>" <EOL> for dns in dns_list [ <NUM_LIT:0> : ra . MAX_RDNSS_ENTRIES ] : <EOL> expected += "<STR_LIT>" % dns <EOL> self . assertIn ( expected , self . utils_replace_file . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] ) <EOL> def _pd_expected_call_external_process ( self , requestor , ri , enable = True ) : <EOL> expected_calls = [ ] <EOL> if enable : <EOL> expected_calls . append ( mock . call ( uuid = requestor , <EOL> service = '<STR_LIT>' , <EOL> default_cmd_callback = mock . ANY , <EOL> namespace = ri . ns_name , <EOL> conf = mock . ANY , <EOL> pid_file = mock . ANY ) ) <EOL> expected_calls . append ( mock . call ( ) . enable ( reload_cfg = False ) ) <EOL> else : <EOL> expected_calls . append ( mock . call ( uuid = requestor , <EOL> service = '<STR_LIT>' , <EOL> namespace = ri . ns_name , <EOL> conf = mock . ANY , <EOL> pid_file = mock . ANY ) ) <EOL> expected_calls . append ( mock . call ( ) . disable ( <EOL> get_stop_command = mock . ANY ) ) <EOL> return expected_calls <EOL> def _pd_setup_agent_router ( self ) : <EOL> router = l3_test_common . prepare_router_data ( ) <EOL> ri = l3router . RouterInfo ( router [ '<STR_LIT:id>' ] , router , ** self . ri_kwargs ) <EOL> agent = l3_agent . L3NATAgent ( HOSTNAME , self . conf ) <EOL> agent . external_gateway_added = mock . Mock ( ) <EOL> ri . process ( agent ) <EOL> agent . _router_added ( router [ '<STR_LIT:id>' ] , router ) <EOL> if not ri . radvd : <EOL> ri . radvd = ra . DaemonMonitor ( router [ '<STR_LIT:id>' ] , <EOL> ri . ns_name , <EOL> agent . process_monitor , <EOL> ri . get_internal_device_name , <EOL> self . conf ) <EOL> return agent , router , ri <EOL> def _pd_remove_gw_interface ( self , intfs , agent , router , ri ) : <EOL> expected_pd_update = { } <EOL> expected_calls = [ ] <EOL> for intf in intfs : <EOL> requestor_id = self . _pd_get_requestor_id ( intf , router , ri ) <EOL> expected_calls += ( self . _pd_expected_call_external_process ( <EOL> requestor_id , ri , False ) ) <EOL> for subnet in intf [ '<STR_LIT>' ] : <EOL> expected_pd_update [ subnet [ '<STR_LIT:id>' ] ] = ( <EOL> n_const . PROVISIONAL_IPV6_PD_PREFIX ) <EOL> self . pd_update = { } <EOL> def pd_notifier ( context , prefix_update ) : <EOL> self . pd_update = prefix_update <EOL> for subnet_id , prefix in six . iteritems ( prefix_update ) : <EOL> for intf in intfs : <EOL> for subnet in intf [ '<STR_LIT>' ] : <EOL> if subnet [ '<STR_LIT:id>' ] == subnet_id : <EOL> subnet [ '<STR_LIT>' ] = prefix <EOL> break <EOL> agent . pd . notifier = pd_notifier <EOL> agent . pd . remove_gw_interface ( router [ '<STR_LIT:id>' ] ) <EOL> self . _pd_assert_dibbler_calls ( expected_calls , <EOL> self . external_process . mock_calls [ - len ( expected_calls ) : ] ) <EOL> self . assertEqual ( expected_pd_update , self . pd_update ) <EOL> def _pd_remove_interfaces ( self , intfs , agent , router , ri ) : <EOL> expected_pd_update = [ ] <EOL> expected_calls = [ ] <EOL> for intf in intfs : <EOL> router [ l3_constants . INTERFACE_KEY ] . remove ( intf ) <EOL> requestor_id = self . _pd_get_requestor_id ( intf , router , ri ) <EOL> expected_calls += ( self . _pd_expected_call_external_process ( <EOL> requestor_id , ri , False ) ) <EOL> for subnet in intf [ '<STR_LIT>' ] : <EOL> expected_pd_update += [ { subnet [ '<STR_LIT:id>' ] : <EOL> n_const . PROVISIONAL_IPV6_PD_PREFIX } ] <EOL> self . pd_update = [ ] <EOL> def pd_notifier ( context , prefix_update ) : <EOL> self . pd_update . append ( prefix_update ) <EOL> for intf in intfs : <EOL> for subnet in intf [ '<STR_LIT>' ] : <EOL> if subnet [ '<STR_LIT:id>' ] in prefix_update : <EOL> subnet [ '<STR_LIT>' ] = prefix_update [ subnet [ '<STR_LIT:id>' ] ] <EOL> agent . pd . notifier = pd_notifier <EOL> ri . process ( agent ) <EOL> self . _pd_assert_dibbler_calls ( expected_calls , <EOL> self . external_process . mock_calls [ - len ( expected_calls ) - <NUM_LIT:2> : ] ) <EOL> self . _pd_assert_radvd_calls ( ri , False ) <EOL> self . assertEqual ( expected_pd_update , self . pd_update ) <EOL> def _pd_get_requestor_id ( self , intf , router , ri ) : <EOL> ifname = ri . get_internal_device_name ( intf [ '<STR_LIT:id>' ] ) <EOL> for subnet in intf [ '<STR_LIT>' ] : <EOL> return dibbler . PDDibbler ( router [ '<STR_LIT:id>' ] , <EOL> subnet [ '<STR_LIT:id>' ] , ifname ) . requestor_id <EOL> def _pd_assert_dibbler_calls ( self , expected , actual ) : <EOL> '''<STR_LIT>''' <EOL> num_ext_calls = len ( expected ) // <NUM_LIT:2> <EOL> expected_ext_calls = [ ] <EOL> actual_ext_calls = [ ] <EOL> expected_action_calls = [ ] <EOL> actual_action_calls = [ ] <EOL> for c in range ( num_ext_calls ) : <EOL> expected_ext_calls . append ( expected [ c * <NUM_LIT:2> ] ) <EOL> actual_ext_calls . append ( actual [ c * <NUM_LIT:2> ] ) <EOL> expected_action_calls . append ( expected [ c * <NUM_LIT:2> + <NUM_LIT:1> ] ) <EOL> actual_action_calls . append ( actual [ c * <NUM_LIT:2> + <NUM_LIT:1> ] ) <EOL> self . assertEqual ( expected_action_calls , actual_action_calls ) <EOL> for exp in expected_ext_calls : <EOL> for act in actual_ext_calls : <EOL> if exp == act : <EOL> break <EOL> else : <EOL> msg = "<STR_LIT>" <EOL> self . fail ( msg ) <EOL> def _pd_assert_radvd_calls ( self , ri , enable = True ) : <EOL> exp_calls = self . _radvd_expected_call_external_process ( ri , enable ) <EOL> self . assertEqual ( exp_calls , <EOL> self . external_process . mock_calls [ - len ( exp_calls ) : ] ) <EOL> def _pd_get_prefixes ( self , agent , router , ri , <EOL> existing_intfs , new_intfs , mock_get_prefix ) : <EOL> prefixes = { } <EOL> expected_pd_update = { } <EOL> expected_calls = [ ] <EOL> for ifno , intf in enumerate ( existing_intfs + new_intfs ) : <EOL> requestor_id = self . _pd_get_requestor_id ( intf , router , ri ) <EOL> prefixes [ requestor_id ] = "<STR_LIT>" % ifno <EOL> if intf in new_intfs : <EOL> subnet_id = ( intf [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] if intf [ '<STR_LIT>' ] <EOL> else None ) <EOL> expected_pd_update [ subnet_id ] = prefixes [ requestor_id ] <EOL> expected_calls += ( <EOL> self . _pd_expected_call_external_process ( requestor_id , ri ) ) <EOL> self . pd_update = { } <EOL> def pd_notifier ( context , prefix_update ) : <EOL> self . pd_update = prefix_update <EOL> for subnet_id , prefix in six . iteritems ( prefix_update ) : <EOL> for intf in new_intfs : <EOL> for subnet in intf [ '<STR_LIT>' ] : <EOL> if subnet [ '<STR_LIT:id>' ] == subnet_id : <EOL> subnet [ '<STR_LIT>' ] = prefix <EOL> break <EOL> agent . pd . notifier = pd_notifier <EOL> agent . pd . process_prefix_update ( ) <EOL> def get_prefix ( pdo ) : <EOL> key = '<STR_LIT>' % ( pdo . router_id , pdo . subnet_id , pdo . ri_ifname ) <EOL> return prefixes [ key ] <EOL> mock_get_prefix . side_effect = get_prefix <EOL> agent . pd . process_prefix_update ( ) <EOL> self . _pd_assert_dibbler_calls ( expected_calls , <EOL> self . external_process . mock_calls [ - len ( expected_calls ) : ] ) <EOL> self . assertEqual ( expected_pd_update , self . pd_update ) <EOL> def _pd_add_gw_interface ( self , agent , router , ri ) : <EOL> gw_ifname = ri . get_external_device_name ( router [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] ) <EOL> agent . pd . add_gw_interface ( router [ '<STR_LIT:id>' ] , gw_ifname ) <EOL> @ mock . patch . object ( dibbler . PDDibbler , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( dibbler . os , '<STR_LIT>' , return_value = <NUM_LIT> ) <EOL> @ mock . patch . object ( pd . PrefixDelegation , '<STR_LIT>' , <EOL> return_value = True ) <EOL> @ mock . patch . object ( dibbler . os , '<STR_LIT>' ) <EOL> @ mock . patch . object ( dibbler . shutil , '<STR_LIT>' ) <EOL> @ mock . patch . object ( pd . PrefixDelegation , '<STR_LIT>' ) <EOL> def test_pd_add_remove_subnet ( self , mock1 , mock2 , mock3 , mock4 , <EOL> mock_getpid , mock_get_prefix ) : <EOL> '''<STR_LIT>''' <EOL> agent , router , ri = self . _pd_setup_agent_router ( ) <EOL> intfs = l3_test_common . router_append_pd_enabled_subnet ( router ) <EOL> ri . process ( agent ) <EOL> self . assertFalse ( self . external_process . call_count ) <EOL> self . assertFalse ( mock_get_prefix . call_count ) <EOL> self . _pd_add_gw_interface ( agent , router , ri ) <EOL> self . _pd_get_prefixes ( agent , router , ri , [ ] , intfs , mock_get_prefix ) <EOL> ri . process ( agent ) <EOL> self . _pd_assert_radvd_calls ( ri ) <EOL> self . _pd_remove_interfaces ( intfs , agent , router , ri ) <EOL> @ mock . patch . object ( dibbler . PDDibbler , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( dibbler . os , '<STR_LIT>' , return_value = <NUM_LIT> ) <EOL> @ mock . patch . object ( pd . PrefixDelegation , '<STR_LIT>' , <EOL> return_value = True ) <EOL> @ mock . patch . object ( dibbler . os , '<STR_LIT>' ) <EOL> @ mock . patch . object ( dibbler . shutil , '<STR_LIT>' ) <EOL> @ mock . patch . object ( pd . PrefixDelegation , '<STR_LIT>' ) <EOL> def test_pd_remove_gateway ( self , mock1 , mock2 , mock3 , mock4 , <EOL> mock_getpid , mock_get_prefix ) : <EOL> '''<STR_LIT>''' <EOL> agent , router , ri = self . _pd_setup_agent_router ( ) <EOL> intfs = l3_test_common . router_append_pd_enabled_subnet ( router ) <EOL> ri . process ( agent ) <EOL> self . _pd_add_gw_interface ( agent , router , ri ) <EOL> self . _pd_get_prefixes ( agent , router , ri , [ ] , intfs , mock_get_prefix ) <EOL> ri . process ( agent ) <EOL> self . _pd_assert_radvd_calls ( ri ) <EOL> self . _pd_remove_gw_interface ( intfs , agent , router , ri ) <EOL> ri . process ( agent ) <EOL> @ mock . patch . object ( dibbler . PDDibbler , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( dibbler . os , '<STR_LIT>' , return_value = <NUM_LIT> ) <EOL> @ mock . patch . object ( pd . PrefixDelegation , '<STR_LIT>' , <EOL> return_value = True ) <EOL> @ mock . patch . object ( dibbler . os , '<STR_LIT>' ) <EOL> @ mock . patch . object ( dibbler . shutil , '<STR_LIT>' ) <EOL> @ mock . patch . object ( pd . PrefixDelegation , '<STR_LIT>' ) <EOL> def test_pd_add_remove_2_subnets ( self , mock1 , mock2 , mock3 , mock4 , <EOL> mock_getpid , mock_get_prefix ) : <EOL> '''<STR_LIT>''' <EOL> agent , router , ri = self . _pd_setup_agent_router ( ) <EOL> intfs = l3_test_common . router_append_pd_enabled_subnet ( router , count = <NUM_LIT:2> ) <EOL> ri . process ( agent ) <EOL> self . assertFalse ( self . external_process . call_count ) <EOL> self . assertFalse ( mock_get_prefix . call_count ) <EOL> self . _pd_add_gw_interface ( agent , router , ri ) <EOL> self . _pd_get_prefixes ( agent , router , ri , [ ] , intfs , mock_get_prefix ) <EOL> ri . process ( agent ) <EOL> self . _pd_assert_radvd_calls ( ri ) <EOL> self . _pd_remove_interfaces ( intfs , agent , router , ri ) <EOL> @ mock . patch . object ( dibbler . PDDibbler , '<STR_LIT>' , autospec = True ) <EOL> @ mock . patch . object ( dibbler . os , '<STR_LIT>' , return_value = <NUM_LIT> ) <EOL> @ mock . patch . object ( pd . PrefixDelegation , '<STR_LIT>' , <EOL> return_value = True ) <EOL> @ mock . patch . object ( dibbler . os , '<STR_LIT>' ) <EOL> @ mock . patch . object ( dibbler . shutil , '<STR_LIT>' ) <EOL> @ mock . patch . object ( pd . PrefixDelegation , '<STR_LIT>' ) <EOL> def test_pd_remove_gateway_2_subnets ( self , mock1 , mock2 , mock3 , mock4 , <EOL> mock_getpid , mock_get_prefix ) : <EOL> '''<STR_LIT>''' <EOL> agent , router , ri = self . _pd_setup_agent_router ( ) <EOL> self . _pd_add_gw_interface ( agent , router , ri ) <EOL> intfs = l3_test_common . router_append_pd_enabled_subnet ( router , count = <NUM_LIT:1> ) <EOL> ri . process ( agent ) <EOL> self . _pd_get_prefixes ( agent , router , ri , [ ] , intfs , mock_get_prefix ) <EOL> ri . process ( agent ) <EOL> self . _pd_assert_radvd_calls ( ri ) <EOL> intfs1 = l3_test_common . router_append_pd_enabled_subnet ( router , <EOL> count = <NUM_LIT:1> ) <EOL> ri . process ( agent ) <EOL> self . _pd_get_prefixes ( agent , router , ri , intfs , <EOL> intfs1 , mock_get_prefix ) <EOL> ri . process ( agent ) <EOL> self . _pd_assert_radvd_calls ( ri ) <EOL> self . _pd_remove_gw_interface ( intfs + intfs1 , agent , router , ri ) <EOL> ri . process ( agent ) </s>
