8 from struct import unpack, unpack_from
11 import unittest2 as unittest
15 from util import ppp, ppc
16 from re import compile
18 from scapy.packet import Raw
19 from scapy.layers.l2 import Ether
20 from scapy.layers.inet import IP, UDP, ICMP
21 import scapy.layers.inet6 as inet6
22 from scapy.layers.inet6 import IPv6, ICMPv6DestUnreach
24 from framework import VppTestCase, VppTestRunner
26 from vpp_ip import DpoProto
27 from vpp_ip_route import VppIpRoute, VppRoutePath
31 def get_mac_addr(bytes_addr):
32 return ':'.join('%02x' % scapy.compat.orb(b) for b in bytes_addr)
37 return '.'.join('%d' % scapy.compat.orb(b) for b in bytes_addr)
40 # Unpack Ethernet Frame
41 def ethernet_frame(data):
42 dest_mac, src_mac, proto = struct.unpack('! 6s 6s H', data[:14])
43 return dest_mac, src_mac, socket.htons(proto), data[14:]
47 def ipv4_packet(data):
48 proto, src, target = struct.unpack('! 8x 1x B 2x 4s 4s', data[:20])
49 return proto, src, target, data[20:]
53 def ipv6_packet(data):
54 nh, src, target = struct.unpack('! 6x B 1x 16s 16s', data[:40])
55 return nh, src, target, data[40:]
58 # Unpacks any UDP Packet
60 src_port, dest_port, size = struct.unpack('! H H 2x H', data[:8])
61 return src_port, dest_port, size, data[8:]
64 # Unpacks any TCP Packet
66 src_port, dest_port, seq, flag = struct.unpack('! H H L 4x H', data[:14])
67 return src_port, dest_port, seq, data[((flag >> 12) * 4):]
70 def receivePackets(sock, counters):
71 # Wait for some packets on socket
73 data = sock.recv(65536)
75 # punt socket metadata
76 # packet_desc = data[0:8]
79 _, _, eth_proto, data = ethernet_frame(data[8:])
82 proto, _, _, data = ipv4_packet(data)
85 _, dst_port, _, data = udp_seg(data)
88 _, dst_port, _, data = udp_seg(data)
89 counters[dst_port] = 0
91 elif eth_proto == 0xdd86:
92 nh, _, _, data = ipv6_packet(data)
95 _, dst_port, _, data = udp_seg(data)
98 _, dst_port, _, data = udp_seg(data)
99 counters[dst_port] = 0
102 class serverSocketThread(threading.Thread):
103 """ Socket server thread"""
105 def __init__(self, threadID, sockName, counters):
106 threading.Thread.__init__(self)
107 self.threadID = threadID
108 self.sockName = sockName
110 self.counters = counters
113 self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
115 os.unlink(self.sockName)
118 self.sock.bind(self.sockName)
120 receivePackets(self.sock, self.counters)
123 class TestPuntSocket(VppTestCase):
126 ports = [1111, 2222, 3333, 4444]
127 sock_servers = list()
133 super(TestPuntSocket, cls).setUpClass()
136 def tearDownClass(cls):
137 super(TestPuntSocket, cls).tearDownClass()
140 def setUpConstants(cls):
141 cls.extra_vpp_punt_config = [
142 "punt", "{", "socket", cls.tempdir+"/socket_punt", "}"]
143 super(TestPuntSocket, cls).setUpConstants()
146 super(TestPuntSocket, self).setUp()
149 self.create_pg_interfaces(range(2))
150 for i in self.pg_interfaces:
154 del self.sock_servers[:]
155 super(TestPuntSocket, self).tearDown()
157 def socket_client_create(self, sock_name, id=None):
158 thread = serverSocketThread(id, sock_name, self.portsCheck)
159 self.sock_servers.append(thread)
162 def socket_client_close(self):
163 for thread in self.sock_servers:
167 class TestIP4PuntSocket(TestPuntSocket):
168 """ Punt Socket for IPv4 """
172 super(TestIP4PuntSocket, cls).setUpClass()
175 def tearDownClass(cls):
176 super(TestIP4PuntSocket, cls).tearDownClass()
179 super(TestIP4PuntSocket, self).setUp()
181 for i in self.pg_interfaces:
186 super(TestIP4PuntSocket, self).tearDown()
187 for i in self.pg_interfaces:
191 def test_punt_socket_dump(self):
192 """ Punt socket registration/deregistration"""
194 punts = self.vapi.punt_socket_dump(is_ip6=0)
195 self.assertEqual(len(punts), 0)
198 # configure a punt socket
200 self.vapi.punt_socket_register(1111, b"%s/socket_punt_1111" %
201 six.ensure_binary(self.tempdir))
202 self.vapi.punt_socket_register(2222, b"%s/socket_punt_2222" %
203 six.ensure_binary(self.tempdir))
204 punts = self.vapi.punt_socket_dump(is_ip6=0)
205 self.assertEqual(len(punts), 2)
206 self.assertEqual(punts[0].punt.l4_port, 1111)
207 self.assertEqual(punts[1].punt.l4_port, 2222)
210 # deregister a punt socket
212 self.vapi.punt_socket_deregister(1111)
213 punts = self.vapi.punt_socket_dump(is_ip6=0)
214 self.assertEqual(len(punts), 1)
217 # configure a punt socket again
219 self.vapi.punt_socket_register(1111, b"%s/socket_punt_1111" %
220 six.ensure_binary(self.tempdir))
221 self.vapi.punt_socket_register(3333, b"%s/socket_punt_3333" %
222 six.ensure_binary(self.tempdir))
223 punts = self.vapi.punt_socket_dump(is_ip6=0)
224 self.assertEqual(len(punts), 3)
227 # deregister all punt socket
229 self.vapi.punt_socket_deregister(1111)
230 self.vapi.punt_socket_deregister(2222)
231 self.vapi.punt_socket_deregister(3333)
232 punts = self.vapi.punt_socket_dump(is_ip6=0)
233 self.assertEqual(len(punts), 0)
235 def test_punt_socket_traffic_single_port_single_socket(self):
236 """ Punt socket traffic single port single socket"""
240 p = (Ether(src=self.pg0.remote_mac,
241 dst=self.pg0.local_mac) /
242 IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
243 UDP(sport=9876, dport=port) /
246 pkts = p * self.nr_packets
247 self.portsCheck[port] = self.nr_packets
249 punts = self.vapi.punt_socket_dump(is_ip6=0)
250 self.assertEqual(len(punts), 0)
253 # expect ICMP - port unreachable for all packets
255 self.vapi.cli("clear trace")
256 self.pg0.add_stream(pkts)
257 self.pg_enable_capture(self.pg_interfaces)
259 # FIXME - when punt socket deregister is implemented
260 # rx = self.pg0.get_capture(self.nr_packets)
262 # self.assertEqual(int(p[IP].proto), 1) # ICMP
263 # self.assertEqual(int(p[ICMP].code), 3) # unreachable
266 # configure a punt socket
268 self.socket_client_create(b"%s/socket_%d" % (
269 six.ensure_binary(self.tempdir), port))
270 self.vapi.punt_socket_register(port, b"%s/socket_%d" % (
271 six.ensure_binary(self.tempdir), port))
272 punts = self.vapi.punt_socket_dump(is_ip6=0)
273 self.assertEqual(len(punts), 1)
275 self.logger.debug("Sending %s packets to port %d",
276 str(self.portsCheck[port]), port)
278 # expect punt socket and no packets on pg0
280 self.vapi.cli("clear errors")
281 self.vapi.cli("clear trace")
282 self.pg0.add_stream(pkts)
283 self.pg_enable_capture(self.pg_interfaces)
285 self.pg0.get_capture(0)
286 self.logger.info(self.vapi.cli("show trace"))
287 self.socket_client_close()
288 self.assertEqual(self.portsCheck[port], 0)
291 # remove punt socket. expect ICMP - port unreachable for all packets
293 self.vapi.punt_socket_deregister(port)
294 punts = self.vapi.punt_socket_dump(is_ip6=0)
295 self.assertEqual(len(punts), 0)
296 self.pg0.add_stream(pkts)
297 self.pg_enable_capture(self.pg_interfaces)
299 # FIXME - when punt socket deregister is implemented
300 # self.pg0.get_capture(nr_packets)
302 def test_punt_socket_traffic_multi_port_multi_sockets(self):
303 """ Punt socket traffic multi ports and multi sockets"""
306 self.portsCheck[p] = 0
309 # create stream with random packets count per given ports
312 for _ in range(0, self.nr_packets):
313 # choose port from port list
314 p = random.choice(self.ports)
316 Ether(src=self.pg0.remote_mac,
317 dst=self.pg0.local_mac) /
318 IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
319 UDP(sport=9876, dport=p) /
321 self.portsCheck[p] += 1
325 punts = self.vapi.punt_socket_dump(is_ip6=0)
326 self.assertEqual(len(punts), 0)
329 # configure a punt socket
332 self.socket_client_create(b"%s/socket_%d" % (
333 six.ensure_binary(self.tempdir), p))
334 self.vapi.punt_socket_register(p, b"%s/socket_%d" % (
335 six.ensure_binary(self.tempdir), p))
336 punts = self.vapi.punt_socket_dump(is_ip6=0)
337 self.assertEqual(len(punts), len(self.ports))
340 self.logger.debug("Sending %s packets to port %d",
341 str(self.portsCheck[p]), p)
344 # expect punt socket and no packets on pg0
346 self.vapi.cli("clear errors")
347 self.vapi.cli("clear trace")
348 self.pg0.add_stream(pkts)
349 self.pg_enable_capture(self.pg_interfaces)
351 self.pg0.get_capture(0)
352 self.logger.info(self.vapi.cli("show trace"))
353 self.socket_client_close()
356 self.assertEqual(self.portsCheck[p], 0)
357 self.vapi.punt_socket_deregister(p)
358 punts = self.vapi.punt_socket_dump(is_ip6=0)
359 self.assertEqual(len(punts), 0)
361 def test_punt_socket_traffic_multi_ports_single_socket(self):
362 """ Punt socket traffic multi ports and single socket"""
365 self.portsCheck[p] = 0
368 # create stream with random packets count per given ports
371 for _ in range(0, self.nr_packets):
372 # choose port from port list
373 p = random.choice(self.ports)
375 Ether(src=self.pg0.remote_mac,
376 dst=self.pg0.local_mac) /
377 IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
378 UDP(sport=9876, dport=p) /
380 self.portsCheck[p] += 1
385 punts = self.vapi.punt_socket_dump(is_ip6=0)
386 self.assertEqual(len(punts), 0)
388 # configure a punt socket
390 self.socket_client_create(b"%s/socket_multi" %
391 six.ensure_binary(self.tempdir))
393 self.vapi.punt_socket_register(p,
395 six.ensure_binary(self.tempdir))
396 punts = self.vapi.punt_socket_dump(is_ip6=0)
397 self.assertEqual(len(punts), len(self.ports))
400 self.logger.debug("Sending %s packets to port %d",
401 str(self.portsCheck[p]), p)
403 # expect punt socket and no packets on pg0
405 self.vapi.cli("clear errors")
406 self.vapi.cli("clear trace")
407 self.pg0.add_stream(pkts)
408 self.pg_enable_capture(self.pg_interfaces)
410 self.pg0.get_capture(0)
411 self.logger.info(self.vapi.cli("show trace"))
412 self.socket_client_close()
415 self.assertEqual(self.portsCheck[p], 0)
416 self.vapi.punt_socket_deregister(p)
417 punts = self.vapi.punt_socket_dump(is_ip6=0)
418 self.assertEqual(len(punts), 0)
421 class TestIP6PuntSocket(TestPuntSocket):
422 """ Punt Socket for IPv6"""
426 super(TestIP6PuntSocket, cls).setUpClass()
429 def tearDownClass(cls):
430 super(TestIP6PuntSocket, cls).tearDownClass()
433 super(TestIP6PuntSocket, self).setUp()
435 for i in self.pg_interfaces:
440 super(TestIP6PuntSocket, self).tearDown()
441 for i in self.pg_interfaces:
445 def test_punt_socket_dump(self):
446 """ Punt socket registration """
448 punts = self.vapi.punt_socket_dump(is_ip6=1)
449 self.assertEqual(len(punts), 0)
452 # configure a punt socket
454 self.vapi.punt_socket_register(1111, b"%s/socket_1111" %
455 six.ensure_binary(self.tempdir),
457 self.vapi.punt_socket_register(2222, b"%s/socket_2222" %
458 six.ensure_binary(self.tempdir),
460 punts = self.vapi.punt_socket_dump(is_ip6=1)
461 self.assertEqual(len(punts), 2)
462 self.assertEqual(punts[0].punt.l4_port, 1111)
463 self.assertEqual(punts[1].punt.l4_port, 2222)
466 # deregister a punt socket
468 self.vapi.punt_socket_deregister(1111, is_ip4=0)
469 punts = self.vapi.punt_socket_dump(is_ip6=1)
470 self.assertEqual(len(punts), 1)
473 # configure a punt socket again
475 self.vapi.punt_socket_register(1111, b"%s/socket_1111" %
476 six.ensure_binary(self.tempdir),
478 punts = self.vapi.punt_socket_dump(is_ip6=1)
479 self.assertEqual(len(punts), 2)
482 # deregister all punt socket
484 self.vapi.punt_socket_deregister(1111, is_ip4=0)
485 self.vapi.punt_socket_deregister(2222, is_ip4=0)
486 self.vapi.punt_socket_deregister(3333, is_ip4=0)
487 punts = self.vapi.punt_socket_dump(is_ip6=1)
488 self.assertEqual(len(punts), 0)
490 def test_punt_socket_traffic_single_port_single_socket(self):
491 """ Punt socket traffic single port single socket"""
495 p = (Ether(src=self.pg0.remote_mac,
496 dst=self.pg0.local_mac) /
497 IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
498 inet6.UDP(sport=9876, dport=port) /
501 pkts = p * self.nr_packets
502 self.portsCheck[port] = self.nr_packets
504 punts = self.vapi.punt_socket_dump(is_ip6=1)
505 self.assertEqual(len(punts), 0)
508 # expect ICMPv6 - destination unreachable for all packets
510 self.vapi.cli("clear trace")
511 self.pg0.add_stream(pkts)
512 self.pg_enable_capture(self.pg_interfaces)
514 # FIXME - when punt socket deregister is implemented
515 # rx = self.pg0.get_capture(self.nr_packets)
517 # self.assertEqual(int(p[IPv6].nh), 58) # ICMPv6
518 # self.assertEqual(int(p[ICMPv6DestUnreach].code),4) # unreachable
521 # configure a punt socket
523 self.socket_client_create(b"%s/socket_%d" % (
524 six.ensure_binary(self.tempdir), port))
525 self.vapi.punt_socket_register(port, b"%s/socket_%d" % (
526 six.ensure_binary(self.tempdir), port), is_ip4=0)
527 punts = self.vapi.punt_socket_dump(is_ip6=1)
528 self.assertEqual(len(punts), 1)
530 self.logger.debug("Sending %s packets to port %d",
531 str(self.portsCheck[port]), port)
533 # expect punt socket and no packets on pg0
535 self.vapi.cli("clear errors")
536 self.vapi.cli("clear trace")
537 self.pg0.add_stream(pkts)
538 self.pg_enable_capture(self.pg_interfaces)
540 self.pg0.get_capture(0)
541 self.logger.info(self.vapi.cli("show trace"))
542 self.socket_client_close()
543 self.assertEqual(self.portsCheck[port], 0)
546 # remove punt socket. expect ICMP - dest. unreachable for all packets
548 self.vapi.punt_socket_deregister(port, is_ip4=0)
549 punts = self.vapi.punt_socket_dump(is_ip6=1)
550 self.assertEqual(len(punts), 0)
551 self.pg0.add_stream(pkts)
552 self.pg_enable_capture(self.pg_interfaces)
554 # FIXME - when punt socket deregister is implemented
555 # self.pg0.get_capture(nr_packets)
557 def test_punt_socket_traffic_multi_port_multi_sockets(self):
558 """ Punt socket traffic multi ports and multi sockets"""
561 self.portsCheck[p] = 0
564 # create stream with random packets count per given ports
567 for _ in range(0, self.nr_packets):
568 # choose port from port list
569 p = random.choice(self.ports)
571 Ether(src=self.pg0.remote_mac,
572 dst=self.pg0.local_mac) /
573 IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
574 inet6.UDP(sport=9876, dport=p) /
576 self.portsCheck[p] += 1
580 punts = self.vapi.punt_socket_dump(is_ip6=1)
581 self.assertEqual(len(punts), 0)
584 # configure a punt socket
587 self.socket_client_create(b"%s/socket_%d" % (
588 six.ensure_binary(self.tempdir), p))
589 self.vapi.punt_socket_register(p, b"%s/socket_%d" % (
590 six.ensure_binary(self.tempdir), p), is_ip4=0)
591 punts = self.vapi.punt_socket_dump(is_ip6=1)
592 self.assertEqual(len(punts), len(self.ports))
595 self.logger.debug("Sending %s packets to port %d",
596 str(self.portsCheck[p]), p)
599 # expect punt socket and no packets on pg0
601 self.vapi.cli("clear errors")
602 self.vapi.cli("clear trace")
603 self.pg0.add_stream(pkts)
604 self.pg_enable_capture(self.pg_interfaces)
606 self.pg0.get_capture(0)
607 self.logger.info(self.vapi.cli("show trace"))
608 self.socket_client_close()
611 self.assertEqual(self.portsCheck[p], 0)
612 self.vapi.punt_socket_deregister(p, is_ip4=0)
613 punts = self.vapi.punt_socket_dump(is_ip6=1)
614 self.assertEqual(len(punts), 0)
616 def test_punt_socket_traffic_multi_ports_single_socket(self):
617 """ Punt socket traffic multi ports and single socket"""
620 self.portsCheck[p] = 0
623 # create stream with random packets count per given ports
626 for _ in range(0, self.nr_packets):
627 # choose port from port list
628 p = random.choice(self.ports)
630 Ether(src=self.pg0.remote_mac,
631 dst=self.pg0.local_mac) /
632 IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
633 inet6.UDP(sport=9876, dport=p) /
635 self.portsCheck[p] += 1
640 punts = self.vapi.punt_socket_dump(is_ip6=1)
641 self.assertEqual(len(punts), 0)
644 # configure a punt socket
646 self.socket_client_create(b"%s/socket_multi" %
647 six.ensure_binary(self.tempdir))
649 self.vapi.punt_socket_register(p,
651 six.ensure_binary(self.tempdir),
653 punts = self.vapi.punt_socket_dump(is_ip6=1)
654 self.assertEqual(len(punts), len(self.ports))
657 self.logger.debug("Send %s packets to port %d",
658 str(self.portsCheck[p]), p)
660 # expect punt socket and no packets on pg0
662 self.vapi.cli("clear errors")
663 self.vapi.cli("clear trace")
664 self.pg0.add_stream(pkts)
665 self.pg_enable_capture(self.pg_interfaces)
667 self.pg0.get_capture(0)
668 self.logger.info(self.vapi.cli("show trace"))
669 self.socket_client_close()
672 self.assertEqual(self.portsCheck[p], 0)
673 self.vapi.punt_socket_deregister(p, is_ip4=0)
674 punts = self.vapi.punt_socket_dump(is_ip6=1)
675 self.assertEqual(len(punts), 0)
678 class TestPunt(VppTestCase):
679 """ Punt Test Case """
683 super(TestPunt, cls).setUpClass()
686 def tearDownClass(cls):
687 super(TestPunt, cls).tearDownClass()
690 super(TestPunt, self).setUp()
692 self.create_pg_interfaces(range(4))
694 for i in self.pg_interfaces:
702 for i in self.pg_interfaces:
707 super(TestPunt, self).tearDown()
710 """ Excpetion Path testing """
713 # Using the test CLI we will hook in a exception path to
714 # send ACL deny packets out of pg0 and pg1.
715 # the ACL is src,dst = 1.1.1.1,1.1.1.2
717 ip_1_1_1_2 = VppIpRoute(self, "1.1.1.2", 32,
718 [VppRoutePath(self.pg3.remote_ip4,
719 self.pg3.sw_if_index)])
720 ip_1_1_1_2.add_vpp_config()
721 ip_1_2 = VppIpRoute(self, "1::2", 128,
722 [VppRoutePath(self.pg3.remote_ip6,
723 self.pg3.sw_if_index,
724 proto=DpoProto.DPO_PROTO_IP6)],
726 ip_1_2.add_vpp_config()
728 p4 = (Ether(src=self.pg2.remote_mac,
729 dst=self.pg2.local_mac) /
730 IP(src="1.1.1.1", dst="1.1.1.2") /
731 UDP(sport=1234, dport=1234) /
733 p6 = (Ether(src=self.pg2.remote_mac,
734 dst=self.pg2.local_mac) /
735 IPv6(src="1::1", dst="1::2") /
736 UDP(sport=1234, dport=1234) /
738 self.send_and_expect(self.pg2, p4*1, self.pg3)
739 self.send_and_expect(self.pg2, p6*1, self.pg3)
742 # apply the punting features
744 self.vapi.cli("test punt pg2")
749 self.send_and_assert_no_replies(self.pg2, p4*65)
750 self.send_and_assert_no_replies(self.pg2, p6*65)
754 # 1 - node error counters
755 # 2 - per-reason counters
756 # 2, 3 are the index of the assigned punt reason
758 stats = self.statistics.get_counter(
759 "/err/punt-dispatch/No registrations")
760 self.assertEqual(stats, 130)
762 stats = self.statistics.get_counter("/net/punt")
763 self.assertEqual(stats[0][7]['packets'], 65)
764 self.assertEqual(stats[0][8]['packets'], 65)
767 # use the test CLI to test a client that punts exception
770 self.vapi.cli("test punt pg0 %s" % self.pg0.remote_ip4)
771 self.vapi.cli("test punt pg0 %s" % self.pg0.remote_ip6)
773 rx4s = self.send_and_expect(self.pg2, p4*65, self.pg0)
774 rx6s = self.send_and_expect(self.pg2, p6*65, self.pg0)
777 # check the packets come out IP unmodified but destined to pg0 host
780 self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
781 self.assertEqual(rx[Ether].src, self.pg0.local_mac)
782 self.assertEqual(p4[IP].dst, rx[IP].dst)
783 self.assertEqual(p4[IP].ttl, rx[IP].ttl)
785 self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
786 self.assertEqual(rx[Ether].src, self.pg0.local_mac)
787 self.assertEqual(p6[IPv6].dst, rx[IPv6].dst)
788 self.assertEqual(p6[IPv6].hlim, rx[IPv6].hlim)
790 stats = self.statistics.get_counter("/net/punt")
791 self.assertEqual(stats[0][7]['packets'], 2*65)
792 self.assertEqual(stats[0][8]['packets'], 2*65)
795 # add another registration for the same reason to send packets
798 self.vapi.cli("test punt pg1 %s" % self.pg1.remote_ip4)
799 self.vapi.cli("test punt pg1 %s" % self.pg1.remote_ip6)
801 self.vapi.cli("clear trace")
802 self.pg2.add_stream(p4 * 65)
803 self.pg_enable_capture(self.pg_interfaces)
806 rxd = self.pg0.get_capture(65)
808 self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
809 self.assertEqual(rx[Ether].src, self.pg0.local_mac)
810 self.assertEqual(p4[IP].dst, rx[IP].dst)
811 self.assertEqual(p4[IP].ttl, rx[IP].ttl)
812 rxd = self.pg1.get_capture(65)
814 self.assertEqual(rx[Ether].dst, self.pg1.remote_mac)
815 self.assertEqual(rx[Ether].src, self.pg1.local_mac)
816 self.assertEqual(p4[IP].dst, rx[IP].dst)
817 self.assertEqual(p4[IP].ttl, rx[IP].ttl)
819 self.vapi.cli("clear trace")
820 self.pg2.add_stream(p6 * 65)
821 self.pg_enable_capture(self.pg_interfaces)
824 rxd = self.pg0.get_capture(65)
826 self.assertEqual(rx[Ether].dst, self.pg0.remote_mac)
827 self.assertEqual(rx[Ether].src, self.pg0.local_mac)
828 self.assertEqual(p6[IPv6].dst, rx[IPv6].dst)
829 self.assertEqual(p6[IPv6].hlim, rx[IPv6].hlim)
830 rxd = self.pg1.get_capture(65)
832 self.assertEqual(rx[Ether].dst, self.pg1.remote_mac)
833 self.assertEqual(rx[Ether].src, self.pg1.local_mac)
834 self.assertEqual(p6[IPv6].dst, rx[IPv6].dst)
835 self.assertEqual(p6[IPv6].hlim, rx[IPv6].hlim)
837 stats = self.statistics.get_counter("/net/punt")
838 self.assertEqual(stats[0][7]['packets'], 3*65)
839 self.assertEqual(stats[0][8]['packets'], 3*65)
841 self.logger.info(self.vapi.cli("show vlib graph punt-dispatch"))
842 self.logger.info(self.vapi.cli("show punt client"))
843 self.logger.info(self.vapi.cli("show punt reason"))
844 self.logger.info(self.vapi.cli("show punt stats"))
845 self.logger.info(self.vapi.cli("show punt db"))
847 self.vapi.cli("test punt clear")
850 if __name__ == '__main__':
851 unittest.main(testRunner=VppTestRunner)