<s> import copy <EOL> import mock <EOL> from oslo_utils import uuidutils <EOL> from neutron . agent . common import utils <EOL> from neutron . agent . l3 import dvr_fip_ns <EOL> from neutron . agent . l3 import link_local_allocator as lla <EOL> from neutron . agent . linux import ip_lib <EOL> from neutron . agent . linux import iptables_manager <EOL> from neutron . tests import base <EOL> _uuid = uuidutils . generate_uuid <EOL> class TestDvrFipNs ( base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestDvrFipNs , self ) . setUp ( ) <EOL> self . conf = mock . Mock ( ) <EOL> self . conf . state_path = '<STR_LIT>' <EOL> self . driver = mock . Mock ( ) <EOL> self . driver . DEV_NAME_LEN = <NUM_LIT> <EOL> self . net_id = _uuid ( ) <EOL> self . fip_ns = dvr_fip_ns . FipNamespace ( self . net_id , <EOL> self . conf , <EOL> self . driver , <EOL> use_ipv6 = True ) <EOL> def test_subscribe ( self ) : <EOL> is_first = self . fip_ns . subscribe ( mock . sentinel . external_net_id ) <EOL> self . assertTrue ( is_first ) <EOL> def test_subscribe_not_first ( self ) : <EOL> self . fip_ns . subscribe ( mock . sentinel . external_net_id ) <EOL> is_first = self . fip_ns . subscribe ( mock . sentinel . external_net_id2 ) <EOL> self . assertFalse ( is_first ) <EOL> def test_unsubscribe ( self ) : <EOL> self . fip_ns . subscribe ( mock . sentinel . external_net_id ) <EOL> is_last = self . fip_ns . unsubscribe ( mock . sentinel . external_net_id ) <EOL> self . assertTrue ( is_last ) <EOL> def test_unsubscribe_not_last ( self ) : <EOL> self . fip_ns . subscribe ( mock . sentinel . external_net_id ) <EOL> self . fip_ns . subscribe ( mock . sentinel . external_net_id2 ) <EOL> is_last = self . fip_ns . unsubscribe ( mock . sentinel . external_net_id2 ) <EOL> self . assertFalse ( is_last ) <EOL> def test_allocate_rule_priority ( self ) : <EOL> pr = self . fip_ns . allocate_rule_priority ( '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , self . fip_ns . _rule_priorities . allocations ) <EOL> self . assertNotIn ( pr , self . fip_ns . _rule_priorities . pool ) <EOL> def test_deallocate_rule_priority ( self ) : <EOL> pr = self . fip_ns . allocate_rule_priority ( '<STR_LIT>' ) <EOL> self . fip_ns . deallocate_rule_priority ( '<STR_LIT>' ) <EOL> self . assertNotIn ( '<STR_LIT>' , self . fip_ns . _rule_priorities . allocations ) <EOL> self . assertIn ( pr , self . fip_ns . _rule_priorities . pool ) <EOL> def _get_agent_gw_port ( self ) : <EOL> v4_subnet_id = _uuid ( ) <EOL> v6_subnet_id = _uuid ( ) <EOL> agent_gw_port = { '<STR_LIT>' : [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : v4_subnet_id } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:64> , <EOL> '<STR_LIT>' : v6_subnet_id } ] , <EOL> '<STR_LIT>' : [ { '<STR_LIT:id>' : v4_subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : v6_subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : self . net_id , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> return agent_gw_port <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> def test_gateway_added ( self , device_exists , ip_wrapper ) : <EOL> agent_gw_port = self . _get_agent_gw_port ( ) <EOL> device_exists . return_value = False <EOL> self . fip_ns . update_gateway_port = mock . Mock ( ) <EOL> self . fip_ns . _gateway_added ( agent_gw_port , <EOL> mock . sentinel . interface_name ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . driver . plug . call_count ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . driver . init_l3 . call_count ) <EOL> self . fip_ns . update_gateway_port . assert_called_once_with ( agent_gw_port ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> def test_update_gateway_port ( self , send_adv_notif , IPDevice ) : <EOL> self . fip_ns . _check_for_gateway_ip_change = mock . Mock ( return_value = True ) <EOL> self . fip_ns . agent_gateway_port = None <EOL> agent_gw_port = self . _get_agent_gw_port ( ) <EOL> self . fip_ns . update_gateway_port ( agent_gw_port ) <EOL> expected = [ <EOL> mock . call ( self . fip_ns . get_name ( ) , <EOL> self . fip_ns . get_ext_device_name ( agent_gw_port [ '<STR_LIT:id>' ] ) , <EOL> agent_gw_port [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> mock . ANY ) , <EOL> mock . call ( self . fip_ns . get_name ( ) , <EOL> self . fip_ns . get_ext_device_name ( agent_gw_port [ '<STR_LIT:id>' ] ) , <EOL> agent_gw_port [ '<STR_LIT>' ] [ <NUM_LIT:1> ] [ '<STR_LIT>' ] , <EOL> mock . ANY ) ] <EOL> send_adv_notif . assert_has_calls ( expected ) <EOL> gw_ipv4 = agent_gw_port [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> gw_ipv6 = agent_gw_port [ '<STR_LIT>' ] [ <NUM_LIT:1> ] [ '<STR_LIT>' ] <EOL> expected = [ mock . call ( gw_ipv4 ) , mock . call ( gw_ipv6 ) ] <EOL> IPDevice ( ) . route . add_gateway . assert_has_calls ( expected ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> def test_update_gateway_port_gateway_outside_subnet_added ( <EOL> self , send_adv_notif , IPDevice ) : <EOL> self . fip_ns . agent_gateway_port = None <EOL> agent_gw_port = self . _get_agent_gw_port ( ) <EOL> agent_gw_port [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . fip_ns . update_gateway_port ( agent_gw_port ) <EOL> IPDevice ( ) . route . add_route . assert_called_once_with ( '<STR_LIT>' , <EOL> scope = '<STR_LIT>' ) <EOL> def test_check_gateway_ip_changed_no_change ( self ) : <EOL> agent_gw_port = self . _get_agent_gw_port ( ) <EOL> self . fip_ns . agent_gateway_port = copy . deepcopy ( agent_gw_port ) <EOL> agent_gw_port [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertFalse ( self . fip_ns . _check_for_gateway_ip_change ( <EOL> agent_gw_port ) ) <EOL> def test_check_gateway_ip_changed_v4 ( self ) : <EOL> agent_gw_port = self . _get_agent_gw_port ( ) <EOL> self . fip_ns . agent_gateway_port = copy . deepcopy ( agent_gw_port ) <EOL> agent_gw_port [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertTrue ( self . fip_ns . _check_for_gateway_ip_change ( <EOL> agent_gw_port ) ) <EOL> def test_check_gateway_ip_changed_v6 ( self ) : <EOL> agent_gw_port = self . _get_agent_gw_port ( ) <EOL> self . fip_ns . agent_gateway_port = copy . deepcopy ( agent_gw_port ) <EOL> agent_gw_port [ '<STR_LIT>' ] [ <NUM_LIT:1> ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assertTrue ( self . fip_ns . _check_for_gateway_ip_change ( <EOL> agent_gw_port ) ) <EOL> @ mock . patch . object ( iptables_manager , '<STR_LIT>' ) <EOL> @ mock . patch . object ( utils , '<STR_LIT>' ) <EOL> @ mock . patch . object ( ip_lib . IpNetnsCommand , '<STR_LIT>' ) <EOL> def _test_create ( self , old_kernel , exists , execute , IPTables ) : <EOL> exists . return_value = True <EOL> execute . side_effect = [ RuntimeError if old_kernel else None , <EOL> None , None , None ] <EOL> self . fip_ns . _iptables_manager = IPTables ( ) <EOL> self . fip_ns . create ( ) <EOL> ns_name = self . fip_ns . get_name ( ) <EOL> netns_cmd = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , ns_name ] <EOL> bind_cmd = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> expected = [ mock . call ( netns_cmd + bind_cmd , check_exit_code = True , <EOL> extra_ok_codes = None , log_fail_as_error = False , <EOL> run_as_root = True ) ] <EOL> if old_kernel : <EOL> expected . append ( mock . call ( bind_cmd , check_exit_code = True , <EOL> extra_ok_codes = None , <EOL> log_fail_as_error = True , <EOL> run_as_root = True ) ) <EOL> execute . assert_has_calls ( expected ) <EOL> def test_create_old_kernel ( self ) : <EOL> self . _test_create ( True ) <EOL> def test_create_new_kernel ( self ) : <EOL> self . _test_create ( False ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> def test_destroy ( self , IPWrapper ) : <EOL> ip_wrapper = IPWrapper ( ) <EOL> dev1 = mock . Mock ( ) <EOL> dev1 . name = '<STR_LIT>' <EOL> dev2 = mock . Mock ( ) <EOL> dev2 . name = '<STR_LIT>' <EOL> ip_wrapper . get_devices . return_value = [ dev1 , dev2 ] <EOL> with mock . patch . object ( self . fip_ns . ip_wrapper_root . netns , <EOL> '<STR_LIT>' ) as delete : <EOL> self . fip_ns . delete ( ) <EOL> delete . assert_called_once_with ( mock . ANY ) <EOL> ext_net_bridge = self . conf . external_network_bridge <EOL> ns_name = self . fip_ns . get_name ( ) <EOL> self . driver . unplug . assert_called_once_with ( '<STR_LIT>' , <EOL> bridge = ext_net_bridge , <EOL> prefix = '<STR_LIT>' , <EOL> namespace = ns_name ) <EOL> ip_wrapper . del_veth . assert_called_once_with ( '<STR_LIT>' ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> def test_create_rtr_2_fip_link ( self , device_exists , IPDevice , IPWrapper ) : <EOL> ri = mock . Mock ( ) <EOL> ri . router_id = _uuid ( ) <EOL> ri . rtr_fip_subnet = None <EOL> ri . ns_name = mock . sentinel . router_ns <EOL> rtr_2_fip_name = self . fip_ns . get_rtr_ext_device_name ( ri . router_id ) <EOL> fip_2_rtr_name = self . fip_ns . get_int_device_name ( ri . router_id ) <EOL> fip_ns_name = self . fip_ns . get_name ( ) <EOL> self . fip_ns . local_subnets = allocator = mock . Mock ( ) <EOL> pair = lla . LinkLocalAddressPair ( '<STR_LIT>' ) <EOL> allocator . allocate . return_value = pair <EOL> device_exists . return_value = False <EOL> ip_wrapper = IPWrapper ( ) <EOL> self . conf . network_device_mtu = <NUM_LIT> <EOL> ip_wrapper . add_veth . return_value = ( IPDevice ( ) , IPDevice ( ) ) <EOL> self . fip_ns . create_rtr_2_fip_link ( ri ) <EOL> ip_wrapper . add_veth . assert_called_with ( rtr_2_fip_name , <EOL> fip_2_rtr_name , <EOL> fip_ns_name ) <EOL> device = IPDevice ( ) <EOL> device . link . set_mtu . assert_called_with ( <NUM_LIT> ) <EOL> self . assertEqual ( <NUM_LIT:2> , device . link . set_mtu . call_count ) <EOL> device . route . add_gateway . assert_called_once_with ( <EOL> '<STR_LIT>' , table = <NUM_LIT:16> ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> def test_create_rtr_2_fip_link_already_exists ( self , <EOL> device_exists , <EOL> IPDevice , <EOL> IPWrapper ) : <EOL> ri = mock . Mock ( ) <EOL> ri . router_id = _uuid ( ) <EOL> ri . rtr_fip_subnet = None <EOL> device_exists . return_value = True <EOL> self . fip_ns . local_subnets = allocator = mock . Mock ( ) <EOL> pair = lla . LinkLocalAddressPair ( '<STR_LIT>' ) <EOL> allocator . allocate . return_value = pair <EOL> self . fip_ns . create_rtr_2_fip_link ( ri ) <EOL> ip_wrapper = IPWrapper ( ) <EOL> self . assertFalse ( ip_wrapper . add_veth . called ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> def _test_scan_fip_ports ( self , ri , ip_list , IPDevice ) : <EOL> IPDevice . return_value = device = mock . Mock ( ) <EOL> device . addr . list . return_value = ip_list <EOL> self . fip_ns . get_rtr_ext_device_name = mock . Mock ( <EOL> return_value = mock . sentinel . rtr_ext_device_name ) <EOL> self . fip_ns . scan_fip_ports ( ri ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> def test_scan_fip_ports_restart_fips ( self , device_exists ) : <EOL> device_exists . return_value = True <EOL> ri = mock . Mock ( ) <EOL> ri . dist_fip_count = None <EOL> ri . floating_ips_dict = { } <EOL> ip_list = [ { '<STR_LIT>' : '<STR_LIT>' } , { '<STR_LIT>' : '<STR_LIT>' } ] <EOL> self . _test_scan_fip_ports ( ri , ip_list ) <EOL> self . assertEqual ( <NUM_LIT:2> , ri . dist_fip_count ) <EOL> @ mock . patch . object ( ip_lib , '<STR_LIT>' ) <EOL> def test_scan_fip_ports_restart_none ( self , device_exists ) : <EOL> device_exists . return_value = True <EOL> ri = mock . Mock ( ) <EOL> ri . dist_fip_count = None <EOL> ri . floating_ips_dict = { } <EOL> self . _test_scan_fip_ports ( ri , [ ] ) <EOL> self . assertEqual ( <NUM_LIT:0> , ri . dist_fip_count ) <EOL> def test_scan_fip_ports_restart_zero ( self ) : <EOL> ri = mock . Mock ( ) <EOL> ri . dist_fip_count = <NUM_LIT:0> <EOL> self . _test_scan_fip_ports ( ri , None ) <EOL> self . assertEqual ( <NUM_LIT:0> , ri . dist_fip_count ) </s>
