Don't fail subnet validation if gw_ip is actually not changed

In subnet update API call Neutron checks if gateway_ip was send to be
updated and if so, it checkes if old gateway_ip isn't already allocated
to some router port. If it's already used, Neutron returns 409 response.
This is valid behaviour but sometimes, some automation tools may do
subnet update request and pass the same gateway ip as already used by
the subnet. In such case, as gateway_ip is actually not changed Neutron
should not raise exception in that validation.

Closes-Bug: #1955121
Change-Id: Iba90b44331fdc63273fd3d19c583a24b5295c0ac
(cherry picked from commit 6809bed632)
(cherry picked from commit 7ab37e92b8)

neutron/db/db_base_plugin_v2.py

@@ -626,17 +626,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

@@ -4821,6 +4821,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: