<s> import mock <EOL> from neutron_lib import constants as l3_const <EOL> from neutron_lib import exceptions <EOL> from oslo_utils import uuidutils <EOL> from neutron . common import constants as n_const <EOL> from neutron import context <EOL> from neutron . db import agents_db <EOL> from neutron . db import common_db_mixin <EOL> from neutron . db import l3_agentschedulers_db <EOL> from neutron . db import l3_dvr_db <EOL> from neutron . extensions import portbindings <EOL> from neutron import manager <EOL> from neutron . plugins . common import constants as plugin_const <EOL> from neutron . tests . unit . db import test_db_base_plugin_v2 <EOL> _uuid = uuidutils . generate_uuid <EOL> class FakeL3Plugin ( common_db_mixin . CommonDbMixin , <EOL> l3_dvr_db . L3_NAT_with_dvr_db_mixin , <EOL> l3_agentschedulers_db . L3AgentSchedulerDbMixin , <EOL> agents_db . AgentDbMixin ) : <EOL> pass <EOL> class L3DvrTestCase ( test_db_base_plugin_v2 . NeutronDbPluginV2TestCase ) : <EOL> def setUp ( self ) : <EOL> core_plugin = '<STR_LIT>' <EOL> super ( L3DvrTestCase , self ) . setUp ( plugin = core_plugin ) <EOL> self . core_plugin = manager . NeutronManager . get_plugin ( ) <EOL> self . ctx = context . get_admin_context ( ) <EOL> self . mixin = FakeL3Plugin ( ) <EOL> def _create_router ( self , router ) : <EOL> with self . ctx . session . begin ( subtransactions = True ) : <EOL> return self . mixin . _create_router_db ( self . ctx , router , '<STR_LIT>' ) <EOL> def _test__create_router_db ( self , expected = False , distributed = None ) : <EOL> router = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : True } <EOL> if distributed is not None : <EOL> router [ '<STR_LIT>' ] = distributed <EOL> result = self . _create_router ( router ) <EOL> self . assertEqual ( expected , result . extra_attributes [ '<STR_LIT>' ] ) <EOL> def test_create_router_db_default ( self ) : <EOL> self . _test__create_router_db ( expected = False ) <EOL> def test_create_router_db_centralized ( self ) : <EOL> self . _test__create_router_db ( expected = False , distributed = False ) <EOL> def test_create_router_db_distributed ( self ) : <EOL> self . _test__create_router_db ( expected = True , distributed = True ) <EOL> def test__validate_router_migration_on_router_update ( self ) : <EOL> router = { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True <EOL> } <EOL> router_db = self . _create_router ( router ) <EOL> self . assertIsNone ( self . mixin . _validate_router_migration ( <EOL> self . ctx , router_db , { '<STR_LIT:name>' : '<STR_LIT>' } ) ) <EOL> def test__validate_router_migration_raise_error ( self ) : <EOL> router = { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True <EOL> } <EOL> router_db = self . _create_router ( router ) <EOL> self . assertRaises ( exceptions . BadRequest , <EOL> self . mixin . _validate_router_migration , <EOL> self . ctx , router_db , { '<STR_LIT>' : False } ) <EOL> def test_upgrade_active_router_to_distributed_validation_failure ( self ) : <EOL> router = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : True } <EOL> router_db = self . _create_router ( router ) <EOL> update = { '<STR_LIT>' : True } <EOL> self . assertRaises ( exceptions . BadRequest , <EOL> self . mixin . _validate_router_migration , <EOL> self . ctx , router_db , update ) <EOL> def test_update_router_db_centralized_to_distributed ( self ) : <EOL> router = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : True } <EOL> agent = { '<STR_LIT:id>' : _uuid ( ) } <EOL> distributed = { '<STR_LIT>' : True } <EOL> router_db = self . _create_router ( router ) <EOL> router_id = router_db [ '<STR_LIT:id>' ] <EOL> self . assertFalse ( router_db . extra_attributes . distributed ) <EOL> self . mixin . _get_router = mock . Mock ( return_value = router_db ) <EOL> self . mixin . _validate_router_migration = mock . Mock ( ) <EOL> self . mixin . _update_distributed_attr = mock . Mock ( ) <EOL> self . mixin . list_l3_agents_hosting_router = mock . Mock ( <EOL> return_value = { '<STR_LIT>' : [ agent ] } ) <EOL> self . mixin . _unbind_router = mock . Mock ( ) <EOL> router_db = self . mixin . _update_router_db ( <EOL> self . ctx , router_id , distributed ) <EOL> self . assertTrue ( router_db . extra_attributes . distributed ) <EOL> self . assertEqual ( <NUM_LIT:1> , <EOL> self . mixin . _update_distributed_attr . call_count ) <EOL> def _test_get_device_owner ( self , is_distributed = False , <EOL> expected = l3_const . DEVICE_OWNER_ROUTER_INTF , <EOL> pass_router_id = True ) : <EOL> router = { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : is_distributed <EOL> } <EOL> router_db = self . _create_router ( router ) <EOL> router_pass = router_db [ '<STR_LIT:id>' ] if pass_router_id else router_db <EOL> with mock . patch . object ( self . mixin , '<STR_LIT>' ) as f : <EOL> f . return_value = router_db <EOL> result = self . mixin . _get_device_owner ( self . ctx , router_pass ) <EOL> self . assertEqual ( expected , result ) <EOL> def test_get_device_owner_by_router_id ( self ) : <EOL> self . _test_get_device_owner ( ) <EOL> def test__get_device_owner_centralized ( self ) : <EOL> self . _test_get_device_owner ( pass_router_id = False ) <EOL> def test__get_device_owner_distributed ( self ) : <EOL> self . _test_get_device_owner ( <EOL> is_distributed = True , <EOL> expected = l3_const . DEVICE_OWNER_DVR_INTERFACE , <EOL> pass_router_id = False ) <EOL> def _test__is_distributed_router ( self , router , expected ) : <EOL> result = l3_dvr_db . is_distributed_router ( router ) <EOL> self . assertEqual ( expected , result ) <EOL> def test__is_distributed_router_by_db_object ( self ) : <EOL> router = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : True } <EOL> router_db = self . _create_router ( router ) <EOL> self . mixin . _get_device_owner ( mock . ANY , router_db ) <EOL> def test__is_distributed_router_default ( self ) : <EOL> router = { '<STR_LIT:id>' : '<STR_LIT>' } <EOL> self . _test__is_distributed_router ( router , False ) <EOL> def test__is_distributed_router_centralized ( self ) : <EOL> router = { '<STR_LIT:id>' : '<STR_LIT>' , '<STR_LIT>' : False } <EOL> self . _test__is_distributed_router ( router , False ) <EOL> def test__is_distributed_router_distributed ( self ) : <EOL> router = { '<STR_LIT:id>' : '<STR_LIT>' , '<STR_LIT>' : True } <EOL> self . _test__is_distributed_router ( router , True ) <EOL> def test__get_agent_gw_ports_exist_for_network ( self ) : <EOL> with mock . patch . object ( manager . NeutronManager , '<STR_LIT>' ) as gp : <EOL> plugin = mock . Mock ( ) <EOL> gp . return_value = plugin <EOL> plugin . get_ports . return_value = [ ] <EOL> self . mixin . _get_agent_gw_ports_exist_for_network ( <EOL> self . ctx , '<STR_LIT>' , '<STR_LIT:host>' , '<STR_LIT>' ) <EOL> plugin . get_ports . assert_called_with ( self . ctx , { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ l3_const . DEVICE_OWNER_AGENT_GW ] } ) <EOL> def _test_prepare_direct_delete_dvr_internal_ports ( self , port ) : <EOL> with mock . patch . object ( manager . NeutronManager , '<STR_LIT>' ) as gp : <EOL> plugin = mock . Mock ( ) <EOL> gp . return_value = plugin <EOL> plugin . get_port . return_value = port <EOL> self . mixin . _router_exists = mock . Mock ( return_value = True ) <EOL> self . assertRaises ( exceptions . ServicePortInUse , <EOL> self . mixin . prevent_l3_port_deletion , <EOL> self . ctx , <EOL> port [ '<STR_LIT:id>' ] ) <EOL> def test_prevent_delete_floatingip_agent_gateway_port ( self ) : <EOL> port = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : mock . ANY , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : l3_const . DEVICE_OWNER_AGENT_GW <EOL> } <EOL> self . _test_prepare_direct_delete_dvr_internal_ports ( port ) <EOL> def test_prevent_delete_csnat_port ( self ) : <EOL> port = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : mock . ANY , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : l3_const . DEVICE_OWNER_ROUTER_SNAT <EOL> } <EOL> self . _test_prepare_direct_delete_dvr_internal_ports ( port ) <EOL> def test__create_gw_port_with_no_gateway ( self ) : <EOL> router = { <EOL> '<STR_LIT:name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True , <EOL> } <EOL> router_db = self . _create_router ( router ) <EOL> router_id = router_db [ '<STR_LIT:id>' ] <EOL> self . assertTrue ( router_db . extra_attributes . distributed ) <EOL> with mock . patch . object ( l3_dvr_db . l3_db . L3_NAT_db_mixin , <EOL> '<STR_LIT>' ) , mock . patch . object ( <EOL> self . mixin , <EOL> '<STR_LIT>' ) as cs : <EOL> self . mixin . _create_gw_port ( <EOL> self . ctx , router_id , router_db , mock . ANY , <EOL> mock . ANY ) <EOL> self . assertFalse ( cs . call_count ) <EOL> def test_build_routers_list_with_gw_port_mismatch ( self ) : <EOL> routers = [ { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:id>' : '<STR_LIT>' } ] <EOL> gw_ports = { } <EOL> routers = self . mixin . _build_routers_list ( self . ctx , routers , gw_ports ) <EOL> self . assertIsNone ( routers [ <NUM_LIT:0> ] . get ( '<STR_LIT>' ) ) <EOL> def setup_port_has_ipv6_address ( self , port ) : <EOL> with mock . patch . object ( l3_dvr_db . l3_db . L3_NAT_db_mixin , <EOL> '<STR_LIT>' ) as pv6 : <EOL> pv6 . return_value = True <EOL> result = self . mixin . _port_has_ipv6_address ( port ) <EOL> return result , pv6 <EOL> def test__port_has_ipv6_address_for_dvr_snat_port ( self ) : <EOL> port = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : l3_const . DEVICE_OWNER_ROUTER_SNAT , <EOL> } <EOL> result , pv6 = self . setup_port_has_ipv6_address ( port ) <EOL> self . assertFalse ( result ) <EOL> self . assertFalse ( pv6 . called ) <EOL> def test__port_has_ipv6_address_for_non_snat_ports ( self ) : <EOL> port = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : l3_const . DEVICE_OWNER_DVR_INTERFACE , <EOL> } <EOL> result , pv6 = self . setup_port_has_ipv6_address ( port ) <EOL> self . assertTrue ( result ) <EOL> self . assertTrue ( pv6 . called ) <EOL> def _helper_delete_floatingip_agent_gateway_port ( self , port_host ) : <EOL> ports = [ { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> portbindings . HOST_ID : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : l3_const . DEVICE_OWNER_ROUTER_GW <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> portbindings . HOST_ID : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : l3_const . DEVICE_OWNER_ROUTER_GW <EOL> } ] <EOL> with mock . patch . object ( manager . NeutronManager , '<STR_LIT>' ) as gp : <EOL> plugin = mock . Mock ( ) <EOL> gp . return_value = plugin <EOL> plugin . get_ports . return_value = ports <EOL> self . mixin . delete_floatingip_agent_gateway_port ( <EOL> self . ctx , port_host , '<STR_LIT>' ) <EOL> plugin . get_ports . assert_called_with ( self . ctx , filters = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ l3_const . DEVICE_OWNER_AGENT_GW ] } ) <EOL> if port_host : <EOL> plugin . ipam . delete_port . assert_called_once_with ( <EOL> self . ctx , '<STR_LIT>' ) <EOL> else : <EOL> plugin . ipam . delete_port . assert_called_with ( <EOL> self . ctx , '<STR_LIT>' ) <EOL> def test_delete_floatingip_agent_gateway_port_without_host_id ( self ) : <EOL> self . _helper_delete_floatingip_agent_gateway_port ( None ) <EOL> def test_delete_floatingip_agent_gateway_port_with_host_id ( self ) : <EOL> self . _helper_delete_floatingip_agent_gateway_port ( <EOL> '<STR_LIT>' ) <EOL> def _setup_delete_current_gw_port_deletes_fip_agent_gw_port ( <EOL> self , port = None , gw_port = True ) : <EOL> router = mock . MagicMock ( ) <EOL> router . extra_attributes . distributed = True <EOL> if gw_port : <EOL> gw_port_db = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : l3_const . DEVICE_OWNER_ROUTER_GW <EOL> } <EOL> router . gw_port = gw_port_db <EOL> else : <EOL> router . gw_port = None <EOL> with mock . patch . object ( manager . NeutronManager , '<STR_LIT>' ) as gp , mock . patch . object ( l3_dvr_db . l3_db . L3_NAT_db_mixin , <EOL> '<STR_LIT>' ) , mock . patch . object ( <EOL> self . mixin , <EOL> '<STR_LIT>' ) as grtr , mock . patch . object ( <EOL> self . mixin , <EOL> '<STR_LIT>' ) as del_csnat_port , mock . patch . object ( <EOL> self . mixin , <EOL> '<STR_LIT>' ) as del_agent_gw_port , mock . patch . object ( <EOL> self . mixin . l3_rpc_notifier , <EOL> '<STR_LIT>' ) as del_fip : <EOL> plugin = mock . Mock ( ) <EOL> gp . return_value = plugin <EOL> plugin . get_ports . return_value = port <EOL> grtr . return_value = router <EOL> self . mixin . _delete_current_gw_port ( <EOL> self . ctx , router [ '<STR_LIT:id>' ] , router , '<STR_LIT>' ) <EOL> return router , plugin , del_csnat_port , del_agent_gw_port , del_fip <EOL> def test_delete_current_gw_port_deletes_fip_agent_gw_port_and_fipnamespace ( <EOL> self ) : <EOL> rtr , plugin , d_csnat_port , d_agent_gw_port , del_fip = ( <EOL> self . _setup_delete_current_gw_port_deletes_fip_agent_gw_port ( ) ) <EOL> self . assertTrue ( d_csnat_port . called ) <EOL> self . assertTrue ( d_agent_gw_port . called ) <EOL> d_csnat_port . assert_called_once_with ( <EOL> mock . ANY , rtr ) <EOL> d_agent_gw_port . assert_called_once_with ( mock . ANY , None , '<STR_LIT>' ) <EOL> del_fip . assert_called_once_with ( mock . ANY , '<STR_LIT>' ) <EOL> def test_delete_current_gw_port_never_calls_delete_fip_agent_gw_port ( self ) : <EOL> port = [ { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : l3_const . DEVICE_OWNER_ROUTER_GW <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : l3_const . DEVICE_OWNER_ROUTER_GW <EOL> } ] <EOL> rtr , plugin , d_csnat_port , d_agent_gw_port , del_fip = ( <EOL> self . _setup_delete_current_gw_port_deletes_fip_agent_gw_port ( <EOL> port = port ) ) <EOL> self . assertTrue ( d_csnat_port . called ) <EOL> self . assertFalse ( d_agent_gw_port . called ) <EOL> self . assertFalse ( del_fip . called ) <EOL> d_csnat_port . assert_called_once_with ( <EOL> mock . ANY , rtr ) <EOL> def test_delete_current_gw_port_never_calls_delete_fipnamespace ( self ) : <EOL> rtr , plugin , d_csnat_port , d_agent_gw_port , del_fip = ( <EOL> self . _setup_delete_current_gw_port_deletes_fip_agent_gw_port ( <EOL> gw_port = False ) ) <EOL> self . assertFalse ( d_csnat_port . called ) <EOL> self . assertFalse ( d_agent_gw_port . called ) <EOL> self . assertFalse ( del_fip . called ) <EOL> def _floatingip_on_port_test_setup ( self , hostid ) : <EOL> router = { '<STR_LIT:id>' : '<STR_LIT>' , '<STR_LIT>' : True } <EOL> floatingip = { <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:host>' : hostid <EOL> } <EOL> if not hostid : <EOL> hostid = '<STR_LIT>' <EOL> routers = { <EOL> '<STR_LIT>' : router <EOL> } <EOL> fipagent = { <EOL> '<STR_LIT:id>' : _uuid ( ) <EOL> } <EOL> self . mixin . _get_dvr_service_port_hostid = mock . Mock ( <EOL> return_value = hostid ) <EOL> self . mixin . _get_agent_by_type_and_host = mock . Mock ( <EOL> return_value = fipagent ) <EOL> self . mixin . _get_fip_sync_interfaces = mock . Mock ( <EOL> return_value = '<STR_LIT>' ) <EOL> agent = mock . Mock ( ) <EOL> agent . id = fipagent [ '<STR_LIT:id>' ] <EOL> self . mixin . _process_floating_ips_dvr ( self . ctx , routers , [ floatingip ] , <EOL> hostid , agent ) <EOL> return ( router , floatingip ) <EOL> def test_floatingip_on_port_not_host ( self ) : <EOL> router , fip = self . _floatingip_on_port_test_setup ( None ) <EOL> self . assertNotIn ( l3_const . FLOATINGIP_KEY , router ) <EOL> self . assertNotIn ( n_const . FLOATINGIP_AGENT_INTF_KEY , router ) <EOL> def test_floatingip_on_port_with_host ( self ) : <EOL> router , fip = self . _floatingip_on_port_test_setup ( _uuid ( ) ) <EOL> self . assertTrue ( self . mixin . _get_fip_sync_interfaces . called ) <EOL> self . assertIn ( l3_const . FLOATINGIP_KEY , router ) <EOL> self . assertIn ( n_const . FLOATINGIP_AGENT_INTF_KEY , router ) <EOL> self . assertIn ( fip , router [ l3_const . FLOATINGIP_KEY ] ) <EOL> self . assertIn ( '<STR_LIT>' , <EOL> router [ n_const . FLOATINGIP_AGENT_INTF_KEY ] ) <EOL> def _setup_test_create_floatingip ( <EOL> self , fip , floatingip_db , router_db ) : <EOL> port = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> portbindings . HOST_ID : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> with mock . patch . object ( self . mixin , '<STR_LIT>' ) as grtr , mock . patch . object ( self . mixin , <EOL> '<STR_LIT>' ) as vmp , mock . patch . object ( <EOL> self . mixin , <EOL> '<STR_LIT>' <EOL> ) as mvmp , mock . patch . object ( <EOL> self . mixin , <EOL> '<STR_LIT>' ) as c_fip , mock . patch . object ( l3_dvr_db . l3_db . L3_NAT_db_mixin , <EOL> '<STR_LIT>' ) : <EOL> grtr . return_value = router_db <EOL> vmp . return_value = '<STR_LIT>' <EOL> mvmp . return_value = '<STR_LIT>' <EOL> self . mixin . _update_fip_assoc ( <EOL> self . ctx , fip , floatingip_db , port ) <EOL> return c_fip <EOL> def test_create_floatingip_agent_gw_port_with_dvr_router ( self ) : <EOL> floatingip = { <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> router = { '<STR_LIT:id>' : '<STR_LIT>' , '<STR_LIT>' : True } <EOL> fip = { <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : _uuid ( ) <EOL> } <EOL> create_fip = ( <EOL> self . _setup_test_create_floatingip ( <EOL> fip , floatingip , router ) ) <EOL> self . assertTrue ( create_fip . called ) <EOL> def test_create_floatingip_agent_gw_port_with_non_dvr_router ( self ) : <EOL> floatingip = { <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> router = { '<STR_LIT:id>' : '<STR_LIT>' , '<STR_LIT>' : False } <EOL> fip = { <EOL> '<STR_LIT:id>' : _uuid ( ) , <EOL> '<STR_LIT>' : _uuid ( ) <EOL> } <EOL> create_fip = ( <EOL> self . _setup_test_create_floatingip ( <EOL> fip , floatingip , router ) ) <EOL> self . assertFalse ( create_fip . called ) <EOL> def test_remove_router_interface_csnat_ports_removal ( self ) : <EOL> router_dict = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True } <EOL> router = self . _create_router ( router_dict ) <EOL> plugin = mock . MagicMock ( ) <EOL> with self . network ( ) as net_ext , self . subnet ( ) as subnet1 , self . subnet ( cidr = '<STR_LIT>' ) as subnet2 : <EOL> ext_net_id = net_ext [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> self . core_plugin . update_network ( <EOL> self . ctx , ext_net_id , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : True } } ) <EOL> self . mixin . update_router ( <EOL> self . ctx , router [ '<STR_LIT:id>' ] , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : <EOL> { '<STR_LIT>' : ext_net_id } } } ) <EOL> self . mixin . add_router_interface ( self . ctx , router [ '<STR_LIT:id>' ] , <EOL> { '<STR_LIT>' : subnet1 [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] } ) <EOL> self . mixin . add_router_interface ( self . ctx , router [ '<STR_LIT:id>' ] , <EOL> { '<STR_LIT>' : subnet2 [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] } ) <EOL> csnat_filters = { '<STR_LIT>' : <EOL> [ l3_const . DEVICE_OWNER_ROUTER_SNAT ] } <EOL> csnat_ports = self . core_plugin . get_ports ( <EOL> self . ctx , filters = csnat_filters ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( csnat_ports ) ) <EOL> dvr_filters = { '<STR_LIT>' : <EOL> [ l3_const . DEVICE_OWNER_DVR_INTERFACE ] } <EOL> dvr_ports = self . core_plugin . get_ports ( <EOL> self . ctx , filters = dvr_filters ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( dvr_ports ) ) <EOL> with mock . patch . object ( manager . NeutronManager , <EOL> '<STR_LIT>' ) as get_svc_plugin : <EOL> get_svc_plugin . return_value = { <EOL> plugin_const . L3_ROUTER_NAT : plugin } <EOL> self . mixin . manager = manager <EOL> self . mixin . remove_router_interface ( <EOL> self . ctx , router [ '<STR_LIT:id>' ] , { '<STR_LIT>' : dvr_ports [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] } ) <EOL> csnat_ports = self . core_plugin . get_ports ( <EOL> self . ctx , filters = csnat_filters ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( csnat_ports ) ) <EOL> self . assertEqual ( dvr_ports [ <NUM_LIT:1> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] , <EOL> csnat_ports [ <NUM_LIT:0> ] [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> dvr_ports = self . core_plugin . get_ports ( <EOL> self . ctx , filters = dvr_filters ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( dvr_ports ) ) <EOL> def test__validate_router_migration_notify_advanced_services ( self ) : <EOL> router = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : False } <EOL> router_db = self . _create_router ( router ) <EOL> with mock . patch . object ( l3_dvr_db . registry , '<STR_LIT>' ) as mock_notify : <EOL> self . mixin . _validate_router_migration ( <EOL> self . ctx , router_db , { '<STR_LIT>' : True } ) <EOL> kwargs = { '<STR_LIT>' : self . ctx , '<STR_LIT>' : router_db } <EOL> mock_notify . assert_called_once_with ( <EOL> '<STR_LIT>' , '<STR_LIT>' , self . mixin , ** kwargs ) <EOL> def _test_update_arp_entry_for_dvr_service_port ( <EOL> self , device_owner , action ) : <EOL> router_dict = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True } <EOL> router = self . _create_router ( router_dict ) <EOL> with mock . patch . object ( manager . NeutronManager , '<STR_LIT>' ) as gp : <EOL> plugin = mock . Mock ( ) <EOL> l3_notify = self . mixin . l3_rpc_notifier = mock . Mock ( ) <EOL> gp . return_value = plugin <EOL> port = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : device_owner <EOL> } <EOL> dvr_port = { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : mock . ANY , <EOL> '<STR_LIT>' : l3_const . DEVICE_OWNER_DVR_INTERFACE , <EOL> '<STR_LIT>' : router [ '<STR_LIT:id>' ] <EOL> } <EOL> plugin . get_ports . return_value = [ dvr_port ] <EOL> if action == '<STR_LIT>' : <EOL> self . mixin . update_arp_entry_for_dvr_service_port ( <EOL> self . ctx , port ) <EOL> self . assertEqual ( <NUM_LIT:3> , l3_notify . add_arp_entry . call_count ) <EOL> elif action == '<STR_LIT>' : <EOL> self . mixin . delete_arp_entry_for_dvr_service_port ( <EOL> self . ctx , port ) <EOL> self . assertTrue ( <NUM_LIT:3> , l3_notify . del_arp_entry . call_count ) <EOL> def test_update_arp_entry_for_dvr_service_port_added ( self ) : <EOL> action = '<STR_LIT>' <EOL> device_owner = l3_const . DEVICE_OWNER_LOADBALANCER <EOL> self . _test_update_arp_entry_for_dvr_service_port ( device_owner , action ) <EOL> def test_update_arp_entry_for_dvr_service_port_deleted ( self ) : <EOL> action = '<STR_LIT>' <EOL> device_owner = l3_const . DEVICE_OWNER_LOADBALANCER <EOL> self . _test_update_arp_entry_for_dvr_service_port ( device_owner , action ) <EOL> def test_add_router_interface_csnat_ports_failure ( self ) : <EOL> router_dict = { '<STR_LIT:name>' : '<STR_LIT>' , '<STR_LIT>' : True , <EOL> '<STR_LIT>' : True } <EOL> router = self . _create_router ( router_dict ) <EOL> with self . network ( ) as net_ext , self . subnet ( ) as subnet : <EOL> ext_net_id = net_ext [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] <EOL> self . core_plugin . update_network ( <EOL> self . ctx , ext_net_id , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : True } } ) <EOL> self . mixin . update_router ( <EOL> self . ctx , router [ '<STR_LIT:id>' ] , <EOL> { '<STR_LIT>' : { '<STR_LIT>' : <EOL> { '<STR_LIT>' : ext_net_id } } } ) <EOL> with mock . patch . object ( <EOL> self . mixin , '<STR_LIT>' ) as f : <EOL> f . side_effect = RuntimeError ( ) <EOL> self . assertRaises ( <EOL> RuntimeError , <EOL> self . mixin . add_router_interface , <EOL> self . ctx , router [ '<STR_LIT:id>' ] , <EOL> { '<STR_LIT>' : subnet [ '<STR_LIT>' ] [ '<STR_LIT:id>' ] } ) <EOL> filters = { <EOL> '<STR_LIT>' : [ router [ '<STR_LIT:id>' ] ] , <EOL> } <EOL> router_ports = self . core_plugin . get_ports ( self . ctx , filters ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( router_ports ) ) <EOL> self . assertEqual ( l3_const . DEVICE_OWNER_ROUTER_GW , <EOL> router_ports [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) </s>
