Detect DBReferenceError when deleting flavor

The flavor framework currently has a TODO where the logic
to ensure the flavor isn't in use should be. Implementing
this logic will be a bit complicated this late in the cycle
since many different services can depend on the flavor.

For now we can at least catch the DBReferenceError when trying
to delete the flavor and convert it into the FlavorInUse
exception.

This leaves some notes inline about how we might go about
implemented the _ensure_flavor_not_inuse function.

Change-Id: I6bfe61645c6cee002020a507e489c3535d5026ab
Closes-Bug: #1621281

neutron/db/flavors_db.py

@@ -13,6 +13,7 @@
 #    under the License.
 
 from neutron_lib.db import model_base
+from oslo_db import exception as db_exc
 from oslo_log import log as logging
 from oslo_utils import uuidutils
 import sqlalchemy as sa
@@ -104,6 +105,10 @@ class FlavorsDbMixin(common_db_mixin.CommonDbMixin):
         # Future TODO(enikanorov): check that there is no binding to
         # instances. Shall address in future upon getting the right
         # flavor supported driver
+        # NOTE(kevinbenton): sqlalchemy utils has a cool dependent
+        # objects function we can use to quickly query all tables
+        # that have a foreign key ref to flavors. Or we could replace
+        # the call to this with callback events.
         pass
 
     def _ensure_service_profile_not_in_use(self, context, sp_id):
@@ -147,10 +152,17 @@ class FlavorsDbMixin(common_db_mixin.CommonDbMixin):
         return self._make_flavor_dict(fl, fields)
 
     def delete_flavor(self, context, flavor_id):
-        with context.session.begin(subtransactions=True):
-            self._ensure_flavor_not_in_use(context, flavor_id)
-            fl_db = self._get_flavor(context, flavor_id)
-            context.session.delete(fl_db)
+        # NOTE(kevinbenton): we need to fix _ensure_flavor_not_in_use,
+        # but the fix is non-trivial since multiple services can use
+        # flavors so for now we just capture the foreign key violation
+        # to detect if it's in use.
+        try:
+            with context.session.begin(subtransactions=True):
+                self._ensure_flavor_not_in_use(context, flavor_id)
+                fl_db = self._get_flavor(context, flavor_id)
+                context.session.delete(fl_db)
+        except db_exc.DBReferenceError:
+            raise ext_flavors.FlavorInUse(flavor_id=flavor_id)
 
     def get_flavors(self, context, filters=None, fields=None,
                     sorts=None, limit=None, marker=None, page_reverse=False):

neutron/tests/tempest/api/admin/test_routers_flavors.py

@@ -11,7 +11,9 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from tempest.lib import exceptions as lib_exc
 from tempest import test
+import testtools
 
 from neutron.tests.tempest.api import base_routers as base
 
@@ -63,3 +65,9 @@ class RoutersFlavorTestCase(base.BaseRouterTest):
         # ensure client can create router with flavor
         router = self.create_router('name', flavor_id=flavor['id'])
         self.assertEqual(flavor['id'], router['flavor_id'])
+
+    @test.idempotent_id('30e73858-a0fc-409c-a2e0-e9cd2826f6a2')
+    def test_delete_router_flavor_in_use(self):
+        self.create_router('name', flavor_id=self.flavor['id'])
+        with testtools.ExpectedException(lib_exc.Conflict):
+            self.admin_client.delete_flavor(self.flavor['id'])

neutron/tests/unit/extensions/test_flavors.py

@@ -25,6 +25,7 @@ from neutron.api.v2 import attributes as attr
 from neutron import context
 from neutron.db import api as dbapi
 from neutron.db import flavors_db
+from neutron.db import l3_db
 from neutron.db import servicetype_db
 from neutron.extensions import flavors
 from neutron.plugins.common import constants
@@ -661,6 +662,17 @@ class FlavorPluginTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
             self.ctx,
             sp['id'])
 
+    def test_delete_flavor_in_use(self):
+        # make use of router since it has a flavor id
+        fl, data = self._create_flavor()
+        with self.ctx.session.begin():
+            self.ctx.session.add(l3_db.Router(flavor_id=fl['id']))
+        self.assertRaises(
+            flavors.FlavorInUse,
+            self.plugin.delete_flavor,
+            self.ctx,
+            fl['id'])
+
     def test_get_flavor_next_provider_no_binding(self):
         fl, data = self._create_flavor()
         self.assertRaises(