Reject deletion of PRS in use by External Policy

Change-Id: I94f4e4ab824bac52a4a4e07390afd75349b995a5
Closes-Bug: #1489090
This commit is contained in:
mageshgv 2015-11-12 19:22:31 +05:30
parent 79cf788b55
commit 155a8271ca
2 changed files with 43 additions and 5 deletions

View File

@ -1047,6 +1047,18 @@ class GroupPolicyDbPlugin(gpolicy.GroupPolicyPluginBase,
context.session.query(PTGToPRSConsumingAssociation).filter_by(
policy_rule_set_id=policy_rule_set_id)]
def _get_eps_for_providing_policy_rule_set(self, context,
policy_rule_set_id):
return [x['external_policy_id'] for x in
context.session.query(EPToPRSProvidingAssociation).filter_by(
policy_rule_set_id=policy_rule_set_id)]
def _get_eps_for_consuming_policy_rule_set(self, context,
policy_rule_set_id):
return [x['external_policy_id'] for x in
context.session.query(EPToPRSConsumingAssociation).filter_by(
policy_rule_set_id=policy_rule_set_id)]
def _get_policy_rule_policy_rule_sets(self, context, policy_rule_id):
return [x['policy_rule_set_id'] for x in
context.session.query(PRSToPRAssociation).filter_by(
@ -1656,10 +1668,16 @@ class GroupPolicyDbPlugin(gpolicy.GroupPolicyPluginBase,
def delete_policy_rule_set(self, context, policy_rule_set_id):
with context.session.begin(subtransactions=True):
prs_db = self._get_policy_rule_set(context, policy_rule_set_id)
prs_ids = self._get_ptgs_for_providing_policy_rule_set(
context, policy_rule_set_id) or (
self._get_ptgs_for_consuming_policy_rule_set(
context, policy_rule_set_id))
prs_ids = (
self._get_ptgs_for_providing_policy_rule_set(
context, policy_rule_set_id) or
self._get_ptgs_for_consuming_policy_rule_set(
context, policy_rule_set_id) or
self._get_eps_for_providing_policy_rule_set(
context, policy_rule_set_id) or
self._get_eps_for_consuming_policy_rule_set(
context, policy_rule_set_id))
if prs_ids:
raise gpolicy.PolicyRuleSetInUse(policy_rule_set_id=
policy_rule_set_id)

View File

@ -1175,7 +1175,7 @@ class TestGroupResources(GroupPolicyDbTestCase):
self.assertRaises(gpolicy.PolicyRuleSetNotFound,
self.plugin.get_policy_rule_set, ctx, prs_id)
def test_delete_policy_rule_set_in_use(self):
def test_delete_policy_rule_set_in_use_by_ptg(self):
ctx = context.get_admin_context()
l3p = self.create_l3_policy()
l3p_id = l3p['l3_policy']['id']
@ -1201,6 +1201,26 @@ class TestGroupResources(GroupPolicyDbTestCase):
self.plugin.delete_policy_rule_set, ctx,
consumed_prs_id)
def test_delete_policy_rule_set_in_use_by_ep(self):
ctx = context.get_admin_context()
provided_prs_id = (
self.create_policy_rule_set()['policy_rule_set']['id'])
consumed_prs_id = (
self.create_policy_rule_set()['policy_rule_set']['id'])
self.create_external_policy(
provided_policy_rule_sets={provided_prs_id: None},
consumed_policy_rule_sets={consumed_prs_id: None})
self.assertRaises(gpolicy.PolicyRuleSetInUse,
self.plugin.delete_policy_rule_set, ctx,
provided_prs_id)
self.assertRaises(gpolicy.PolicyRuleSetInUse,
self.plugin.delete_policy_rule_set, ctx,
consumed_prs_id)
def test_prs_one_hierarchy_children(self):
child = self.create_policy_rule_set()['policy_rule_set']
parent = self.create_policy_rule_set(