Use admin context when updating router gateway

When a router gateway is updated and the subnet changes,
we need to use the admin context in order for the operation
to succeed.  Since the target network has already been
verified to be external, this is OK. The other operations
in this area such as create and delete already do the
same thing.

Added a fullstack test that moves a gateway between subnets
and verifies it's reachable afterwards.

Change-Id: Iead87ba6182d633f0f808032166e02e767fcffae
Closes-bug: #1812118

neutron/db/l3_db.py

@@ -433,8 +433,8 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase,
                                  resource_id=router_id))
 
     def _update_current_gw_port(self, context, router_id, router, ext_ips):
-        self._core_plugin.update_port(context, router.gw_port['id'], {'port':
-                                      {'fixed_ips': ext_ips}})
+        self._core_plugin.update_port(context.elevated(), router.gw_port['id'],
+                                      {'port': {'fixed_ips': ext_ips}})
         context.session.expire(router.gw_port)
 
     def _update_router_gw_info(self, context, router_id, info, router=None):

neutron/tests/fullstack/resources/client.py

@@ -63,13 +63,16 @@ class ClientFixture(fixtures.Fixture):
         return delete(id)
 
     def create_router(self, tenant_id, name=None, ha=False,
-                      external_network=None):
+                      external_network=None, external_subnet=None):
         resource_type = 'router'
 
         name = name or utils.get_rand_name(prefix=resource_type)
         spec = {'tenant_id': tenant_id, 'name': name, 'ha': ha}
         if external_network:
             spec['external_gateway_info'] = {"network_id": external_network}
+            if external_subnet:
+                spec['external_gateway_info']['external_fixed_ips'] = (
+                    [{"subnet_id": external_subnet}])
 
         return self._create_resource(resource_type, spec)
 

neutron/tests/fullstack/test_l3_agent.py

@@ -40,11 +40,15 @@ class TestL3Agent(base.BaseFullStackTestCase):
     def _create_external_network_and_subnet(self, tenant_id):
         network = self.safe_client.create_network(
             tenant_id, name='public', external=True)
+        subnet = self._create_external_subnet(tenant_id, network['id'])
+        return network, subnet
+
+    def _create_external_subnet(self, tenant_id, network_id):
         cidr = self.useFixture(
             ip_network.ExclusiveIPNetwork(
                 "240.0.0.0", "240.255.255.255", "24")).network
-        subnet = self.safe_client.create_subnet(tenant_id, network['id'], cidr)
-        return network, subnet
+        subnet = self.safe_client.create_subnet(tenant_id, network_id, cidr)
+        return subnet
 
     def block_until_port_status_active(self, port_id):
         def is_port_status_active():
@@ -116,6 +120,51 @@ class TestL3Agent(base.BaseFullStackTestCase):
         # ping router old gateway IP, should fail now
         external_vm.block_until_no_ping(old_gw_ip)
 
+    def _test_external_subnet_changed(self):
+        tenant_id = uuidutils.generate_uuid()
+        ext_net, ext_sub = self._create_external_network_and_subnet(tenant_id)
+        external_vm = self._create_external_vm(ext_net, ext_sub)
+
+        router = self.safe_client.create_router(tenant_id,
+                                                external_network=ext_net['id'])
+
+        vm = self._create_net_subnet_and_vm(
+            tenant_id, ['20.0.0.0/24', '2001:db8:aaaa::/64'],
+            self.environment.hosts[1], router)
+        # ping external vm to test snat
+        vm.block_until_ping(external_vm.ip)
+
+        # ping router gateway IP
+        gw_ip = router['external_gateway_info'][
+            'external_fixed_ips'][0]['ip_address']
+        external_vm.block_until_ping(gw_ip)
+
+        # create second external subnet and external vm on it
+        ext_sub_2 = self._create_external_subnet(tenant_id, ext_net['id'])
+        external_vm_2 = self._create_external_vm(ext_net, ext_sub_2)
+
+        # move original router gateway IP to be on second subnet
+        ip_1, ip_2 = self._find_available_ips(ext_net, ext_sub_2, 2)
+        ext_info = {
+            'network_id': ext_net['id'],
+            'external_fixed_ips':
+                [{'ip_address': ip_2, 'subnet_id': ext_sub_2['id']}]}
+        self.safe_client.update_router(router['id'],
+                                       external_gateway_info=ext_info)
+
+        # ping external vm_2 to test snat
+        vm.block_until_ping(external_vm_2.ip)
+
+        # ping router gateway new IP
+        external_vm_2.block_until_ping(ip_2)
+
+        # ping original router old gateway IP, should fail now
+        external_vm.block_until_no_ping(gw_ip)
+
+        # clear the external gateway so ext_sub_2 can be deleted
+        self.safe_client.update_router(router['id'],
+                                       external_gateway_info={})
+
     def _get_namespace(self, router_id, agent=None):
         namespace = namespaces.build_ns_name(namespaces.NS_PREFIX, router_id)
         if agent:
@@ -352,6 +401,9 @@ class TestLegacyL3Agent(TestL3Agent):
     def test_gateway_ip_changed(self):
         self._test_gateway_ip_changed()
 
+    def test_external_subnet_changed(self):
+        self._test_external_subnet_changed()
+
     def test_router_fip_qos_after_admin_state_down_up(self):
         self._router_fip_qos_after_admin_state_down_up()
 
@@ -499,5 +551,8 @@ class TestHAL3Agent(TestL3Agent):
     def test_gateway_ip_changed(self):
         self._test_gateway_ip_changed()
 
+    def test_external_subnet_changed(self):
+        self._test_external_subnet_changed()
+
     def test_router_fip_qos_after_admin_state_down_up(self):
         self._router_fip_qos_after_admin_state_down_up(ha=True)