Return 409 Conflict to tenant user deleting port attached to FIP

When a tenant user try to delete a port that has attached a FIP by
an admin user is getting a 500 ServerError.

This patch improves the error to 409 Conflict doing some additionals
checks on the delete_port method.

New exception has been included locally, but will be removed as soon
neutron-lib bumps to a newer release.

Closes-Bug: 2017680
Change-Id: Iab77c64c03fd0d44ff7a3fc1c556d85a8c480bb9

neutron/db/l3_db.py

@@ -83,6 +83,13 @@ FIP_ASSOC_MSG = ('Floating IP %(fip_id)s %(assoc)s. External IP: %(ext_ip)s, '
                  'port: %(port_id)s.')
 
 
+# TODO(froyo): Move this exception to neutron-lib as soon as possible, and when
+# a new release is created and pointed to in the requirements remove this code.
+class FipAssociated(n_exc.InUse):
+    message = _('Unable to complete the operation on port "%(port_id)s" '
+                'because the port still has an associated floating IP.')
+
+
 @registry.has_registry_receivers
 class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
                           base_services.WorkerBase,
@@ -1776,12 +1783,27 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
         @return: set of router-ids that require notification updates
         """
         with db_api.CONTEXT_WRITER.using(context):
+            # NOTE(froyo): Context is elevated to confirm the presence of at
+            # least one FIP associated to the port_id. Additional checks
+            # regarding the tenant's grants will be carried out in following
+            # lines.
             if not l3_obj.FloatingIP.objects_exist(
-                    context, fixed_port_id=port_id):
+                    context.elevated(), fixed_port_id=port_id):
                 return []
 
             floating_ip_objs = l3_obj.FloatingIP.get_objects(
                 context, fixed_port_id=port_id)
+
+            # NOTE(froyo): To ensure that a FIP assigned by an admin user
+            # cannot be disassociated by a tenant user, we raise exception to
+            # generate a 409 Conflict response message that prompts the tenant
+            # user to contact an admin, rather than a 500 error message.
+            if not context.is_admin:
+                floating_ip_objs_admin = l3_obj.FloatingIP.get_objects(
+                    context.elevated(), fixed_port_id=port_id)
+                if floating_ip_objs_admin != floating_ip_objs:
+                    raise FipAssociated(port_id=port_id)
+
             router_ids = {fip.router_id for fip in floating_ip_objs}
             old_fips = {fip.id: self._make_floatingip_dict(fip)
                         for fip in floating_ip_objs}

neutron/tests/unit/db/test_l3_db.py

@@ -307,6 +307,27 @@ class TestL3_NAT_dbonly_mixin(
 
             self.db.prevent_l3_port_deletion(ctx, None)
 
+    @mock.patch.object(l3_obj.FloatingIP, 'objects_exist')
+    @mock.patch.object(l3_obj.FloatingIP, 'get_objects')
+    def test_disassociate_floatingips_conflict_by_fip_attached(self,
+                                                               get_objects,
+                                                               objects_exist):
+        context_tenant = context.Context('tenant', 'tenant', is_admin=False)
+        objects_exist.return_value = True
+        get_objects.side_effect = [
+            [],
+            [{'id': 'floating_ip1', 'port_id': 'port_id'}]]
+        self.assertRaises(l3_db.FipAssociated,
+                          self.db.disassociate_floatingips,
+                          context_tenant,
+                          'port_id')
+        objects_exist.assert_called_once_with(
+            mock.ANY, fixed_port_id='port_id')
+        expected_calls = [
+                mock.call(context_tenant, fixed_port_id='port_id'),
+                mock.call(mock.ANY, fixed_port_id='port_id')]
+        get_objects.assert_has_calls(expected_calls)
+
     @mock.patch.object(directory, 'get_plugin')
     def test_subscribe_address_scope_of_subnetpool(self, gp):
         l3_db.L3RpcNotifierMixin()