Fix DVR interface delete by port when gateway is set

When removing a DVR interface by port, the subnet_id
passed to delete_csnat_router_interface_ports is None,
and so it deletes all the DVR SNAT ports for the
router.

This patch fixes this issue by passing in the right
subnet_id to the delete_csnat_router_interface_ports.

Change-Id: I16735195c6575454876acd0e99ef45f382963566
Closes-Bug: #1443524
Co-Authored-By: Swaminathan Vasudevan <swaminathan.vasudevan@hp.com>
Co-Authored-By: Oleg Bondarev <obondarev@mirantis.com>

diff --git a/neutron/db/l3_dvr_db.py b/neutron/db/l3_dvr_db.py
index f332a6db7dc429c0469c2691cd788dfcd0c6b66b..9438ab04371ce19e65fcaa510551bc21520ef622 100644 (file)
--- a/neutron/db/l3_dvr_db.py
+++ b/neutron/db/l3_dvr_db.py
@@ -319,6 +319,28 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         return super(L3_NAT_with_dvr_db_mixin,
                      self)._port_has_ipv6_address(port)
 
+    def _check_dvr_router_remove_required_and_notify_agent(
+        self, context, router, port, subnets):
+        if router.extra_attributes.distributed:
+            if router.gw_port and subnets[0]['id']:
+                self.delete_csnat_router_interface_ports(
+                    context.elevated(), router, subnet_id=subnets[0]['id'])
+            plugin = manager.NeutronManager.get_service_plugins().get(
+                        constants.L3_ROUTER_NAT)
+            l3_agents = plugin.get_l3_agents_hosting_routers(context,
+                                                             [router['id']])
+            for l3_agent in l3_agents:
+                if not plugin.check_ports_exist_on_l3agent(context, l3_agent,
+                                                           router['id']):
+                    plugin.remove_router_from_l3_agent(
+                        context, l3_agent['id'], router['id'])
+        router_interface_info = self._make_router_interface_info(
+            router['id'], port['tenant_id'], port['id'], subnets[0]['id'],
+            [subnet['id'] for subnet in subnets])
+        self.notify_router_interface_action(
+            context, router_interface_info, 'remove')
+        return router_interface_info
+
     def remove_router_interface(self, context, router_id, interface_info):
         remove_by_port, remove_by_subnet = (
             self._validate_interface_info(interface_info, for_removal=True)
@@ -331,32 +353,16 @@ class L3_NAT_with_dvr_db_mixin(l3_db.L3_NAT_db_mixin,
         if remove_by_port:
             port, subnets = self._remove_interface_by_port(
                     context, router_id, port_id, subnet_id, device_owner)
+
         # remove_by_subnet is not used here, because the validation logic of
         # _validate_interface_info ensures that at least one of remote_by_*
         # is True.
         else:
             port, subnets = self._remove_interface_by_subnet(
                     context, router_id, subnet_id, device_owner)
-
-        if router.extra_attributes.distributed:
-            if router.gw_port:
-                self.delete_csnat_router_interface_ports(
-                    context.elevated(), router, subnet_id=subnet_id)
-            plugin = manager.NeutronManager.get_service_plugins().get(
-                        constants.L3_ROUTER_NAT)
-            l3_agents = plugin.get_l3_agents_hosting_routers(context,
-                                                             [router_id])
-            for l3_agent in l3_agents:
-                if not plugin.check_ports_exist_on_l3agent(context, l3_agent,
-                                                           router_id):
-                    plugin.remove_router_from_l3_agent(
-                        context, l3_agent['id'], router_id)
-
-        router_interface_info = self._make_router_interface_info(
-            router_id, port['tenant_id'], port['id'], subnets[0]['id'],
-            [subnet['id'] for subnet in subnets])
-        self.notify_router_interface_action(
-            context, router_interface_info, 'remove')
+        router_interface_info = (
+            self._check_dvr_router_remove_required_and_notify_agent(
+                context, router, port, subnets))
         return router_interface_info
 
     def _get_snat_sync_interfaces(self, context, router_ids):

diff --git a/neutron/tests/unit/db/test_l3_dvr_db.py b/neutron/tests/unit/db/test_l3_dvr_db.py
index b10ee8d9c3ddcd6191016b670d9267393f744545..24ec3518a8d616e5c2b05d75d780caee2175b0e3 100644 (file)
--- a/neutron/tests/unit/db/test_l3_dvr_db.py
+++ b/neutron/tests/unit/db/test_l3_dvr_db.py
@@ -19,21 +19,32 @@ from oslo_utils import uuidutils
 from neutron.common import constants as l3_const
 from neutron.common import exceptions
 from neutron import context
+from neutron.db import agents_db
 from neutron.db import common_db_mixin
+from neutron.db import l3_agentschedulers_db
 from neutron.db import l3_dvr_db
 from neutron import manager
 from neutron.plugins.common import constants as plugin_const
-from neutron.tests.unit import testlib_api
+from neutron.tests.unit.db import test_db_base_plugin_v2
 
 _uuid = uuidutils.generate_uuid
 
 
-class L3DvrTestCase(testlib_api.SqlTestCase):
+class FakeL3Plugin(common_db_mixin.CommonDbMixin,
+                   l3_dvr_db.L3_NAT_with_dvr_db_mixin,
+                   l3_agentschedulers_db.L3AgentSchedulerDbMixin,
+                   agents_db.AgentDbMixin):
+    pass
+
+
+class L3DvrTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
 
     def setUp(self):
-        super(L3DvrTestCase, self).setUp()
+        core_plugin = 'neutron.plugins.ml2.plugin.Ml2Plugin'
+        super(L3DvrTestCase, self).setUp(plugin=core_plugin)
+        self.core_plugin = manager.NeutronManager.get_plugin()
         self.ctx = context.get_admin_context()
-        self.mixin = l3_dvr_db.L3_NAT_with_dvr_db_mixin()
+        self.mixin = FakeL3Plugin()
 
     def _create_router(self, router):
         with self.ctx.session.begin(subtransactions=True):
@@ -542,6 +553,55 @@ class L3DvrTestCase(testlib_api.SqlTestCase):
             self.assertTrue(plugin.check_ports_exist_on_l3agent.called)
             self.assertTrue(plugin.remove_router_from_l3_agent.called)
 
+    def test_remove_router_interface_csnat_ports_removal(self):
+        router_dict = {'name': 'test_router', 'admin_state_up': True,
+                       'distributed': True}
+        router = self._create_router(router_dict)
+        with self.network() as net_ext,\
+                self.subnet() as subnet1,\
+                self.subnet(cidr='20.0.0.0/24') as subnet2:
+            ext_net_id = net_ext['network']['id']
+            self.core_plugin.update_network(
+                self.ctx, ext_net_id,
+                {'network': {'router:external': True}})
+            self.mixin.update_router(
+                self.ctx, router['id'],
+                {'router': {'external_gateway_info':
+                            {'network_id': ext_net_id}}})
+            self.mixin.add_router_interface(self.ctx, router['id'],
+                {'subnet_id': subnet1['subnet']['id']})
+            self.mixin.add_router_interface(self.ctx, router['id'],
+                {'subnet_id': subnet2['subnet']['id']})
+
+            csnat_filters = {'device_owner':
+                             [l3_const.DEVICE_OWNER_ROUTER_SNAT]}
+            csnat_ports = self.core_plugin.get_ports(
+                self.ctx, filters=csnat_filters)
+            self.assertEqual(2, len(csnat_ports))
+
+            dvr_filters = {'device_owner':
+                           [l3_const.DEVICE_OWNER_DVR_INTERFACE]}
+            dvr_ports = self.core_plugin.get_ports(
+                self.ctx, filters=dvr_filters)
+            self.assertEqual(2, len(dvr_ports))
+
+            with mock.patch.object(manager.NeutronManager,
+                                  'get_service_plugins') as get_svc_plugin:
+                get_svc_plugin.return_value = {
+                    plugin_const.L3_ROUTER_NAT: self.mixin}
+                self.mixin.remove_router_interface(
+                    self.ctx, router['id'], {'port_id': dvr_ports[0]['id']})
+
+            csnat_ports = self.core_plugin.get_ports(
+                self.ctx, filters=csnat_filters)
+            self.assertEqual(1, len(csnat_ports))
+            self.assertEqual(dvr_ports[1]['fixed_ips'][0]['subnet_id'],
+                             csnat_ports[0]['fixed_ips'][0]['subnet_id'])
+
+            dvr_ports = self.core_plugin.get_ports(
+                self.ctx, filters=dvr_filters)
+            self.assertEqual(1, len(dvr_ports))
+
     def test__validate_router_migration_notify_advanced_services(self):
         router = {'name': 'foo_router', 'admin_state_up': True}
         router_db = self._create_router(router)