Merge "fullstack: test for IPv6 north-south traffic"

neutron/tests/common/machine_fixtures.py

@@ -100,10 +100,11 @@ class FakeMachineBase(fixtures.Fixture):
 
 class FakeMachine(FakeMachineBase):
 
-    def __init__(self, bridge, ip_cidr, gateway_ip=None):
+    def __init__(self, bridge, ip_cidr, gateway_ip=None, ipv6_cidr=None):
         super(FakeMachine, self).__init__()
         self.bridge = bridge
         self._ip_cidr = ip_cidr
+        self._ipv6_cidr = ipv6_cidr
         self.gateway_ip = gateway_ip
 
     def _setUp(self):
@@ -130,6 +131,21 @@ class FakeMachine(FakeMachineBase):
         self.port.addr.delete(self._ip_cidr)
         self._ip_cidr = ip_cidr
 
+    @property
+    def ipv6(self):
+        return self._ipv6_cidr.partition('/')[0]
+
+    @property
+    def ipv6_cidr(self):
+        return self._ipv6_cidr
+
+    @ipv6_cidr.setter
+    def ipv6_cidr(self, ipv6_cidr):
+        if self._ipv6_cidr:
+            self.port.addr.delete(self._ipv6_cidr)
+        self.port.addr.add(ipv6_cidr)
+        self._ipv6_cidr = ipv6_cidr
+
     @FakeMachineBase.mac_address.setter
     def mac_address(self, mac_address):
         self.port.link.set_down()

neutron/tests/fullstack/resources/machine.py

@@ -98,14 +98,15 @@ class FakeFullstackMachine(machine_fixtures.FakeMachineBase):
         return new_bridge
 
     def _configure_ipaddress(self, fixed_ip):
+        subnet_id = fixed_ip['subnet_id']
+        subnet = self.safe_client.client.show_subnet(subnet_id)
         if (netaddr.IPAddress(fixed_ip['ip_address']).version ==
             constants.IP_VERSION_6):
             # v6Address/default_route is auto-configured.
             self._ipv6 = fixed_ip['ip_address']
+            self.gateway_ipv6 = subnet['subnet']['gateway_ip']
         else:
             self._ip = fixed_ip['ip_address']
-            subnet_id = fixed_ip['subnet_id']
-            subnet = self.safe_client.client.show_subnet(subnet_id)
             prefixlen = netaddr.IPNetwork(subnet['subnet']['cidr']).prefixlen
             self._ip_cidr = '%s/%s' % (self._ip, prefixlen)
 

neutron/tests/fullstack/test_l3_agent.py

@@ -120,23 +120,45 @@ class TestLegacyL3Agent(TestL3Agent):
         # Verify ping6 from vm2 to vm1 IPv6 Address
         vm2.block_until_ping(vm1.ipv6)
 
-    def test_snat_and_floatingip(self):
+    def test_north_south_traffic(self):
-        # This function creates external network and boots an extrenal vm
+        # This function creates an external network which is connected to
-        # on it with gateway ip and connected to central_external_bridge.
+        # central_external_bridge and spawns an external_vm on it.
-        # Later it creates a tenant vm on tenant network, with tenant router
+        # The external_vm is configured with the gateway_ip (both v4 & v6
-        # connected to tenant network and external network.
+        # addresses) of external subnet. Later, it creates a tenant router,
-        # To test snat and floatingip, try ping between tenant and external vms
+        # a tenant network and two tenant subnets (v4 and v6). The tenant
+        # router is associated with tenant network and external network to
+        # provide north-south connectivity to the VMs.
+        # We validate the following in this testcase.
+        # 1. SNAT support: using ping from tenant VM to external_vm
+        # 2. Floating IP support: using ping from external_vm to VM floating ip
+        # 3. IPv6 ext connectivity: using ping6 from tenant vm to external_vm.
         tenant_id = uuidutils.generate_uuid()
         ext_net, ext_sub = self._create_external_network_and_subnet(tenant_id)
         external_vm = self.useFixture(
             machine_fixtures.FakeMachine(
                 self.environment.central_external_bridge,
                 common_utils.ip_to_cidr(ext_sub['gateway_ip'], 24)))
+        # Create an IPv6 subnet in the external network
+        v6network = self.useFixture(
+            ip_network.ExclusiveIPNetwork(
+                "2001:db8:1234::1", "2001:db8:1234::10", "64")).network
+        ext_v6sub = self.safe_client.create_subnet(
+            tenant_id, ext_net['id'], v6network)
 
         router = self.safe_client.create_router(tenant_id,
                                                 external_network=ext_net['id'])
+
+        # Configure the gateway_ip of external v6subnet on the external_vm.
+        external_vm.ipv6_cidr = common_utils.ip_to_cidr(
+            ext_v6sub['gateway_ip'], 64)
+
+        # Configure an IPv6 downstream route to the v6Address of router gw port
+        for fixed_ip in router['external_gateway_info']['external_fixed_ips']:
+            if netaddr.IPNetwork(fixed_ip['ip_address']).version == 6:
+                external_vm.set_default_gateway(fixed_ip['ip_address'])
+
         vm = self._create_net_subnet_and_vm(
-            tenant_id, ['20.0.0.0/24'],
+            tenant_id, ['20.0.0.0/24', '2001:db8:aaaa::/64'],
             self.environment.hosts[1], router)
 
         # ping external vm to test snat
@@ -148,6 +170,11 @@ class TestLegacyL3Agent(TestL3Agent):
         # ping floating ip from external vm
         external_vm.block_until_ping(fip['floating_ip_address'])
 
+        # Verify VM is able to reach the router interface.
+        vm.block_until_ping(vm.gateway_ipv6)
+        # Verify north-south connectivity using ping6 to external_vm.
+        vm.block_until_ping(external_vm.ipv6)
+
 
 class TestHAL3Agent(base.BaseFullStackTestCase):