Merge "Don't fail subnet validation if gw_ip is actually not changed" into stable/victoria

neutron/db/db_base_plugin_v2.py

@@ -628,17 +628,20 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
             # NOTE(salv-orlando): There is slight chance of a race, when
             # a subnet-update and a router-interface-add operation are
             # executed concurrently
-            if cur_subnet and not ipv6_utils.is_ipv6_pd_enabled(s):
+            s_gateway_ip = netaddr.IPAddress(s['gateway_ip'])
+            if (cur_subnet and
+                    s_gateway_ip != cur_subnet['gateway_ip'] and
+                    not ipv6_utils.is_ipv6_pd_enabled(s)):
                 gateway_ip = str(cur_subnet['gateway_ip'])
                 with db_api.CONTEXT_READER.using(context):
-                    allocated = port_obj.IPAllocation.get_alloc_routerports(
+                    alloc = port_obj.IPAllocation.get_alloc_routerports(
                         context, cur_subnet['id'], gateway_ip=gateway_ip,
                         first=True)
 
-                if allocated and allocated.port_id:
+                if alloc and alloc.port_id:
                     raise exc.GatewayIpInUse(
                         ip_address=gateway_ip,
-                        port_id=allocated.port_id)
+                        port_id=alloc.port_id)
 
         if validators.is_attr_set(s.get('dns_nameservers')):
             if len(s['dns_nameservers']) > cfg.CONF.max_dns_nameservers:

neutron/tests/unit/db/test_db_base_plugin_v2.py

@@ -4812,6 +4812,70 @@ class TestSubnetsV2(NeutronDbPluginV2TestCase):
                     res = req.get_response(self.api)
                     self.assertEqual(res.status_int, 200)
 
+    def test_update_subnet_the_same_gw_as_in_use_by_router(self):
+        with self.network() as network:
+            with self.subnet(network=network,
+                             allocation_pools=[{'start': '10.0.0.2',
+                                                'end': '10.0.0.8'}]) as subnet:
+                s = subnet['subnet']
+                with self.port(
+                    subnet=subnet, fixed_ips=[{'subnet_id': s['id'],
+                                               'ip_address': s['gateway_ip']}]
+                ) as port:
+                    # this protection only applies to router ports so we need
+                    # to make this port belong to a router
+                    ctx = context.get_admin_context()
+                    with db_api.CONTEXT_WRITER.using(ctx):
+                        router = l3_models.Router()
+                        ctx.session.add(router)
+                    rp = l3_obj.RouterPort(ctx, router_id=router.id,
+                                           port_id=port['port']['id'])
+                    rp.create()
+
+                    # update subnet will be with the same gateway_ip as was
+                    # used before, thus it should be fine
+                    data = {'subnet': {
+                        'gateway_ip': s['gateway_ip'],
+                        'description': 'test update subnet'}}
+                    req = self.new_update_request('subnets', data,
+                                                  s['id'])
+                    res = req.get_response(self.api)
+                    self.assertEqual(200, res.status_int)
+
+    def test_update_subnet_the_same_gw_as_in_use_by_router_ipv6(self):
+        with self.network() as network:
+            with self.subnet(network=network,
+                             ip_version=constants.IP_VERSION_6,
+                             cidr="fe80::/48") as subnet:
+                s = subnet['subnet']
+                with self.port(
+                    subnet=subnet, fixed_ips=[{'subnet_id': s['id'],
+                                               'ip_address': s['gateway_ip']}]
+                ) as port:
+                    # this protection only applies to router ports so we need
+                    # to make this port belong to a router
+                    ctx = context.get_admin_context()
+                    with db_api.CONTEXT_WRITER.using(ctx):
+                        router = l3_models.Router()
+                        ctx.session.add(router)
+                    rp = l3_obj.RouterPort(ctx, router_id=router.id,
+                                           port_id=port['port']['id'])
+                    rp.create()
+
+                    # It's the same IP address but with all zeros now so string
+                    # is different
+                    new_gw_ip = netaddr.IPAddress(s['gateway_ip']).format(
+                            dialect=netaddr.ipv6_verbose)
+                    # update subnet will be with the same gateway_ip as was
+                    # used before, thus it should be fine
+                    data = {'subnet': {
+                        'gateway_ip': new_gw_ip,
+                        'description': 'test update subnet'}}
+                    req = self.new_update_request('subnets', data,
+                                                  s['id'])
+                    res = req.get_response(self.api)
+                    self.assertEqual(200, res.status_int)
+
     def test_update_subnet_invalid_gw_V4_cidr(self):
         with self.network() as network:
             with self.subnet(network=network) as subnet: