Forbid deleting a policy in use

This patch added a check before policy deleting to prevent user
from deleting a policy that is still attached to cluster(s).

Change-Id: I695fa9d6501b4f7afba6ca27ed546612193843ac

senlin/common/exception.py

@@ -191,6 +191,10 @@ class PolicyExists(SenlinException):
     msg_fmt = _("The policy type (%(policy_type)s) already exists.")
 
 
+class PolicyInUse(SenlinException):
+    msg_fmt = _("The policy (%(policy)s) is still attached to cluster(s).")
+
+
 class PolicyNotAttached(SenlinException):
     msg_fmt = _("The policy (%(policy)s) is not attached to the specified "
                 "cluster (%(cluster)s).")

senlin/db/sqlalchemy/api.py

@@ -709,10 +709,13 @@ def policy_delete(context, policy_id, force=False):
     if not policy:
         return
 
+    query = model_query(context, models.ClusterPolicies)
+    bindings = query.filter_by(policy_id=policy_id)
+    if bindings.count():
+        raise exception.PolicyInUse(policy=policy_id)
+
     session = orm_session.Session.object_session(policy)
 
-    # TODO(Qiming): Check if a policy is still in use, raise an exception
-    # if so
     policy.soft_delete(session=session)
     session.flush()
 

senlin/tests/db/test_policy_api.py

@@ -16,6 +16,7 @@ from senlin.common import exception
 from senlin.db.sqlalchemy import api as db_api
 from senlin.tests.common import base
 from senlin.tests.common import utils
+from senlin.tests.db import shared
 
 sample_spec = {
     'min_size': 1,
@@ -28,6 +29,8 @@ class DBAPIPolicyTest(base.SenlinTestCase):
     def setUp(self):
         super(DBAPIPolicyTest, self).setUp()
         self.ctx = utils.dummy_context()
+        self.profile = shared.create_profile(self.ctx)
+        self.cluster = shared.create_cluster(self.ctx, self.profile)
 
     def new_policy_data(self, **kwargs):
         data = {
@@ -368,3 +371,22 @@ class DBAPIPolicyTest(base.SenlinTestCase):
         # not found in delete is okay
         res = db_api.policy_delete(self.ctx, policy_id)
         self.assertIsNone(res)
+
+    def test_policy_delete_in_use(self):
+        policy = db_api.policy_create(self.ctx, self.new_policy_data())
+        self.assertIsNotNone(policy)
+
+        fields = {
+            'enabled': True,
+            'level': 50
+        }
+        db_api.cluster_policy_attach(self.ctx, self.cluster.id, policy.id,
+                                     fields)
+        self.assertRaises(exception.PolicyInUse,
+                          db_api.policy_delete,
+                          self.ctx, policy.id)
+
+        db_api.cluster_policy_detach(self.ctx, self.cluster.id, policy.id)
+        db_api.policy_delete(self.ctx, policy.id)
+        policy = db_api.policy_get(self.ctx, policy.id)
+        self.assertIsNone(policy)