7 from framework import VppTestCase, VppTestRunner
8 from vpp_sub_interface import VppSubInterface, VppDot1QSubint, VppDot1ADSubint
10 from scapy.packet import Raw
11 from scapy.layers.l2 import Ether, Dot1Q
12 from scapy.layers.inet6 import ICMPv6ND_NS, IPv6, UDP
15 class TestIPv6(VppTestCase):
16 """ IPv6 Test Case """
20 super(TestIPv6, cls).setUpClass()
23 super(TestIPv6, self).setUp()
25 # create 3 pg interfaces
26 self.create_pg_interfaces(range(3))
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)
34 # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
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]
41 self.pg_if_packet_sizes = [64, 512, 1518, 9018]
42 self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
44 self.interfaces = list(self.pg_interfaces)
45 self.interfaces.extend(self.sub_interfaces)
47 # setup all interfaces
48 for i in self.interfaces:
53 # config 2M FIB enries
54 self.config_fib_entries(200)
57 super(TestIPv6, self).tearDown()
59 info(self.vapi.cli("show ip6 neighbors"))
60 # info(self.vapi.cli("show ip6 fib")) # many entries
62 def config_fib_entries(self, count):
63 n_int = len(self.interfaces)
66 dest_addr = socket.inet_pton(socket.AF_INET6, "fd02::1")
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)
74 if counter / count * 100 > percent:
75 info("Configure %d FIB entries .. %d%% done" %
79 def create_stream(self, src_if, packet_sizes):
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) /
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)
98 def verify_capture(self, dst_if, capture):
99 info("Verifying capture on interface %s" % dst_if.name)
101 for i in self.interfaces:
102 last_info[i.sw_if_index] = None
104 dst_sw_if_index = dst_if.sw_if_index
105 if hasattr(dst_if, 'parent'):
107 for packet in capture:
109 # Check VLAN tags and Ethernet header
110 packet = dst_if.remove_dot1_layer(packet)
111 self.assertTrue(Dot1Q not in packet)
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)
133 error("Unexpected or invalid packet:")
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])
140 remaining_packet is None,
141 "Interface %s: Packet expected from interface %s didn't arrive" %
142 (dst_if.name, i.name))
145 """ IPv6 FIB test """
147 pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
148 self.pg0.add_stream(pkts)
150 for i in self.sub_interfaces:
151 pkts = self.create_stream(i, self.sub_if_packet_sizes)
152 i.parent.add_stream(pkts)
154 self.pg_enable_capture(self.pg_interfaces)
157 pkts = self.pg0.get_capture()
158 self.verify_capture(self.pg0, pkts)
160 for i in self.sub_interfaces:
161 pkts = i.parent.get_capture()
162 self.verify_capture(i, pkts)
165 if __name__ == '__main__':
166 unittest.main(testRunner=VppTestRunner)