return router_db
def _delete_current_gw_port(self, context, router_id, router, new_network):
+ """
+ Overriden here to handle deletion of dvr internal ports.
+
+ If there is a valid router update with gateway port to be deleted,
+ then go ahead and delete the csnat ports and the floatingip
+ agent gateway port associated with the dvr router.
+ """
+
+ gw_ext_net_id = (
+ router.gw_port['network_id'] if router.gw_port else None)
+
super(L3_NAT_with_dvr_db_mixin,
self)._delete_current_gw_port(context, router_id,
router, new_network)
- if router.extra_attributes.distributed:
+ if (is_distributed_router(router) and
+ gw_ext_net_id != new_network):
self.delete_csnat_router_interface_ports(
context.elevated(), router)
+ # NOTE(Swami): Delete the Floatingip agent gateway port
+ # on all hosts when it is the last gateway port in the
+ # given external network.
+ filters = {'network_id': [gw_ext_net_id],
+ 'device_owner': [l3_const.DEVICE_OWNER_ROUTER_GW]}
+ ext_net_gw_ports = self._core_plugin.get_ports(
+ context.elevated(), filters)
+ if not ext_net_gw_ports:
+ self._delete_floatingip_agent_gateway_port(
+ context.elevated(), None, gw_ext_net_id)
def _create_gw_port(self, context, router_id, router, new_network,
ext_ips):
ports = self._core_plugin.get_ports(context,
filters=device_filter)
for p in ports:
- if self._get_vm_port_hostid(context, p['id'], p) == host_id:
+ if not host_id or p[portbindings.HOST_ID] == host_id:
self._core_plugin.ipam.delete_port(context, p['id'])
- return
+ if host_id:
+ return
def create_fip_agent_gw_port_if_not_exists(
self, context, network_id, host):
self._test_remove_router_interface_leaves_snat_intact(
by_subnet=False)
- def setup_create_agent_gw_port_for_network(self):
- network = self._make_network(self.fmt, '', True)
+ def setup_create_agent_gw_port_for_network(self, network=None):
+ if not network:
+ network = self._make_network(self.fmt, '', True)
network_id = network['network']['id']
port = self.core_plugin.create_port(
self.context,
self._create_router()
self.assertEqual(
2, len(self.l3_plugin._get_router_ids(self.context)))
+
+ def test_agent_gw_port_delete_when_last_gateway_for_ext_net_removed(self):
+ kwargs = {'arg_list': (external_net.EXTERNAL,),
+ external_net.EXTERNAL: True}
+ net1 = self._make_network(self.fmt, 'net1', True)
+ net2 = self._make_network(self.fmt, 'net2', True)
+ subnet1 = self._make_subnet(
+ self.fmt, net1, '10.1.0.1', '10.1.0.0/24', enable_dhcp=True)
+ subnet2 = self._make_subnet(
+ self.fmt, net2, '10.1.0.1', '10.1.0.0/24', enable_dhcp=True)
+ ext_net = self._make_network(self.fmt, 'ext_net', True, **kwargs)
+ self._make_subnet(
+ self.fmt, ext_net, '20.0.0.1', '20.0.0.0/24', enable_dhcp=True)
+ # Create first router and add an interface
+ router1 = self._create_router()
+ ext_net_id = ext_net['network']['id']
+ self.l3_plugin.add_router_interface(
+ self.context, router1['id'],
+ {'subnet_id': subnet1['subnet']['id']})
+ # Set gateway to first router
+ self.l3_plugin._update_router_gw_info(
+ self.context, router1['id'],
+ {'network_id': ext_net_id})
+ # Create second router and add an interface
+ router2 = self._create_router()
+ self.l3_plugin.add_router_interface(
+ self.context, router2['id'],
+ {'subnet_id': subnet2['subnet']['id']})
+ # Set gateway to second router
+ self.l3_plugin._update_router_gw_info(
+ self.context, router2['id'],
+ {'network_id': ext_net_id})
+ # Create an agent gateway port for the external network
+ net_id, agent_gw_port = (
+ self.setup_create_agent_gw_port_for_network(network=ext_net))
+ # Check for agent gateway ports
+ self.assertIsNotNone(
+ self.l3_plugin._get_agent_gw_ports_exist_for_network(
+ self.context, ext_net_id, "", self.l3_agent['id']))
+ self.l3_plugin._update_router_gw_info(
+ self.context, router1['id'], {})
+ # Check for agent gateway port after deleting one of the gw
+ self.assertIsNotNone(
+ self.l3_plugin._get_agent_gw_ports_exist_for_network(
+ self.context, ext_net_id, "", self.l3_agent['id']))
+ self.l3_plugin._update_router_gw_info(
+ self.context, router2['id'], {})
+ # Check for agent gateway port after deleting last gw
+ self.assertIsNone(
+ self.l3_plugin._get_agent_gw_ports_exist_for_network(
+ self.context, ext_net_id, "", self.l3_agent['id']))
self.assertTrue(result)
self.assertTrue(pv6.called)
- def test__delete_floatingip_agent_gateway_port(self):
- port = {
+ def _helper_delete_floatingip_agent_gateway_port(self, port_host):
+ ports = [{
'id': 'my_port_id',
'binding:host_id': 'foo_host',
'network_id': 'ext_network_id',
- 'device_owner': l3_const.DEVICE_OWNER_AGENT_GW
- }
- with mock.patch.object(manager.NeutronManager, 'get_plugin') as gp,\
- mock.patch.object(self.mixin,
- '_get_vm_port_hostid') as vm_host:
+ 'device_owner': l3_const.DEVICE_OWNER_ROUTER_GW
+ },
+ {
+ 'id': 'my_new_port_id',
+ 'binding:host_id': 'my_foo_host',
+ 'network_id': 'ext_network_id',
+ 'device_owner': l3_const.DEVICE_OWNER_ROUTER_GW
+ }]
+ with mock.patch.object(manager.NeutronManager, 'get_plugin') as gp:
plugin = mock.Mock()
gp.return_value = plugin
- plugin.get_ports.return_value = [port]
- vm_host.return_value = 'foo_host'
+ plugin.get_ports.return_value = ports
self.mixin._delete_floatingip_agent_gateway_port(
- self.ctx, 'foo_host', 'network_id')
+ self.ctx, port_host, 'ext_network_id')
plugin.get_ports.assert_called_with(self.ctx, filters={
- 'network_id': ['network_id'],
+ 'network_id': ['ext_network_id'],
'device_owner': [l3_const.DEVICE_OWNER_AGENT_GW]})
- plugin.ipam.delete_port.assert_called_with(self.ctx, 'my_port_id')
+ if port_host:
+ plugin.ipam.delete_port.assert_called_once_with(
+ self.ctx, 'my_port_id')
+ else:
+ plugin.ipam.delete_port.assert_called_with(
+ self.ctx, 'my_new_port_id')
+
+ def test__delete_floatingip_agent_gateway_port_without_host_id(self):
+ self._helper_delete_floatingip_agent_gateway_port(None)
+
+ def test__delete_floatingip_agent_gateway_port_with_host_id(self):
+ self._helper_delete_floatingip_agent_gateway_port(
+ 'foo_host')
+
+ def _setup_delete_current_gw_port_deletes_fip_agent_gw_port(
+ self, port=None):
+ gw_port_db = {
+ 'id': 'my_gw_id',
+ 'network_id': 'ext_net_id',
+ 'device_owner': l3_const.DEVICE_OWNER_ROUTER_GW
+ }
+ router = mock.MagicMock()
+ router.extra_attributes.distributed = True
+ router['gw_port_id'] = gw_port_db['id']
+ router.gw_port = gw_port_db
+ with mock.patch.object(manager.NeutronManager, 'get_plugin') as gp,\
+ mock.patch.object(l3_dvr_db.l3_db.L3_NAT_db_mixin,
+ '_delete_current_gw_port'),\
+ mock.patch.object(
+ self.mixin,
+ '_get_router') as grtr,\
+ mock.patch.object(
+ self.mixin,
+ 'delete_csnat_router_interface_ports') as del_csnat_port,\
+ mock.patch.object(
+ self.mixin,
+ '_delete_floatingip_agent_gateway_port') as del_agent_gw_port:
+ plugin = mock.Mock()
+ gp.return_value = plugin
+ plugin.get_ports.return_value = port
+ grtr.return_value = router
+ self.mixin._delete_current_gw_port(
+ self.ctx, router['id'], router, 'ext_network_id')
+ return router, plugin, del_csnat_port, del_agent_gw_port
+
+ def test_delete_current_gw_port_deletes_fip_agent_gw_port(self):
+ rtr, plugin, d_csnat_port, d_agent_gw_port = (
+ self._setup_delete_current_gw_port_deletes_fip_agent_gw_port())
+ self.assertTrue(d_csnat_port.called)
+ self.assertTrue(d_agent_gw_port.called)
+ d_csnat_port.assert_called_once_with(
+ mock.ANY, rtr)
+ d_agent_gw_port.assert_called_once_with(
+ mock.ANY, None, 'ext_net_id')
+
+ def test_delete_current_gw_port_never_calls_delete_fip_agent_gw_port(self):
+ port = [{
+ 'id': 'my_port_id',
+ 'network_id': 'ext_net_id',
+ 'device_owner': l3_const.DEVICE_OWNER_ROUTER_GW
+ },
+ {
+ 'id': 'my_new_port_id',
+ 'network_id': 'ext_net_id',
+ 'device_owner': l3_const.DEVICE_OWNER_ROUTER_GW
+ }]
+ rtr, plugin, d_csnat_port, d_agent_gw_port = (
+ self._setup_delete_current_gw_port_deletes_fip_agent_gw_port(
+ port=port))
+ self.assertTrue(d_csnat_port.called)
+ self.assertFalse(d_agent_gw_port.called)
+ d_csnat_port.assert_called_once_with(
+ mock.ANY, rtr)
def _delete_floatingip_test_setup(self, floatingip):
fip_id = floatingip['id']
Code Review / openstack-build / neutron-build.git / commitdiff