test/test_ip6.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import socket
   5
   6 from framework import VppTestCase, VppTestRunner
   7 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
   8
   9 from scapy.packet import Raw
  10 from scapy.layers.l2 import Ether, Dot1Q
  11 from scapy.layers.inet6 import IPv6, UDP
  12 from util import ppp
  13
  14
  15 class TestIPv6(VppTestCase):
  16     """ IPv6 Test Case """
  17
  18     @classmethod
  19     def setUpClass(cls):
  20         super(TestIPv6, cls).setUpClass()
  21
  22     def setUp(self):
  23         """
  24         Perform test setup before test case.
  25
  26         **Config:**
  27             - create 3 pg interfaces
  28                 - untagged pg0 interface
  29                 - Dot1Q subinterface on pg1
  30                 - Dot1AD subinterface on pg2
  31             - setup interfaces:
  32                 - put it into UP state
  33                 - set IPv6 addresses
  34                 - resolve neighbor address using NDP
  35             - configure 200 fib entries
  36
  37         :ivar list interfaces: pg interfaces and subinterfaces.
  38         :ivar dict flows: IPv4 packet flows in test.
  39         :ivar list pg_if_packet_sizes: packet sizes in test.
  40
  41         *TODO:* Create AD sub interface
  42         """
  43         super(TestIPv6, self).setUp()
  44
  45         # create 3 pg interfaces
  46         self.create_pg_interfaces(range(3))
  47
  48         # create 2 subinterfaces for p1 and pg2
  49         self.sub_interfaces = [
  50             VppDot1QSubint(self, self.pg1, 100),
  51             VppDot1QSubint(self, self.pg2, 200)
  52             # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
  53         ]
  54
  55         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
  56         self.flows = dict()
  57         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
  58         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
  59         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
  60
  61         # packet sizes
  62         self.pg_if_packet_sizes = [64, 512, 1518, 9018]
  63         self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
  64
  65         self.interfaces = list(self.pg_interfaces)
  66         self.interfaces.extend(self.sub_interfaces)
  67
  68         # setup all interfaces
  69         for i in self.interfaces:
  70             i.admin_up()
  71             i.config_ip6()
  72             i.resolve_ndp()
  73
  74         # config 2M FIB entries
  75         self.config_fib_entries(200)
  76
  77     def tearDown(self):
  78         """Run standard test teardown and log ``show ip6 neighbors``."""
  79         super(TestIPv6, self).tearDown()
  80         if not self.vpp_dead:
  81             self.logger.info(self.vapi.cli("show ip6 neighbors"))
  82             # info(self.vapi.cli("show ip6 fib"))  # many entries
  83
  84     def config_fib_entries(self, count):
  85         """For each interface add to the FIB table *count* routes to
  86         "fd02::1/128" destination with interface's local address as next-hop
  87         address.
  88
  89         :param int count: Number of FIB entries.
  90
  91         - *TODO:* check if the next-hop address shouldn't be remote address
  92           instead of local address.
  93         """
  94         n_int = len(self.interfaces)
  95         percent = 0
  96         counter = 0.0
  97         dest_addr = socket.inet_pton(socket.AF_INET6, "fd02::1")
  98         dest_addr_len = 128
  99         for i in self.interfaces:
 100             next_hop_address = i.local_ip6n
 101             for j in range(count / n_int):
 102                 self.vapi.ip_add_del_route(
 103                     dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
 104                 counter += 1
 105                 if counter / count * 100 > percent:
 106                     self.logger.info("Configure %d FIB entries .. %d%% done" %
 107                                      (count, percent))
 108                     percent += 1
 109
 110     def create_stream(self, src_if, packet_sizes):
 111         """Create input packet stream for defined interface.
 112
 113         :param VppInterface src_if: Interface to create packet stream for.
 114         :param list packet_sizes: Required packet sizes.
 115         """
 116         pkts = []
 117         for i in range(0, 257):
 118             dst_if = self.flows[src_if][i % 2]
 119             info = self.create_packet_info(
 120                 src_if.sw_if_index, dst_if.sw_if_index)
 121             payload = self.info_to_payload(info)
 122             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 123                  IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
 124                  UDP(sport=1234, dport=1234) /
 125                  Raw(payload))
 126             info.data = p.copy()
 127             if isinstance(src_if, VppSubInterface):
 128                 p = src_if.add_dot1_layer(p)
 129             size = packet_sizes[(i // 2) % len(packet_sizes)]
 130             self.extend_packet(p, size)
 131             pkts.append(p)
 132         return pkts
 133
 134     def verify_capture(self, dst_if, capture):
 135         """Verify captured input packet stream for defined interface.
 136
 137         :param VppInterface dst_if: Interface to verify captured packet stream
 138                                     for.
 139         :param list capture: Captured packet stream.
 140         """
 141         self.logger.info("Verifying capture on interface %s" % dst_if.name)
 142         last_info = dict()
 143         for i in self.interfaces:
 144             last_info[i.sw_if_index] = None
 145         is_sub_if = False
 146         dst_sw_if_index = dst_if.sw_if_index
 147         if hasattr(dst_if, 'parent'):
 148             is_sub_if = True
 149         for packet in capture:
 150             if is_sub_if:
 151                 # Check VLAN tags and Ethernet header
 152                 packet = dst_if.remove_dot1_layer(packet)
 153             self.assertTrue(Dot1Q not in packet)
 154             try:
 155                 ip = packet[IPv6]
 156                 udp = packet[UDP]
 157                 payload_info = self.payload_to_info(str(packet[Raw]))
 158                 packet_index = payload_info.index
 159                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 160                 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
 161                                   (dst_if.name, payload_info.src, packet_index))
 162                 next_info = self.get_next_packet_info_for_interface2(
 163                     payload_info.src, dst_sw_if_index,
 164                     last_info[payload_info.src])
 165                 last_info[payload_info.src] = next_info
 166                 self.assertTrue(next_info is not None)
 167                 self.assertEqual(packet_index, next_info.index)
 168                 saved_packet = next_info.data
 169                 # Check standard fields
 170                 self.assertEqual(ip.src, saved_packet[IPv6].src)
 171                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
 172                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
 173                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
 174             except:
 175                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
 176                 raise
 177         for i in self.interfaces:
 178             remaining_packet = self.get_next_packet_info_for_interface2(
 179                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 180             self.assertTrue(remaining_packet is None,
 181                             "Interface %s: Packet expected from interface %s "
 182                             "didn't arrive" % (dst_if.name, i.name))
 183
 184     def test_fib(self):
 185         """ IPv6 FIB test
 186
 187         Test scenario:
 188             - Create IPv6 stream for pg0 interface
 189             - Create IPv6 tagged streams for pg1's and pg2's subinterface.
 190             - Send and verify received packets on each interface.
 191         """
 192
 193         pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
 194         self.pg0.add_stream(pkts)
 195
 196         for i in self.sub_interfaces:
 197             pkts = self.create_stream(i, self.sub_if_packet_sizes)
 198             i.parent.add_stream(pkts)
 199
 200         self.pg_enable_capture(self.pg_interfaces)
 201         self.pg_start()
 202
 203         pkts = self.pg0.get_capture()
 204         self.verify_capture(self.pg0, pkts)
 205
 206         for i in self.sub_interfaces:
 207             pkts = i.parent.get_capture()
 208             self.verify_capture(i, pkts)
 209
 210
 211 if __name__ == '__main__':
 212     unittest.main(testRunner=VppTestRunner)