test/test_ip6.py

   1 #!/usr/bin/env python
   2
   3 import unittest
   4 import socket
   5 from logging import *
   6
   7 from framework import VppTestCase, VppTestRunner
   8 from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
   9
  10 from scapy.packet import Raw
  11 from scapy.layers.l2 import Ether, Dot1Q
  12 from scapy.layers.inet6 import ICMPv6ND_NS, IPv6, UDP
  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         super(TestIPv6, self).setUp()
  24
  25         # create 3 pg interfaces
  26         self.create_pg_interfaces(range(3))
  27
  28         # create 2 subinterfaces for p1 and pg2
  29         self.sub_interfaces = [
  30             VppDot1QSubint(self, self.pg1, 100),
  31             VppDot1QSubint(self, self.pg2, 200)]
  32             # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
  33
  34         # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
  35         self.flows = dict()
  36         self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
  37         self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
  38         self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
  39
  40         # packet sizes
  41         self.pg_if_packet_sizes = [64, 512, 1518, 9018]
  42         self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
  43
  44         self.interfaces = list(self.pg_interfaces)
  45         self.interfaces.extend(self.sub_interfaces)
  46
  47         # setup all interfaces
  48         for i in self.interfaces:
  49             i.admin_up()
  50             i.config_ip6()
  51             i.resolve_ndp()
  52
  53         # config 2M FIB enries
  54         self.config_fib_entries(200)
  55
  56     def tearDown(self):
  57         super(TestIPv6, self).tearDown()
  58         if not self.vpp_dead:
  59             info(self.vapi.cli("show ip6 neighbors"))
  60             # info(self.vapi.cli("show ip6 fib"))  # many entries
  61
  62     def config_fib_entries(self, count):
  63         n_int = len(self.interfaces)
  64         percent = 0
  65         counter = 0.0
  66         dest_addr = socket.inet_pton(socket.AF_INET6, "fd02::1")
  67         dest_addr_len = 128
  68         for i in self.interfaces:
  69             next_hop_address = i.local_ip6n
  70             for j in range(count / n_int):
  71                 self.vapi.ip_add_del_route(
  72                     dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
  73                 counter = counter + 1
  74                 if counter / count * 100 > percent:
  75                     info("Configure %d FIB entries .. %d%% done" %
  76                          (count, percent))
  77                     percent = percent + 1
  78
  79     def create_stream(self, src_if, packet_sizes):
  80         pkts = []
  81         for i in range(0, 257):
  82             dst_if = self.flows[src_if][i % 2]
  83             info = self.create_packet_info(
  84                 src_if.sw_if_index, dst_if.sw_if_index)
  85             payload = self.info_to_payload(info)
  86             p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
  87                  IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
  88                  UDP(sport=1234, dport=1234) /
  89                  Raw(payload))
  90             info.data = p.copy()
  91             if isinstance(src_if, VppSubInterface):
  92                 p = src_if.add_dot1_layer(p)
  93             size = packet_sizes[(i // 2) % len(packet_sizes)]
  94             self.extend_packet(p, size)
  95             pkts.append(p)
  96         return pkts
  97
  98     def verify_capture(self, dst_if, capture):
  99         info("Verifying capture on interface %s" % dst_if.name)
 100         last_info = dict()
 101         for i in self.interfaces:
 102             last_info[i.sw_if_index] = None
 103         is_sub_if = False
 104         dst_sw_if_index = dst_if.sw_if_index
 105         if hasattr(dst_if, 'parent'):
 106             is_sub_if = True
 107         for packet in capture:
 108             if is_sub_if:
 109                 # Check VLAN tags and Ethernet header
 110                 packet = dst_if.remove_dot1_layer(packet)
 111             self.assertTrue(Dot1Q not in packet)
 112             try:
 113                 ip = packet[IPv6]
 114                 udp = packet[UDP]
 115                 payload_info = self.payload_to_info(str(packet[Raw]))
 116                 packet_index = payload_info.index
 117                 self.assertEqual(payload_info.dst, dst_sw_if_index)
 118                 debug("Got packet on port %s: src=%u (id=%u)" %
 119                       (dst_if.name, payload_info.src, packet_index))
 120                 next_info = self.get_next_packet_info_for_interface2(
 121                     payload_info.src, dst_sw_if_index,
 122                     last_info[payload_info.src])
 123                 last_info[payload_info.src] = next_info
 124                 self.assertTrue(next_info is not None)
 125                 self.assertEqual(packet_index, next_info.index)
 126                 saved_packet = next_info.data
 127                 # Check standard fields
 128                 self.assertEqual(ip.src, saved_packet[IPv6].src)
 129                 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
 130                 self.assertEqual(udp.sport, saved_packet[UDP].sport)
 131                 self.assertEqual(udp.dport, saved_packet[UDP].dport)
 132             except:
 133                 error("Unexpected or invalid packet:")
 134                 error(packet.show())
 135                 raise
 136         for i in self.interfaces:
 137             remaining_packet = self.get_next_packet_info_for_interface2(
 138                 i.sw_if_index, dst_sw_if_index, last_info[i.sw_if_index])
 139             self.assertTrue(
 140                 remaining_packet is None,
 141                 "Interface %s: Packet expected from interface %s didn't arrive" %
 142                 (dst_if.name, i.name))
 143
 144     def test_fib(self):
 145         """ IPv6 FIB test """
 146
 147         pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
 148         self.pg0.add_stream(pkts)
 149
 150         for i in self.sub_interfaces:
 151             pkts = self.create_stream(i, self.sub_if_packet_sizes)
 152             i.parent.add_stream(pkts)
 153
 154         self.pg_enable_capture(self.pg_interfaces)
 155         self.pg_start()
 156
 157         pkts = self.pg0.get_capture()
 158         self.verify_capture(self.pg0, pkts)
 159
 160         for i in self.sub_interfaces:
 161             pkts = i.parent.get_capture()
 162             self.verify_capture(i, pkts)
 163
 164
 165 if __name__ == '__main__':
 166     unittest.main(testRunner=VppTestRunner)