6 from framework import VppTestCase, VppTestRunner
7 from vpp_sub_interface import VppSubInterface, VppDot1QSubint
9 from scapy.packet import Raw
10 from scapy.layers.l2 import Ether, Dot1Q
11 from scapy.layers.inet6 import IPv6, UDP
14 class TestIPv6(VppTestCase):
15 """ IPv6 Test Case """
19 super(TestIPv6, cls).setUpClass()
23 Perform test setup before test case.
26 - create 3 pg interfaces
27 - untagged pg0 interface
28 - Dot1Q subinterface on pg1
29 - Dot1AD subinterface on pg2
31 - put it into UP state
33 - resolve neighbor address using NDP
34 - configure 200 fib entries
36 :ivar list interfaces: pg interfaces and subinterfaces.
37 :ivar dict flows: IPv4 packet flows in test.
38 :ivar list pg_if_packet_sizes: packet sizes in test.
40 *TODO:* Create AD sub interface
42 super(TestIPv6, self).setUp()
44 # create 3 pg interfaces
45 self.create_pg_interfaces(range(3))
47 # create 2 subinterfaces for p1 and pg2
48 self.sub_interfaces = [
49 VppDot1QSubint(self, self.pg1, 100),
50 VppDot1QSubint(self, self.pg2, 200)
51 # TODO: VppDot1ADSubint(self, self.pg2, 200, 300, 400)
54 # packet flows mapping pg0 -> pg1.sub, pg2.sub, etc.
56 self.flows[self.pg0] = [self.pg1.sub_if, self.pg2.sub_if]
57 self.flows[self.pg1.sub_if] = [self.pg0, self.pg2.sub_if]
58 self.flows[self.pg2.sub_if] = [self.pg0, self.pg1.sub_if]
61 self.pg_if_packet_sizes = [64, 512, 1518, 9018]
62 self.sub_if_packet_sizes = [64, 512, 1518 + 4, 9018 + 4]
64 self.interfaces = list(self.pg_interfaces)
65 self.interfaces.extend(self.sub_interfaces)
67 # setup all interfaces
68 for i in self.interfaces:
73 # config 2M FIB entries
74 self.config_fib_entries(200)
77 """Run standard test teardown and log ``show ip6 neighbors``."""
78 super(TestIPv6, self).tearDown()
80 self.logger.info(self.vapi.cli("show ip6 neighbors"))
81 # info(self.vapi.cli("show ip6 fib")) # many entries
83 def config_fib_entries(self, count):
84 """For each interface add to the FIB table *count* routes to
85 "fd02::1/128" destination with interface's local address as next-hop
88 :param int count: Number of FIB entries.
90 - *TODO:* check if the next-hop address shouldn't be remote address
91 instead of local address.
93 n_int = len(self.interfaces)
96 dest_addr = socket.inet_pton(socket.AF_INET6, "fd02::1")
98 for i in self.interfaces:
99 next_hop_address = i.local_ip6n
100 for j in range(count / n_int):
101 self.vapi.ip_add_del_route(
102 dest_addr, dest_addr_len, next_hop_address, is_ipv6=1)
104 if counter / count * 100 > percent:
105 self.logger.info("Configure %d FIB entries .. %d%% done" %
109 def create_stream(self, src_if, packet_sizes):
110 """Create input packet stream for defined interface.
112 :param VppInterface src_if: Interface to create packet stream for.
113 :param list packet_sizes: Required packet sizes.
116 for i in range(0, 257):
117 dst_if = self.flows[src_if][i % 2]
118 info = self.create_packet_info(
119 src_if.sw_if_index, dst_if.sw_if_index)
120 payload = self.info_to_payload(info)
121 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
122 IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6) /
123 UDP(sport=1234, dport=1234) /
126 if isinstance(src_if, VppSubInterface):
127 p = src_if.add_dot1_layer(p)
128 size = packet_sizes[(i // 2) % len(packet_sizes)]
129 self.extend_packet(p, size)
133 def verify_capture(self, dst_if, capture):
134 """Verify captured input packet stream for defined interface.
136 :param VppInterface dst_if: Interface to verify captured packet stream
138 :param list capture: Captured packet stream.
140 self.logger.info("Verifying capture on interface %s" % dst_if.name)
142 for i in self.interfaces:
143 last_info[i.sw_if_index] = None
145 dst_sw_if_index = dst_if.sw_if_index
146 if hasattr(dst_if, 'parent'):
148 for packet in capture:
150 # Check VLAN tags and Ethernet header
151 packet = dst_if.remove_dot1_layer(packet)
152 self.assertTrue(Dot1Q not in packet)
156 payload_info = self.payload_to_info(str(packet[Raw]))
157 packet_index = payload_info.index
158 self.assertEqual(payload_info.dst, dst_sw_if_index)
159 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
160 (dst_if.name, payload_info.src, packet_index))
161 next_info = self.get_next_packet_info_for_interface2(
162 payload_info.src, dst_sw_if_index,
163 last_info[payload_info.src])
164 last_info[payload_info.src] = next_info
165 self.assertTrue(next_info is not None)
166 self.assertEqual(packet_index, next_info.index)
167 saved_packet = next_info.data
168 # Check standard fields
169 self.assertEqual(ip.src, saved_packet[IPv6].src)
170 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
171 self.assertEqual(udp.sport, saved_packet[UDP].sport)
172 self.assertEqual(udp.dport, saved_packet[UDP].dport)
174 self.logger.error("Unexpected or invalid packet:")
175 self.logger.error(packet.show())
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])
181 remaining_packet is None,
182 "Interface %s: Packet expected from interface %s didn't arrive" %
183 (dst_if.name, i.name))
189 - Create IPv6 stream for pg0 interface
190 - Create IPv6 tagged streams for pg1's and pg2's subinterface.
191 - Send and verify received packets on each interface.
194 pkts = self.create_stream(self.pg0, self.pg_if_packet_sizes)
195 self.pg0.add_stream(pkts)
197 for i in self.sub_interfaces:
198 pkts = self.create_stream(i, self.sub_if_packet_sizes)
199 i.parent.add_stream(pkts)
201 self.pg_enable_capture(self.pg_interfaces)
204 pkts = self.pg0.get_capture()
205 self.verify_capture(self.pg0, pkts)
207 for i in self.sub_interfaces:
208 pkts = i.parent.get_capture()
209 self.verify_capture(i, pkts)
212 if __name__ == '__main__':
213 unittest.main(testRunner=VppTestRunner)