2 """IP6 VRF Multi-instance Test Case HLD:
5 - higher number of pg-ip6 interfaces causes problems => only 15 pg-ip6 \
6 interfaces in 5 VRFs are tested
7 - jumbo packets in configuration with 15 pg-ip6 interfaces leads to \
11 - add 15 pg-ip6 interfaces
12 - configure 5 hosts per pg-ip6 interface
14 - add 3 pg-ip6 interfaces per VRF
17 - send IP6 packets between all pg-ip6 interfaces in all VRF groups
20 - check VRF data by parsing output of ip_route_dump API command
21 - all packets received correctly in case of pg-ip6 interfaces in the same
23 - no packet received in case of pg-ip6 interfaces not in VRF
24 - no packet received in case of pg-ip6 interfaces in different VRFs
30 - send IP6 packets between all pg-ip6 interfaces in all VRF groups
33 - check VRF data by parsing output of ip_route_dump API command
34 - all packets received correctly in case of pg-ip6 interfaces in the same
36 - no packet received in case of pg-ip6 interfaces not in VRF
37 - no packet received in case of pg-ip6 interfaces in different VRFs
40 - add 1 of reset VRFs and 1 new VRF
43 - send IP6 packets between all pg-ip6 interfaces in all VRF groups
46 - check VRF data by parsing output of ip_route_dump API command
47 - all packets received correctly in case of pg-ip6 interfaces in the same
49 - no packet received in case of pg-ip6 interfaces not in VRF
50 - no packet received in case of pg-ip6 interfaces in different VRFs
53 - reset all VRFs (i.e. no VRF except VRF=0 created)
56 - send IP6 packets between all pg-ip6 interfaces in all VRF groups
59 - check VRF data by parsing output of ip_route_dump API command
60 - all packets received correctly in case of pg-ip6 interfaces in the same
62 - no packet received in case of pg-ip6 interfaces not in VRF
63 - no packet received in case of pg-ip6 interfaces in different VRFs
70 from scapy.packet import Raw
71 from scapy.layers.l2 import Ether
72 from scapy.layers.inet6 import UDP, IPv6, ICMPv6ND_NS, ICMPv6ND_RA, \
73 RouterAlert, IPv6ExtHdrHopByHop
74 from scapy.utils6 import in6_ismaddr, in6_isllsnmaddr, in6_getAddrType
75 from scapy.pton_ntop import inet_ntop
77 from framework import VppTestCase, VppTestRunner
79 from vrf import VRFState
82 def is_ipv6_misc_ext(p):
83 """ Is packet one of uninteresting IPv6 broadcasts (extended to filter out
84 ICMPv6 Neighbor Discovery - Neighbor Advertisement packets too)? """
85 if p.haslayer(ICMPv6ND_RA):
86 if in6_ismaddr(p[IPv6].dst):
88 if p.haslayer(ICMPv6ND_NS):
89 if in6_isllsnmaddr(p[IPv6].dst):
91 if p.haslayer(IPv6ExtHdrHopByHop):
92 for o in p[IPv6ExtHdrHopByHop].options:
93 if isinstance(o, RouterAlert):
98 class TestIP6VrfMultiInst(VppTestCase):
99 """ IP6 VRF Multi-instance Test Case """
104 Perform standard class setup (defined by class method setUpClass in
105 class VppTestCase) before running the test case, set test case related
106 variables and configure VPP.
108 super(TestIP6VrfMultiInst, cls).setUpClass()
113 cls.pg_ifs_per_vrf = 3
116 # Create pg interfaces
117 cls.create_pg_interfaces(
118 range(cls.nr_of_vrfs * cls.pg_ifs_per_vrf))
120 # Packet flows mapping pg0 -> pg1, pg2 etc.
122 for i in range(len(cls.pg_interfaces)):
123 multiplicand = i // cls.pg_ifs_per_vrf
125 cls.pg_interfaces[multiplicand * cls.pg_ifs_per_vrf + j]
126 for j in range(cls.pg_ifs_per_vrf)
127 if (multiplicand * cls.pg_ifs_per_vrf + j) != i]
128 cls.flows[cls.pg_interfaces[i]] = pg_list
130 # Packet sizes - jumbo packet (9018 bytes) skipped
131 cls.pg_if_packet_sizes = [64, 512, 1518]
133 # Set up all interfaces
134 for pg_if in cls.pg_interfaces:
136 pg_if.generate_remote_hosts(cls.hosts_per_pg)
138 # Create list of VRFs
139 cls.vrf_list = list()
141 # Create list of reset VRFs
142 cls.vrf_reset_list = list()
144 # Create list of pg_interfaces in VRFs
145 cls.pg_in_vrf = list()
147 # Create list of pg_interfaces not in VRFs
148 cls.pg_not_in_vrf = [pg_if for pg_if in cls.pg_interfaces]
150 # Create mapping of pg_interfaces to VRF IDs
151 cls.pg_if_by_vrf_id = dict()
152 for i in range(cls.nr_of_vrfs):
155 cls.pg_interfaces[i * cls.pg_ifs_per_vrf + j]
156 for j in range(cls.pg_ifs_per_vrf)]
157 cls.pg_if_by_vrf_id[vrf_id] = pg_list
160 super(TestIP6VrfMultiInst, cls).tearDownClass()
164 def tearDownClass(cls):
165 super(TestIP6VrfMultiInst, cls).tearDownClass()
169 Clear trace and packet infos before running each test.
171 super(TestIP6VrfMultiInst, self).setUp()
172 self.reset_packet_infos()
176 Show various debug prints after each test.
178 super(TestIP6VrfMultiInst, self).tearDown()
180 def show_commands_at_teardown(self):
181 self.logger.info(self.vapi.ppcli("show ip6 fib"))
182 self.logger.info(self.vapi.ppcli("show ip6 neighbors"))
184 def create_vrf_and_assign_interfaces(self, count, start=1):
186 Create required number of FIB tables / VRFs, put 3 pg-ip6 interfaces
187 to every FIB table / VRF.
189 :param int count: Number of FIB tables / VRFs to be created.
190 :param int start: Starting number of the FIB table / VRF ID. \
193 for i in range(count):
195 pg_if = self.pg_if_by_vrf_id[vrf_id][0]
196 dest_addr = pg_if.local_ip6n
198 self.vapi.ip_table_add_del(is_add=1,
199 table={'table_id': vrf_id, 'is_ip6': 1})
200 self.logger.info("IPv6 VRF ID %d created" % vrf_id)
201 if vrf_id not in self.vrf_list:
202 self.vrf_list.append(vrf_id)
203 if vrf_id in self.vrf_reset_list:
204 self.vrf_reset_list.remove(vrf_id)
205 for j in range(self.pg_ifs_per_vrf):
206 pg_if = self.pg_if_by_vrf_id[vrf_id][j]
207 pg_if.set_table_ip6(vrf_id)
208 self.logger.info("pg-interface %s added to IPv6 VRF ID %d"
209 % (pg_if.name, vrf_id))
210 if pg_if not in self.pg_in_vrf:
211 self.pg_in_vrf.append(pg_if)
212 if pg_if in self.pg_not_in_vrf:
213 self.pg_not_in_vrf.remove(pg_if)
215 pg_if.disable_ipv6_ra()
216 pg_if.configure_ipv6_neighbors()
217 self.logger.debug(self.vapi.ppcli("show ip6 fib"))
218 self.logger.debug(self.vapi.ppcli("show ip6 neighbors"))
220 def reset_vrf_and_remove_from_vrf_list(self, vrf_id):
222 Reset required FIB table / VRF and remove it from VRF list.
224 :param int vrf_id: The FIB table / VRF ID to be reset.
226 self.vapi.ip_table_flush(table={'table_id': vrf_id, 'is_ip6': 1})
227 if vrf_id in self.vrf_list:
228 self.vrf_list.remove(vrf_id)
229 if vrf_id not in self.vrf_reset_list:
230 self.vrf_reset_list.append(vrf_id)
231 for j in range(self.pg_ifs_per_vrf):
232 pg_if = self.pg_if_by_vrf_id[vrf_id][j]
234 if pg_if in self.pg_in_vrf:
235 self.pg_in_vrf.remove(pg_if)
236 if pg_if not in self.pg_not_in_vrf:
237 self.pg_not_in_vrf.append(pg_if)
238 self.logger.info("IPv6 VRF ID %d reset finished" % vrf_id)
239 self.logger.debug(self.vapi.ppcli("show ip6 fib"))
240 self.logger.debug(self.vapi.ppcli("show ip6 neighbors"))
241 self.vapi.ip_table_add_del(is_add=0,
242 table={'table_id': vrf_id, 'is_ip6': 1})
244 def create_stream(self, src_if, packet_sizes):
246 Create input packet stream for defined interface using hosts list.
248 :param object src_if: Interface to create packet stream for.
249 :param list packet_sizes: List of required packet sizes.
250 :return: Stream of packets.
253 src_hosts = src_if.remote_hosts
254 for dst_if in self.flows[src_if]:
255 for dst_host in dst_if.remote_hosts:
256 src_host = random.choice(src_hosts)
257 pkt_info = self.create_packet_info(src_if, dst_if)
258 payload = self.info_to_payload(pkt_info)
259 p = (Ether(dst=src_if.local_mac, src=src_host.mac) /
260 IPv6(src=src_host.ip6, dst=dst_host.ip6) /
261 UDP(sport=1234, dport=1234) /
263 pkt_info.data = p.copy()
264 size = random.choice(packet_sizes)
265 self.extend_packet(p, size)
267 self.logger.debug("Input stream created for port %s. Length: %u pkt(s)"
268 % (src_if.name, len(pkts)))
271 def create_stream_crosswise_vrf(self, src_if, vrf_id, packet_sizes):
273 Create input packet stream for negative test for leaking across
274 different VRFs for defined interface using hosts list.
276 :param object src_if: Interface to create packet stream for.
277 :param int vrf_id: The FIB table / VRF ID where src_if is assigned.
278 :param list packet_sizes: List of required packet sizes.
279 :return: Stream of packets.
282 src_hosts = src_if.remote_hosts
283 vrf_lst = list(self.vrf_list)
284 vrf_lst.remove(vrf_id)
286 for dst_if in self.pg_if_by_vrf_id[vrf]:
287 for dst_host in dst_if.remote_hosts:
288 src_host = random.choice(src_hosts)
289 pkt_info = self.create_packet_info(src_if, dst_if)
290 payload = self.info_to_payload(pkt_info)
291 p = (Ether(dst=src_if.local_mac, src=src_host.mac) /
292 IPv6(src=src_host.ip6, dst=dst_host.ip6) /
293 UDP(sport=1234, dport=1234) /
295 pkt_info.data = p.copy()
296 size = random.choice(packet_sizes)
297 self.extend_packet(p, size)
299 self.logger.debug("Input stream created for port %s. Length: %u pkt(s)"
300 % (src_if.name, len(pkts)))
303 def verify_capture(self, pg_if, capture):
305 Verify captured input packet stream for defined interface.
307 :param object pg_if: Interface to verify captured packet stream for.
308 :param list capture: Captured packet stream.
311 for i in self.pg_interfaces:
312 last_info[i.sw_if_index] = None
313 dst_sw_if_index = pg_if.sw_if_index
314 for packet in capture:
318 payload_info = self.payload_to_info(packet[Raw])
319 packet_index = payload_info.index
320 self.assertEqual(payload_info.dst, dst_sw_if_index)
321 self.logger.debug("Got packet on port %s: src=%u (id=%u)" %
322 (pg_if.name, payload_info.src, packet_index))
323 next_info = self.get_next_packet_info_for_interface2(
324 payload_info.src, dst_sw_if_index,
325 last_info[payload_info.src])
326 last_info[payload_info.src] = next_info
327 self.assertIsNotNone(next_info)
328 self.assertEqual(packet_index, next_info.index)
329 saved_packet = next_info.data
330 # Check standard fields
331 self.assertEqual(ip.src, saved_packet[IPv6].src)
332 self.assertEqual(ip.dst, saved_packet[IPv6].dst)
333 self.assertEqual(udp.sport, saved_packet[UDP].sport)
334 self.assertEqual(udp.dport, saved_packet[UDP].dport)
336 self.logger.error(ppp("Unexpected or invalid packet:", packet))
338 for i in self.pg_interfaces:
339 remaining_packet = self.get_next_packet_info_for_interface2(
340 i, dst_sw_if_index, last_info[i.sw_if_index])
343 "Port %u: Packet expected from source %u didn't arrive" %
344 (dst_sw_if_index, i.sw_if_index))
346 def verify_vrf(self, vrf_id):
348 Check if the FIB table / VRF ID is configured.
350 :param int vrf_id: The FIB table / VRF ID to be verified.
351 :return: 1 if the FIB table / VRF ID is configured, otherwise return 0.
353 ip6_fib_dump = self.vapi.ip_route_dump(vrf_id, True)
354 vrf_exist = len(ip6_fib_dump)
356 for ip6_fib_details in ip6_fib_dump:
357 addr = ip6_fib_details.route.prefix.network_address
359 for pg_if in self.pg_if_by_vrf_id[vrf_id]:
362 for host in pg_if.remote_hosts:
363 if str(addr) == host.ip6:
367 if not vrf_exist and vrf_count == 0:
368 self.logger.info("IPv6 VRF ID %d is not configured" % vrf_id)
369 return VRFState.not_configured
370 elif vrf_exist and vrf_count == 0:
371 self.logger.info("IPv6 VRF ID %d has been reset" % vrf_id)
372 return VRFState.reset
374 self.logger.info("IPv6 VRF ID %d is configured" % vrf_id)
375 return VRFState.configured
377 def run_verify_test(self):
379 Create packet streams for all configured pg interfaces, send all \
380 prepared packet streams and verify that:
381 - all packets received correctly on all pg-ip6 interfaces assigned
383 - no packet received on all pg-ip6 interfaces not assigned to VRFs
385 :raise RuntimeError: If no packet captured on pg-ip6 interface assigned
386 to VRF or if any packet is captured on pg-ip6 interface not
390 # Create incoming packet streams for packet-generator interfaces
391 for pg_if in self.pg_interfaces:
392 pkts = self.create_stream(pg_if, self.pg_if_packet_sizes)
393 pg_if.add_stream(pkts)
395 # Enable packet capture and start packet sending
396 self.pg_enable_capture(self.pg_interfaces)
400 # Verify outgoing packet streams per packet-generator interface
401 for pg_if in self.pg_interfaces:
402 if pg_if in self.pg_in_vrf:
403 capture = pg_if.get_capture(remark="interface is in VRF")
404 self.verify_capture(pg_if, capture)
405 elif pg_if in self.pg_not_in_vrf:
406 pg_if.assert_nothing_captured(remark="interface is not in VRF",
407 filter_out_fn=is_ipv6_misc_ext)
408 self.logger.debug("No capture for interface %s" % pg_if.name)
410 raise Exception("Unknown interface: %s" % pg_if.name)
412 def run_crosswise_vrf_test(self):
414 Create packet streams for every pg-ip6 interface in VRF towards all
415 pg-ip6 interfaces in other VRFs, send all prepared packet streams and \
417 - no packet received on all configured pg-ip6 interfaces
419 :raise RuntimeError: If any packet is captured on any pg-ip6 interface.
422 # Create incoming packet streams for packet-generator interfaces
423 for vrf_id in self.vrf_list:
424 for pg_if in self.pg_if_by_vrf_id[vrf_id]:
425 pkts = self.create_stream_crosswise_vrf(
426 pg_if, vrf_id, self.pg_if_packet_sizes)
427 pg_if.add_stream(pkts)
429 # Enable packet capture and start packet sending
430 self.pg_enable_capture(self.pg_interfaces)
434 # Verify outgoing packet streams per packet-generator interface
435 for pg_if in self.pg_interfaces:
436 pg_if.assert_nothing_captured(remark="interface is in other VRF",
437 filter_out_fn=is_ipv6_misc_ext)
438 self.logger.debug("No capture for interface %s" % pg_if.name)
440 def test_ip6_vrf_01(self):
441 """ IP6 VRF Multi-instance test 1 - create 4 VRFs
445 self.create_vrf_and_assign_interfaces(4)
448 for vrf_id in self.vrf_list:
449 self.assert_equal(self.verify_vrf(vrf_id),
450 VRFState.configured, VRFState)
453 self.run_verify_test()
454 self.run_crosswise_vrf_test()
456 def test_ip6_vrf_02(self):
457 """ IP6 VRF Multi-instance test 2 - reset 2 VRFs
461 self.reset_vrf_and_remove_from_vrf_list(1)
462 self.reset_vrf_and_remove_from_vrf_list(2)
465 for vrf_id in self.vrf_reset_list:
466 self.assert_equal(self.verify_vrf(vrf_id),
467 VRFState.reset, VRFState)
468 for vrf_id in self.vrf_list:
469 self.assert_equal(self.verify_vrf(vrf_id),
470 VRFState.configured, VRFState)
473 self.run_verify_test()
474 self.run_crosswise_vrf_test()
476 # Reset routes learned from ICMPv6 Neighbor Discovery
477 for vrf_id in self.vrf_reset_list:
478 self.reset_vrf_and_remove_from_vrf_list(vrf_id)
480 def test_ip6_vrf_03(self):
481 """ IP6 VRF Multi-instance 3 - add 2 VRFs
484 # Add 1 of reset VRFs and 1 new VRF
485 self.create_vrf_and_assign_interfaces(1)
486 self.create_vrf_and_assign_interfaces(1, start=5)
489 for vrf_id in self.vrf_reset_list:
490 self.assert_equal(self.verify_vrf(vrf_id),
491 VRFState.reset, VRFState)
492 for vrf_id in self.vrf_list:
493 self.assert_equal(self.verify_vrf(vrf_id),
494 VRFState.configured, VRFState)
497 self.run_verify_test()
498 self.run_crosswise_vrf_test()
500 # Reset routes learned from ICMPv6 Neighbor Discovery
501 for vrf_id in self.vrf_reset_list:
502 self.reset_vrf_and_remove_from_vrf_list(vrf_id)
504 def test_ip6_vrf_04(self):
505 """ IP6 VRF Multi-instance test 4 - reset 4 VRFs
508 # Reset all VRFs (i.e. no VRF except VRF=0 configured)
509 for i in range(len(self.vrf_list)):
510 self.reset_vrf_and_remove_from_vrf_list(self.vrf_list[0])
513 for vrf_id in self.vrf_reset_list:
514 self.assert_equal(self.verify_vrf(vrf_id),
515 VRFState.reset, VRFState)
516 vrf_list_length = len(self.vrf_list)
519 "List of configured VRFs is not empty: %s != 0" % vrf_list_length)
522 self.run_verify_test()
523 self.run_crosswise_vrf_test()
526 if __name__ == '__main__':
527 unittest.main(testRunner=VppTestRunner)