Merge "[L3][QoS] Remove router QoS binding when GW is removed" into stable/xena

neutron/db/l3_gateway_ip_qos.py

@@ -58,7 +58,7 @@ class L3_gw_ip_qos_dbonly_mixin(l3_gwmode_db.L3_NAT_dbonly_mixin):
 
     def _delete_gw_ip_qos_db(self, context, router_id, policy_id):
         policy = policy_object.QosPolicy.get_policy_obj(context, policy_id)
-        policy.detach_router(router_id)
+        policy.detach_router(router_id, if_exists=True)
 
     def _update_router_gw_info(self, context, router_id, info, router=None):
         # Calls superclass, pass router db object for avoiding re-loading
@@ -66,10 +66,13 @@ class L3_gw_ip_qos_dbonly_mixin(l3_gwmode_db.L3_NAT_dbonly_mixin):
                        self)._update_router_gw_info(
             context, router_id, info, router)
 
+        if not self._is_gw_ip_qos_supported:
+            return router
+
         with db_api.CONTEXT_WRITER.using(context):
-            if self._is_gw_ip_qos_supported and router.gw_port:
-                self._update_router_gw_qos_policy(context, router_id,
-                                                  info, router)
+            if not router.gw_port:
+                info = {qos_consts.QOS_POLICY_ID: None}
+            self._update_router_gw_qos_policy(context, router_id, info, router)
 
         return router
 
@@ -79,7 +82,7 @@ class L3_gw_ip_qos_dbonly_mixin(l3_gwmode_db.L3_NAT_dbonly_mixin):
 
     def _update_router_gw_qos_policy(self, context, router_id, info, router):
         if not info or qos_consts.QOS_POLICY_ID not in info:
-            # An explicit 'None' for `qos_polcy_id` indicates to clear
+            # An explicit 'None' for `qos_policy_id` indicates to clear
             # the router gateway IP policy. So if info does not have
             # the key `qos_polcy_id`, we can not decide what behavior
             # to be done, then directly return here.

neutron/objects/qos/policy.py

@@ -294,10 +294,10 @@ class QosPolicy(rbac_db.NeutronRbacObject):
             raise qos_exc.FloatingIPQosBindingNotFound(fip_id=fip_id,
                                                        policy_id=self.id)
 
-    def detach_router(self, router_id):
+    def detach_router(self, router_id, if_exists=False):
         deleted = binding.QosPolicyRouterGatewayIPBinding.delete_objects(
             self.obj_context, router_id=router_id)
-        if not deleted:
+        if not deleted and not if_exists:
             raise qos_exc.RouterQosBindingNotFound(router_id=router_id,
                                                    policy_id=self.id)
 

neutron/tests/unit/extensions/test_qos_gateway_ip.py

@@ -27,6 +27,7 @@ from oslo_utils import uuidutils
 from neutron.conf.db import extraroute_db
 from neutron.db import l3_gateway_ip_qos
 from neutron.extensions import l3
+from neutron.objects.qos import binding
 from neutron.objects.qos import policy
 from neutron.tests.unit.extensions import test_l3
 
@@ -110,8 +111,7 @@ class GatewayIPQoSDBTestCaseBase(object):
                 res['router']['external_gateway_info'].get(
                     qos_consts.QOS_POLICY_ID))
 
-    def test_clear_router_gateway_and_create_with_old_qos_policy_implicitly(
-            self):
+    def test_clear_router_gateway_and_create_again(self):
         ctx = context.get_admin_context()
         policy_obj = policy.QosPolicy(ctx,
                                       id=uuidutils.generate_uuid(),
@@ -129,22 +129,25 @@ class GatewayIPQoSDBTestCaseBase(object):
                 policy_obj.id,
                 res['router']['external_gateway_info'].get(
                     qos_consts.QOS_POLICY_ID))
+            gw_binding = binding.QosPolicyRouterGatewayIPBinding.get_object(
+                ctx, router_id=r['router']['id'])
+            self.assertEqual(r['router']['id'], gw_binding.router_id)
 
-            # Clear router gateway
+            # Clear router gateway, the QoS policy must be removed.
             self._remove_external_gateway_from_router(
                 r['router']['id'],
                 public_sub['subnet']['network_id'],
                 external_gw_info={})
+            gw_binding = binding.QosPolicyRouterGatewayIPBinding.get_object(
+                ctx, router_id=r['router']['id'])
+            self.assertIsNone(gw_binding)
 
-            # Create router gateway again, then the qos policy binding will be
-            # reused here.
+            # Create router gateway again.
             res = self._add_external_gateway_to_router(
                 r['router']['id'],
                 public_sub['subnet']['network_id'])
-            self.assertEqual(
-                policy_obj.id,
-                res['router']['external_gateway_info'].get(
-                    qos_consts.QOS_POLICY_ID))
+            self.assertIsNone(res['router']['external_gateway_info'].get(
+                qos_consts.QOS_POLICY_ID))
 
     def test_clear_router_gateway_qos_policy(self):
         ctx = context.get_admin_context()