From 780d81ec00abada9afb01546cb0c976562a502a6 Mon Sep 17 00:00:00 2001
From: Rodolfo Alonso Hernandez <ralonsoh@redhat.com>
Date: Sun, 14 Jan 2024 10:02:13 +0000
Subject: [PATCH] Add router check, subnet attached gateway IP update or
 deletion

Added a new test to ``RoutersTest``. If a subnet has a router
interface, the subnet gateway IP cannot be modified or deleted.
Both operations will raise a ``GatewayIpInUse`` exception.

Depends-On: https://review.opendev.org/c/openstack/neutron/+/904713

Related-Bug: #2036423
Change-Id: I46a39c53017589e23e03ceabc45c2f144ca2f3bb
---
 neutron_tempest_plugin/api/base.py         | 14 +++++++++-
 neutron_tempest_plugin/api/test_routers.py | 32 ++++++++++++++++++++++
 2 files changed, 45 insertions(+), 1 deletion(-)

diff --git a/neutron_tempest_plugin/api/base.py b/neutron_tempest_plugin/api/base.py
index 28d556a9..7f961685 100644
--- a/neutron_tempest_plugin/api/base.py
+++ b/neutron_tempest_plugin/api/base.py
@@ -416,7 +416,7 @@ class BaseNetworkTest(test.BaseTestCase):
     @classmethod
     def create_subnet(cls, network, gateway='', cidr=None, mask_bits=None,
                       ip_version=None, client=None, reserve_cidr=True,
-                      **kwargs):
+                      allocation_pool_size=None, **kwargs):
         """Wrapper utility that returns a test subnet.
 
         Convenient wrapper for client.create_subnet method. It reserves and
@@ -454,10 +454,19 @@ class BaseNetworkTest(test.BaseTestCase):
         using the same CIDR for further subnets in the scope of the same
         test case class
 
+        :param allocation_pool_size: if the CIDR is not defined, this method
+        will assign one in ``get_subnet_cidrs``. Once done, the allocation pool
+        will be defined reserving the number of IP addresses requested,
+        starting from the end of the assigned CIDR.
+
         :param **kwargs: optional parameters to be forwarded to wrapped method
 
         [1] http://netaddr.readthedocs.io/en/latest/tutorial_01.html#supernets-and-subnets  # noqa
         """
+        def allocation_pool(cidr, pool_size):
+            start = str(netaddr.IPAddress(cidr.last) - pool_size)
+            end = str(netaddr.IPAddress(cidr.last) - 1)
+            return {'start': start, 'end': end}
 
         # allow tests to use admin client
         if not client:
@@ -480,6 +489,9 @@ class BaseNetworkTest(test.BaseTestCase):
                 kwargs['gateway_ip'] = str(gateway or (subnet_cidr.ip + 1))
             else:
                 kwargs['gateway_ip'] = None
+            if allocation_pool_size:
+                kwargs['allocation_pools'] = [
+                    allocation_pool(subnet_cidr, allocation_pool_size)]
             try:
                 body = client.create_subnet(
                     network_id=network['id'],
diff --git a/neutron_tempest_plugin/api/test_routers.py b/neutron_tempest_plugin/api/test_routers.py
index 4179e6d4..ff5d3916 100644
--- a/neutron_tempest_plugin/api/test_routers.py
+++ b/neutron_tempest_plugin/api/test_routers.py
@@ -293,6 +293,38 @@ class RoutersTest(base_routers.BaseRouterTest):
         self.assertEqual(port_show1['device_id'], router1['id'])
         self.assertEqual(port_show2['device_id'], router2['id'])
 
+    @decorators.idempotent_id('4f8a2a1e-7fe9-4d99-9bff-5dc0e78b7e06')
+    def test_router_interface_update_and_remove_gateway_ip(self):
+        network = self.create_network()
+        subnet = self.create_subnet(network, allocation_pool_size=5)
+
+        # Update the subnet gateway IP, using the next one. Because the
+        # allocation pool is on the upper part of the CIDR, the lower IP
+        # addresses are free. This operation must be allowed because the subnet
+        # does not have yet a router port.
+        gateway_ip = netaddr.IPAddress(subnet['gateway_ip'])
+        self.client.update_subnet(subnet['id'], gateway_ip=str(gateway_ip + 1))
+
+        router = self._create_router(data_utils.rand_name('router'), True)
+        intf = self.create_router_interface(router['id'], subnet['id'])
+
+        def _status_active():
+            return self.client.show_port(
+                intf['port_id'])['port']['status'] == 'ACTIVE'
+
+        utils.wait_until_true(_status_active, exception=AssertionError)
+
+        # The gateway update must raise a ``GatewayIpInUse`` exception because
+        # there is an allocated router port.
+        gateway_ip = netaddr.IPAddress(subnet['gateway_ip'])
+        self.assertRaises(lib_exc.Conflict, self.client.update_subnet,
+                          subnet['id'], gateway_ip=str(gateway_ip + 2))
+
+        # The gateway deletion returns the same exception.
+        gateway_ip = netaddr.IPAddress(subnet['gateway_ip'])
+        self.assertRaises(lib_exc.Conflict, self.client.update_subnet,
+                          subnet['id'], gateway_ip=None)
+
 
 class RoutersIpV6Test(RoutersTest):
     _ip_version = 6