validate l2p update at plugin level

No driver currently supports updating a PTG's L2P, we should
validate this at the plugin level to avoid code duplication.

Change-Id: Iecd061f1ade16b57e0588ae984a4c570730956e1
Closes-Bug: 1500641

gbpservice/neutron/services/grouppolicy/drivers/resource_mapping.py

@@ -552,19 +552,6 @@ class ResourceMappingDriver(api.PolicyDriver, local_api.LocalAPI,
 
     @log.log
     def update_policy_target_group_precommit(self, context):
-        # REVISIT(rkukura): We could potentially allow updates to
-        # l2_policy_id when no policy targets exist. This would
-        # involve removing each old subnet from the l3_policy's
-        # router, deleting each old subnet, creating a new subnet on
-        # the new l2_policy's network, and adding that subnet to the
-        # l3_policy's router in postcommit. Its also possible that new
-        # subnet[s] would be provided explicitly as part of the
-        # update.
-        old_l2p = context.original['l2_policy_id']
-        new_l2p = context.current['l2_policy_id']
-        if old_l2p and old_l2p != new_l2p:
-            raise exc.L2PolicyUpdateOfPolicyTargetGroupNotSupported()
-
         if set(context.original['subnets']) - set(context.current['subnets']):
             raise exc.PolicyTargetGroupSubnetRemovalNotSupported()
 

gbpservice/neutron/services/grouppolicy/plugin.py

@@ -460,6 +460,19 @@ class GroupPolicyPlugin(group_policy_mapping_db.GroupPolicyMappingDbPlugin):
             updated_policy_target_group = super(
                 GroupPolicyPlugin, self).update_policy_target_group(
                     context, policy_target_group_id, policy_target_group)
+            # REVISIT(rkukura): We could potentially allow updates to
+            # l2_policy_id when no policy targets exist. This would
+            # involve removing each old subnet from the l3_policy's
+            # router, deleting each old subnet, creating a new subnet on
+            # the new l2_policy's network, and adding that subnet to the
+            # l3_policy's router in postcommit. Its also possible that new
+            # subnet[s] would be provided explicitly as part of the
+            # update.
+            old_l2p = original_policy_target_group['l2_policy_id']
+            new_l2p = updated_policy_target_group['l2_policy_id']
+            if old_l2p and old_l2p != new_l2p:
+                raise gp_exc.L2PolicyUpdateOfPolicyTargetGroupNotSupported()
+
             self.extension_manager.process_update_policy_target_group(
                 session, policy_target_group, updated_policy_target_group)
             self._validate_shared_update(

gbpservice/neutron/tests/unit/services/grouppolicy/test_grouppolicy_plugin.py

@@ -654,6 +654,17 @@ class TestPolicyTargetGroup(GroupPolicyPluginTestCase):
         self.assertEqual('ManagementPolicyTargetGroupExists',
                          res['NeutronError']['type'])
 
+    def test_update_l2p_rejectet(self):
+        l2p_1 = self.create_l2_policy()['l2_policy']
+        l2p_2 = self.create_l2_policy()['l2_policy']
+        ptg = self.create_policy_target_group(
+            l2_policy_id=l2p_1['id'],
+            expected_res_status=201)['policy_target_group']
+        res = self.update_policy_target_group(
+            ptg['id'], l2_policy_id=l2p_2['id'], expected_res_status=400)
+        self.assertEqual('L2PolicyUpdateOfPolicyTargetGroupNotSupported',
+                         res['NeutronError']['type'])
+
 
 class TestExternalSegment(GroupPolicyPluginTestCase):
 

gbpservice/neutron/tests/unit/services/grouppolicy/test_implicit_policy.py

@@ -108,39 +108,7 @@ class TestImplicitL2Policy(ImplicitPolicyTestCase):
         res = req.get_response(self.ext_api)
         self.assertEqual(res.status_int, webob.exc.HTTPOk.code)
 
-    def test_update_from_implicit(self):
-        # Create policy_target group with implicit L2 policy.
-        ptg = self.create_policy_target_group()
-        ptg_id = ptg['policy_target_group']['id']
-        l2p1_id = ptg['policy_target_group']['l2_policy_id']
-        req = self.new_show_request('l2_policies', l2p1_id, fmt=self.fmt)
-        l2p1 = self.deserialize(self.fmt, req.get_response(self.ext_api))
-        self.assertEqual(ptg['policy_target_group']['name'],
-                         l2p1['l2_policy']['name'])
-
-        # Update policy_target group to explicit L2 policy.
-        l2p2 = self.create_l2_policy()
-        l2p2_id = l2p2['l2_policy']['id']
-        data = {'policy_target_group': {'l2_policy_id': l2p2_id}}
-        req = self.new_update_request('policy_target_groups', data, ptg_id)
-        ptg = self.deserialize(self.fmt, req.get_response(self.ext_api))
-        self.assertEqual(l2p2_id, ptg['policy_target_group']['l2_policy_id'])
-
-        # Verify old L2 policy was cleaned up.
-        req = self.new_show_request('l2_policies', l2p1_id, fmt=self.fmt)
-        res = req.get_response(self.ext_api)
-        self.assertEqual(res.status_int, webob.exc.HTTPNotFound.code)
-
-        # Verify deleting policy_target group does not cleanup new L2
-        # policy.
-        req = self.new_delete_request('policy_target_groups', ptg_id)
-        res = req.get_response(self.ext_api)
-        self.assertEqual(res.status_int, webob.exc.HTTPNoContent.code)
-        req = self.new_show_request('l2_policies', l2p2_id, fmt=self.fmt)
-        res = req.get_response(self.ext_api)
-        self.assertEqual(res.status_int, webob.exc.HTTPOk.code)
-
-    def test_update_to_implicit(self):
+    def test_delete_from_implicit(self):
         # Create policy_target group with explicit L2 policy.
         l2p1 = self.create_l2_policy()
         l2p1_id = l2p1['l2_policy']['id']
@@ -148,31 +116,14 @@ class TestImplicitL2Policy(ImplicitPolicyTestCase):
         ptg_id = ptg['policy_target_group']['id']
         self.assertEqual(l2p1_id, ptg['policy_target_group']['l2_policy_id'])
 
-        # Update policy_target group to implicit L2 policy.
-        data = {'policy_target_group': {'l2_policy_id': None}}
-        req = self.new_update_request('policy_target_groups', data, ptg_id)
-        ptg = self.deserialize(self.fmt, req.get_response(self.ext_api))
-        l2p2_id = ptg['policy_target_group']['l2_policy_id']
-        self.assertNotEqual(l2p1_id, l2p2_id)
-        self.assertIsNotNone(l2p2_id)
-        req = self.new_show_request('l2_policies', l2p2_id, fmt=self.fmt)
-        l2p2 = self.deserialize(self.fmt, req.get_response(self.ext_api))
-        self.assertEqual(ptg['policy_target_group']['name'],
-                         l2p2['l2_policy']['name'])
+        # Delete PTG
+        self.delete_policy_target_group(ptg_id, expected_res_status=204)
 
         # Verify old L2 policy was not cleaned up.
         req = self.new_show_request('l2_policies', l2p1_id, fmt=self.fmt)
         res = req.get_response(self.ext_api)
         self.assertEqual(res.status_int, webob.exc.HTTPOk.code)
 
-        # Verify deleting policy_target group does cleanup new L2 policy.
-        req = self.new_delete_request('policy_target_groups', ptg_id)
-        res = req.get_response(self.ext_api)
-        self.assertEqual(res.status_int, webob.exc.HTTPNoContent.code)
-        req = self.new_show_request('l2_policies', l2p2_id, fmt=self.fmt)
-        res = req.get_response(self.ext_api)
-        self.assertEqual(res.status_int, webob.exc.HTTPNotFound.code)
-
 
 class TestImplicitL3Policy(ImplicitPolicyTestCase):