<s> import copy <EOL> import mock <EOL> from neutron_lib import constants as l3_constants <EOL> import six <EOL> from neutron . agent . l3 import agent as neutron_l3_agent <EOL> from neutron . agent . l3 import namespaces <EOL> from neutron . agent . linux import ip_lib <EOL> from neutron . agent . linux import utils <EOL> from neutron . common import constants <EOL> from neutron . common import utils as common_utils <EOL> from neutron . tests . common import l3_test_common <EOL> from neutron . tests . common import net_helpers <EOL> from neutron . tests . functional . agent . l3 import framework <EOL> class L3HATestCase ( framework . L3AgentTestFramework ) : <EOL> def test_ha_router_update_floatingip_statuses ( self ) : <EOL> self . _test_update_floatingip_statuses ( <EOL> self . generate_router_info ( enable_ha = True ) ) <EOL> def test_keepalived_state_change_notification ( self ) : <EOL> enqueue_mock = mock . patch . object ( <EOL> self . agent , '<STR_LIT>' ) . start ( ) <EOL> router_info = self . generate_router_info ( enable_ha = True ) <EOL> router = self . manage_router ( self . agent , router_info ) <EOL> utils . wait_until_true ( lambda : router . ha_state == '<STR_LIT>' ) <EOL> self . fail_ha_router ( router ) <EOL> utils . wait_until_true ( lambda : router . ha_state == '<STR_LIT>' ) <EOL> utils . wait_until_true ( lambda : enqueue_mock . call_count == <NUM_LIT:3> ) <EOL> calls = [ args [ <NUM_LIT:0> ] for args in enqueue_mock . call_args_list ] <EOL> self . assertEqual ( ( router . router_id , '<STR_LIT>' ) , calls [ <NUM_LIT:0> ] ) <EOL> self . assertEqual ( ( router . router_id , '<STR_LIT>' ) , calls [ <NUM_LIT:1> ] ) <EOL> self . assertEqual ( ( router . router_id , '<STR_LIT>' ) , calls [ <NUM_LIT:2> ] ) <EOL> def _expected_rpc_report ( self , expected ) : <EOL> calls = ( args [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] for args in <EOL> self . agent . plugin_rpc . update_ha_routers_states . call_args_list ) <EOL> actual_router_states = { } <EOL> for call in calls : <EOL> for router_id , state in six . iteritems ( call ) : <EOL> actual_router_states [ router_id ] = state <EOL> return actual_router_states == expected <EOL> def test_keepalived_state_change_bulk_rpc ( self ) : <EOL> router_info = self . generate_router_info ( enable_ha = True ) <EOL> router1 = self . manage_router ( self . agent , router_info ) <EOL> self . fail_ha_router ( router1 ) <EOL> router_info = self . generate_router_info ( enable_ha = True ) <EOL> router2 = self . manage_router ( self . agent , router_info ) <EOL> utils . wait_until_true ( lambda : router1 . ha_state == '<STR_LIT>' ) <EOL> utils . wait_until_true ( lambda : router2 . ha_state == '<STR_LIT>' ) <EOL> utils . wait_until_true ( <EOL> lambda : self . _expected_rpc_report ( <EOL> { router1 . router_id : '<STR_LIT>' , router2 . router_id : '<STR_LIT>' } ) ) <EOL> def test_ha_router_lifecycle ( self ) : <EOL> self . _router_lifecycle ( enable_ha = True ) <EOL> def test_conntrack_disassociate_fip_ha_router ( self ) : <EOL> self . _test_conntrack_disassociate_fip ( ha = True ) <EOL> def test_ipv6_ha_router_lifecycle ( self ) : <EOL> self . _router_lifecycle ( enable_ha = True , ip_version = <NUM_LIT:6> ) <EOL> def test_ipv6_ha_router_lifecycle_with_no_gw_subnet ( self ) : <EOL> self . agent . conf . set_override ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . _router_lifecycle ( enable_ha = True , ip_version = <NUM_LIT:6> , <EOL> v6_ext_gw_with_sub = False ) <EOL> def test_ipv6_ha_router_lifecycle_with_no_gw_subnet_for_router_advts ( self ) : <EOL> self . _router_lifecycle ( enable_ha = True , dual_stack = True , <EOL> v6_ext_gw_with_sub = False ) <EOL> def test_keepalived_configuration ( self ) : <EOL> router_info = self . generate_router_info ( enable_ha = True ) <EOL> router = self . manage_router ( self . agent , router_info ) <EOL> expected = self . get_expected_keepalive_configuration ( router ) <EOL> self . assertEqual ( expected , <EOL> router . keepalived_manager . get_conf_on_disk ( ) ) <EOL> router . router = copy . deepcopy ( router . router ) <EOL> existing_fip = '<STR_LIT>' <EOL> new_fip = '<STR_LIT>' <EOL> self . _add_fip ( router , new_fip ) <EOL> subnet_id = framework . _uuid ( ) <EOL> fixed_ips = [ { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> '<STR_LIT>' : subnet_id } ] <EOL> subnets = [ { '<STR_LIT:id>' : subnet_id , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> router . router [ '<STR_LIT>' ] [ '<STR_LIT>' ] = subnets <EOL> router . router [ '<STR_LIT>' ] [ '<STR_LIT>' ] = fixed_ips <EOL> router . process ( self . agent ) <EOL> new_config = router . keepalived_manager . config . get_config_str ( ) <EOL> old_gw = '<STR_LIT>' <EOL> new_gw = '<STR_LIT>' <EOL> old_external_device_ip = '<STR_LIT>' <EOL> new_external_device_ip = '<STR_LIT>' <EOL> self . assertIn ( existing_fip , new_config ) <EOL> self . assertIn ( new_fip , new_config ) <EOL> self . assertNotIn ( old_gw , new_config ) <EOL> self . assertIn ( new_gw , new_config ) <EOL> external_port = router . get_ex_gw_port ( ) <EOL> external_device_name = router . get_external_device_name ( <EOL> external_port [ '<STR_LIT:id>' ] ) <EOL> self . assertNotIn ( '<STR_LIT>' % <EOL> ( old_external_device_ip , external_device_name ) , <EOL> new_config ) <EOL> self . assertIn ( '<STR_LIT>' % <EOL> ( new_external_device_ip , external_device_name ) , <EOL> new_config ) <EOL> def test_ha_router_conf_on_restarted_agent ( self ) : <EOL> router_info = self . generate_router_info ( enable_ha = True ) <EOL> router1 = self . manage_router ( self . agent , router_info ) <EOL> self . _add_fip ( router1 , '<STR_LIT>' ) <EOL> restarted_agent = neutron_l3_agent . L3NATAgentWithStateReport ( <EOL> self . agent . host , self . agent . conf ) <EOL> self . manage_router ( restarted_agent , router1 . router ) <EOL> utils . wait_until_true ( lambda : self . floating_ips_configured ( router1 ) ) <EOL> self . assertIn ( <EOL> router1 . _get_primary_vip ( ) , <EOL> self . _get_addresses_on_device ( <EOL> router1 . ns_name , <EOL> router1 . get_ha_device_name ( ) ) ) <EOL> def test_ha_router_ipv6_radvd_status ( self ) : <EOL> router_info = self . generate_router_info ( ip_version = <NUM_LIT:6> , enable_ha = True ) <EOL> router1 = self . manage_router ( self . agent , router_info ) <EOL> utils . wait_until_true ( lambda : router1 . ha_state == '<STR_LIT>' ) <EOL> utils . wait_until_true ( lambda : router1 . radvd . enabled ) <EOL> def _check_lla_status ( router , expected ) : <EOL> internal_devices = router . router [ l3_constants . INTERFACE_KEY ] <EOL> for device in internal_devices : <EOL> lladdr = ip_lib . get_ipv6_lladdr ( device [ '<STR_LIT>' ] ) <EOL> exists = ip_lib . device_exists_with_ips_and_mac ( <EOL> router . get_internal_device_name ( device [ '<STR_LIT:id>' ] ) , [ lladdr ] , <EOL> device [ '<STR_LIT>' ] , router . ns_name ) <EOL> self . assertEqual ( expected , exists ) <EOL> _check_lla_status ( router1 , True ) <EOL> device_name = router1 . get_ha_device_name ( ) <EOL> ha_device = ip_lib . IPDevice ( device_name , namespace = router1 . ns_name ) <EOL> ha_device . link . set_down ( ) <EOL> utils . wait_until_true ( lambda : router1 . ha_state == '<STR_LIT>' ) <EOL> utils . wait_until_true ( lambda : not router1 . radvd . enabled , timeout = <NUM_LIT:10> ) <EOL> _check_lla_status ( router1 , False ) <EOL> def test_ha_router_process_ipv6_subnets_to_existing_port ( self ) : <EOL> router_info = self . generate_router_info ( enable_ha = True , ip_version = <NUM_LIT:6> ) <EOL> router = self . manage_router ( self . agent , router_info ) <EOL> def verify_ip_in_keepalived_config ( router , iface ) : <EOL> config = router . keepalived_manager . config . get_config_str ( ) <EOL> ip_cidrs = common_utils . fixed_ip_cidrs ( iface [ '<STR_LIT>' ] ) <EOL> for ip_addr in ip_cidrs : <EOL> self . assertIn ( ip_addr , config ) <EOL> interface_id = router . router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> slaac = constants . IPV6_SLAAC <EOL> slaac_mode = { '<STR_LIT>' : slaac , '<STR_LIT>' : slaac } <EOL> self . _add_internal_interface_by_subnet ( router . router , count = <NUM_LIT:1> , <EOL> ip_version = <NUM_LIT:6> , ipv6_subnet_modes = [ slaac_mode ] , <EOL> interface_id = interface_id ) <EOL> router . process ( self . agent ) <EOL> utils . wait_until_true ( lambda : router . ha_state == '<STR_LIT>' ) <EOL> internal_iface = router . router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( <NUM_LIT:2> , len ( internal_iface [ '<STR_LIT>' ] ) ) <EOL> self . _assert_internal_devices ( router ) <EOL> verify_ip_in_keepalived_config ( router , internal_iface ) <EOL> interfaces = copy . deepcopy ( router . router . get ( <EOL> l3_constants . INTERFACE_KEY , [ ] ) ) <EOL> fixed_ips , subnets = [ ] , [ ] <EOL> fixed_ips . append ( interfaces [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> subnets . append ( interfaces [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> interfaces [ <NUM_LIT:0> ] . update ( { '<STR_LIT>' : fixed_ips , '<STR_LIT>' : subnets } ) <EOL> router . router [ l3_constants . INTERFACE_KEY ] = interfaces <EOL> router . process ( self . agent ) <EOL> internal_iface = router . router [ l3_constants . INTERFACE_KEY ] [ <NUM_LIT:0> ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( internal_iface [ '<STR_LIT>' ] ) ) <EOL> self . _assert_internal_devices ( router ) <EOL> verify_ip_in_keepalived_config ( router , internal_iface ) <EOL> def test_delete_external_gateway_on_standby_router ( self ) : <EOL> router_info = self . generate_router_info ( enable_ha = True ) <EOL> router = self . manage_router ( self . agent , router_info ) <EOL> self . fail_ha_router ( router ) <EOL> utils . wait_until_true ( lambda : router . ha_state == '<STR_LIT>' ) <EOL> port = router . get_ex_gw_port ( ) <EOL> interface_name = router . get_external_device_name ( port [ '<STR_LIT:id>' ] ) <EOL> router . external_gateway_removed ( port , interface_name ) <EOL> def test_removing_floatingip_immediately ( self ) : <EOL> router_info = self . generate_router_info ( enable_ha = True ) <EOL> router = self . manage_router ( self . agent , router_info ) <EOL> ex_gw_port = router . get_ex_gw_port ( ) <EOL> interface_name = router . get_external_device_interface_name ( ex_gw_port ) <EOL> utils . wait_until_true ( lambda : router . ha_state == '<STR_LIT>' ) <EOL> self . _add_fip ( router , '<STR_LIT>' , fixed_address = '<STR_LIT>' ) <EOL> router . process ( self . agent ) <EOL> router . router [ l3_constants . FLOATINGIP_KEY ] = [ ] <EOL> router . configure_fip_addresses ( interface_name ) <EOL> class L3HATestFailover ( framework . L3AgentTestFramework ) : <EOL> NESTED_NAMESPACE_SEPARATOR = '<STR_LIT:@>' <EOL> def setUp ( self ) : <EOL> super ( L3HATestFailover , self ) . setUp ( ) <EOL> conf = self . _configure_agent ( '<STR_LIT>' ) <EOL> self . failover_agent = neutron_l3_agent . L3NATAgentWithStateReport ( <EOL> '<STR_LIT>' , conf ) <EOL> br_int_1 = self . _get_agent_ovs_integration_bridge ( self . agent ) <EOL> br_int_2 = self . _get_agent_ovs_integration_bridge ( self . failover_agent ) <EOL> veth1 , veth2 = self . useFixture ( net_helpers . VethFixture ( ) ) . ports <EOL> br_int_1 . add_port ( veth1 . name ) <EOL> br_int_2 . add_port ( veth2 . name ) <EOL> def test_ha_router_failover ( self ) : <EOL> router_info = self . generate_router_info ( enable_ha = True ) <EOL> get_ns_name = mock . patch . object ( <EOL> namespaces . RouterNamespace , '<STR_LIT>' ) . start ( ) <EOL> get_ns_name . return_value = "<STR_LIT>" % ( <EOL> '<STR_LIT>' + router_info [ '<STR_LIT:id>' ] , <EOL> self . NESTED_NAMESPACE_SEPARATOR , self . agent . host ) <EOL> router1 = self . manage_router ( self . agent , router_info ) <EOL> router_info_2 = copy . deepcopy ( router_info ) <EOL> router_info_2 [ l3_constants . HA_INTERFACE_KEY ] = ( <EOL> l3_test_common . get_ha_interface ( ip = '<STR_LIT>' , <EOL> mac = '<STR_LIT>' ) ) <EOL> get_ns_name . return_value = "<STR_LIT>" % ( <EOL> namespaces . RouterNamespace . _get_ns_name ( router_info_2 [ '<STR_LIT:id>' ] ) , <EOL> self . NESTED_NAMESPACE_SEPARATOR , self . failover_agent . host ) <EOL> router2 = self . manage_router ( self . failover_agent , router_info_2 ) <EOL> utils . wait_until_true ( lambda : router1 . ha_state == '<STR_LIT>' ) <EOL> utils . wait_until_true ( lambda : router2 . ha_state == '<STR_LIT>' ) <EOL> self . fail_ha_router ( router1 ) <EOL> utils . wait_until_true ( lambda : router2 . ha_state == '<STR_LIT>' ) <EOL> utils . wait_until_true ( lambda : router1 . ha_state == '<STR_LIT>' ) </s>
