test/test_punt.py

   1 #!/usr/bin/env python
   2 import binascii
   3 import random
   4 import socket
   5 import os
   6 import threading
   7 import struct
   8 from struct import unpack, unpack_from
   9
  10 try:
  11     import unittest2 as unittest
  12 except ImportError:
  13     import unittest
  14
  15 from util import ppp, ppc
  16 from re import compile
  17 import scapy.compat
  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
  23 import six
  24 from framework import VppTestCase, VppTestRunner
  25
  26
  27 # Format MAC Address
  28 def get_mac_addr(bytes_addr):
  29     return ':'.join('%02x' % scapy.compat.orb(b) for b in bytes_addr)
  30
  31
  32 # Format IP Address
  33 def ipv4(bytes_addr):
  34     return '.'.join('%d' % scapy.compat.orb(b) for b in bytes_addr)
  35
  36
  37 # Unpack Ethernet Frame
  38 def ethernet_frame(data):
  39     dest_mac, src_mac, proto = struct.unpack('! 6s 6s H', data[:14])
  40     return dest_mac, src_mac, socket.htons(proto), data[14:]
  41
  42
  43 # Unpack IPv4 Packets
  44 def ipv4_packet(data):
  45     proto, src, target = struct.unpack('! 8x 1x B 2x 4s 4s', data[:20])
  46     return proto, src, target, data[20:]
  47
  48
  49 # Unpack IPv6 Packets
  50 def ipv6_packet(data):
  51     nh, src, target = struct.unpack('! 6x B 1x 16s 16s', data[:40])
  52     return nh, src, target, data[40:]
  53
  54
  55 # Unpacks any UDP Packet
  56 def udp_seg(data):
  57     src_port, dest_port, size = struct.unpack('! H H 2x H', data[:8])
  58     return src_port, dest_port, size, data[8:]
  59
  60
  61 # Unpacks any TCP Packet
  62 def tcp_seg(data):
  63     src_port, dest_port, seq, flag = struct.unpack('! H H L 4x H', data[:14])
  64     return src_port, dest_port, seq, data[((flag >> 12) * 4):]
  65
  66
  67 def receivePackets(sock, counters):
  68     # Wait for some packets on socket
  69     while True:
  70         data = sock.recv(65536)
  71
  72         # punt socket metadata
  73         # packet_desc = data[0:8]
  74
  75         # Ethernet
  76         _, _, eth_proto, data = ethernet_frame(data[8:])
  77         # Ipv4
  78         if eth_proto == 8:
  79             proto, _, _, data = ipv4_packet(data)
  80             # TCP
  81             if proto == 6:
  82                 _, dst_port, _, data = udp_seg(data)
  83             # UDP
  84             elif proto == 17:
  85                 _, dst_port, _, data = udp_seg(data)
  86                 counters[dst_port] = 0
  87         # Ipv6
  88         elif eth_proto == 0xdd86:
  89             nh, _, _, data = ipv6_packet(data)
  90             # TCP
  91             if nh == 6:
  92                 _, dst_port, _, data = udp_seg(data)
  93             # UDP
  94             elif nh == 17:
  95                 _, dst_port, _, data = udp_seg(data)
  96                 counters[dst_port] = 0
  97
  98
  99 class serverSocketThread(threading.Thread):
 100     """ Socket server thread"""
 101
 102     def __init__(self, threadID, sockName, counters):
 103         threading.Thread.__init__(self)
 104         self.threadID = threadID
 105         self.sockName = sockName
 106         self.sock = None
 107         self.counters = counters
 108
 109     def run(self):
 110         self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
 111         try:
 112             os.unlink(self.sockName)
 113         except:
 114             pass
 115         self.sock.bind(self.sockName)
 116
 117         receivePackets(self.sock, self.counters)
 118
 119
 120 class TestPuntSocket(VppTestCase):
 121     """ Punt Socket """
 122
 123     ports = [1111, 2222, 3333, 4444]
 124     sock_servers = list()
 125     portsCheck = dict()
 126     nr_packets = 256
 127
 128     @classmethod
 129     def setUpClass(cls):
 130         super(TestPuntSocket, cls).setUpClass()
 131
 132     @classmethod
 133     def tearDownClass(cls):
 134         super(TestPuntSocket, cls).tearDownClass()
 135
 136     @classmethod
 137     def setUpConstants(cls):
 138         cls.extra_vpp_punt_config = [
 139             "punt", "{", "socket", cls.tempdir+"/socket_punt", "}"]
 140         super(TestPuntSocket, cls).setUpConstants()
 141
 142     def setUp(self):
 143         super(TestPuntSocket, self).setUp()
 144         random.seed()
 145
 146         self.create_pg_interfaces(range(2))
 147         for i in self.pg_interfaces:
 148             i.admin_up()
 149
 150     def tearDown(self):
 151         del self.sock_servers[:]
 152         super(TestPuntSocket, self).tearDown()
 153
 154     def socket_client_create(self, sock_name, id=None):
 155         thread = serverSocketThread(id, sock_name, self.portsCheck)
 156         self.sock_servers.append(thread)
 157         thread.start()
 158
 159     def socket_client_close(self):
 160         for thread in self.sock_servers:
 161             thread.sock.close()
 162
 163
 164 class TestIP4PuntSocket(TestPuntSocket):
 165     """ Punt Socket for IPv4 """
 166
 167     @classmethod
 168     def setUpClass(cls):
 169         super(TestIP4PuntSocket, cls).setUpClass()
 170
 171     @classmethod
 172     def tearDownClass(cls):
 173         super(TestIP4PuntSocket, cls).tearDownClass()
 174
 175     def setUp(self):
 176         super(TestIP4PuntSocket, self).setUp()
 177
 178         for i in self.pg_interfaces:
 179             i.config_ip4()
 180             i.resolve_arp()
 181
 182     def tearDown(self):
 183         super(TestIP4PuntSocket, self).tearDown()
 184         for i in self.pg_interfaces:
 185             i.unconfig_ip4()
 186             i.admin_down()
 187
 188     def test_punt_socket_dump(self):
 189         """ Punt socket registration/deregistration"""
 190
 191         punts = self.vapi.punt_socket_dump(is_ip6=0)
 192         self.assertEqual(len(punts), 0)
 193
 194         #
 195         # configure a punt socket
 196         #
 197         self.vapi.punt_socket_register(1111, b"%s/socket_punt_1111" %
 198                                        six.ensure_binary(self.tempdir))
 199         self.vapi.punt_socket_register(2222, b"%s/socket_punt_2222" %
 200                                        six.ensure_binary(self.tempdir))
 201         punts = self.vapi.punt_socket_dump(is_ip6=0)
 202         self.assertEqual(len(punts), 2)
 203         self.assertEqual(punts[0].punt.l4_port, 1111)
 204         self.assertEqual(punts[1].punt.l4_port, 2222)
 205
 206         #
 207         # deregister a punt socket
 208         #
 209         self.vapi.punt_socket_deregister(1111)
 210         punts = self.vapi.punt_socket_dump(is_ip6=0)
 211         self.assertEqual(len(punts), 1)
 212
 213         #
 214         # configure a punt socket again
 215         #
 216         self.vapi.punt_socket_register(1111, b"%s/socket_punt_1111" %
 217                                        six.ensure_binary(self.tempdir))
 218         self.vapi.punt_socket_register(3333, b"%s/socket_punt_3333" %
 219                                        six.ensure_binary(self.tempdir))
 220         punts = self.vapi.punt_socket_dump(is_ip6=0)
 221         self.assertEqual(len(punts), 3)
 222
 223         #
 224         # deregister all punt socket
 225         #
 226         self.vapi.punt_socket_deregister(1111)
 227         self.vapi.punt_socket_deregister(2222)
 228         self.vapi.punt_socket_deregister(3333)
 229         punts = self.vapi.punt_socket_dump(is_ip6=0)
 230         self.assertEqual(len(punts), 0)
 231
 232     def test_punt_socket_traffic_single_port_single_socket(self):
 233         """ Punt socket traffic single port single socket"""
 234
 235         port = self.ports[0]
 236
 237         p = (Ether(src=self.pg0.remote_mac,
 238                    dst=self.pg0.local_mac) /
 239              IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
 240              UDP(sport=9876, dport=port) /
 241              Raw('\xa5' * 100))
 242
 243         pkts = p * self.nr_packets
 244         self.portsCheck[port] = self.nr_packets
 245
 246         punts = self.vapi.punt_socket_dump(is_ip6=0)
 247         self.assertEqual(len(punts), 0)
 248
 249         #
 250         # expect ICMP - port unreachable for all packets
 251         #
 252         self.vapi.cli("clear trace")
 253         self.pg0.add_stream(pkts)
 254         self.pg_enable_capture(self.pg_interfaces)
 255         self.pg_start()
 256         # FIXME - when punt socket deregister is implemented
 257         # rx = self.pg0.get_capture(self.nr_packets)
 258         # for p in rx:
 259         #     self.assertEqual(int(p[IP].proto), 1)   # ICMP
 260         #     self.assertEqual(int(p[ICMP].code), 3)  # unreachable
 261
 262         #
 263         # configure a punt socket
 264         #
 265         self.socket_client_create(b"%s/socket_%d" % (
 266             six.ensure_binary(self.tempdir), port))
 267         self.vapi.punt_socket_register(port, b"%s/socket_%d" % (
 268             six.ensure_binary(self.tempdir), port))
 269         punts = self.vapi.punt_socket_dump(is_ip6=0)
 270         self.assertEqual(len(punts), 1)
 271
 272         self.logger.debug("Sending %s packets to port %d",
 273                           str(self.portsCheck[port]), port)
 274         #
 275         # expect punt socket and no packets on pg0
 276         #
 277         self.vapi.cli("clear errors")
 278         self.vapi.cli("clear trace")
 279         self.pg0.add_stream(pkts)
 280         self.pg_enable_capture(self.pg_interfaces)
 281         self.pg_start()
 282         self.pg0.get_capture(0)
 283         self.logger.info(self.vapi.cli("show trace"))
 284         self.socket_client_close()
 285         self.assertEqual(self.portsCheck[port], 0)
 286
 287         #
 288         # remove punt socket. expect ICMP - port unreachable for all packets
 289         #
 290         self.vapi.punt_socket_deregister(port)
 291         punts = self.vapi.punt_socket_dump(is_ip6=0)
 292         self.assertEqual(len(punts), 0)
 293         self.pg0.add_stream(pkts)
 294         self.pg_enable_capture(self.pg_interfaces)
 295         self.pg_start()
 296         # FIXME - when punt socket deregister is implemented
 297         # self.pg0.get_capture(nr_packets)
 298
 299     def test_punt_socket_traffic_multi_port_multi_sockets(self):
 300         """ Punt socket traffic multi ports and multi sockets"""
 301
 302         for p in self.ports:
 303             self.portsCheck[p] = 0
 304
 305         #
 306         # create stream with random packets count per given ports
 307         #
 308         pkts = list()
 309         for _ in range(0, self.nr_packets):
 310             # choose port from port list
 311             p = random.choice(self.ports)
 312             pkts.append((
 313                 Ether(src=self.pg0.remote_mac,
 314                       dst=self.pg0.local_mac) /
 315                 IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
 316                 UDP(sport=9876, dport=p) /
 317                 Raw('\xa5' * 100)))
 318             self.portsCheck[p] += 1
 319         #
 320         # no punt socket
 321         #
 322         punts = self.vapi.punt_socket_dump(is_ip6=0)
 323         self.assertEqual(len(punts), 0)
 324
 325         #
 326         # configure a punt socket
 327         #
 328         for p in self.ports:
 329             self.socket_client_create(b"%s/socket_%d" % (
 330                 six.ensure_binary(self.tempdir), p))
 331             self.vapi.punt_socket_register(p, b"%s/socket_%d" % (
 332                 six.ensure_binary(self.tempdir),  p))
 333         punts = self.vapi.punt_socket_dump(is_ip6=0)
 334         self.assertEqual(len(punts), len(self.ports))
 335
 336         for p in self.ports:
 337             self.logger.debug("Sending %s packets to port %d",
 338                               str(self.portsCheck[p]), p)
 339
 340         #
 341         # expect punt socket and no packets on pg0
 342         #
 343         self.vapi.cli("clear errors")
 344         self.vapi.cli("clear trace")
 345         self.pg0.add_stream(pkts)
 346         self.pg_enable_capture(self.pg_interfaces)
 347         self.pg_start()
 348         self.pg0.get_capture(0)
 349         self.logger.info(self.vapi.cli("show trace"))
 350         self.socket_client_close()
 351
 352         for p in self.ports:
 353             self.assertEqual(self.portsCheck[p], 0)
 354             self.vapi.punt_socket_deregister(p)
 355         punts = self.vapi.punt_socket_dump(is_ip6=0)
 356         self.assertEqual(len(punts), 0)
 357
 358     def test_punt_socket_traffic_multi_ports_single_socket(self):
 359         """ Punt socket traffic multi ports and single socket"""
 360
 361         for p in self.ports:
 362             self.portsCheck[p] = 0
 363
 364         #
 365         # create stream with random packets count per given ports
 366         #
 367         pkts = list()
 368         for _ in range(0, self.nr_packets):
 369             # choose port from port list
 370             p = random.choice(self.ports)
 371             pkts.append((
 372                 Ether(src=self.pg0.remote_mac,
 373                       dst=self.pg0.local_mac) /
 374                 IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4) /
 375                 UDP(sport=9876, dport=p) /
 376                 Raw('\xa5' * 100)))
 377             self.portsCheck[p] += 1
 378
 379         #
 380         # no punt socket
 381         #
 382         punts = self.vapi.punt_socket_dump(is_ip6=0)
 383         self.assertEqual(len(punts), 0)
 384
 385         # configure a punt socket
 386         #
 387         self.socket_client_create(b"%s/socket_multi" %
 388                                   six.ensure_binary(self.tempdir))
 389         for p in self.ports:
 390             self.vapi.punt_socket_register(p,
 391                                            b"%s/socket_multi" %
 392                                            six.ensure_binary(self.tempdir))
 393         punts = self.vapi.punt_socket_dump(is_ip6=0)
 394         self.assertEqual(len(punts), len(self.ports))
 395
 396         for p in self.ports:
 397             self.logger.debug("Sending %s packets to port %d",
 398                               str(self.portsCheck[p]), p)
 399         #
 400         # expect punt socket and no packets on pg0
 401         #
 402         self.vapi.cli("clear errors")
 403         self.vapi.cli("clear trace")
 404         self.pg0.add_stream(pkts)
 405         self.pg_enable_capture(self.pg_interfaces)
 406         self.pg_start()
 407         self.pg0.get_capture(0)
 408         self.logger.info(self.vapi.cli("show trace"))
 409         self.socket_client_close()
 410
 411         for p in self.ports:
 412             self.assertEqual(self.portsCheck[p], 0)
 413             self.vapi.punt_socket_deregister(p)
 414         punts = self.vapi.punt_socket_dump(is_ip6=0)
 415         self.assertEqual(len(punts), 0)
 416
 417
 418 class TestIP6PuntSocket(TestPuntSocket):
 419     """ Punt Socket for IPv6"""
 420
 421     @classmethod
 422     def setUpClass(cls):
 423         super(TestIP6PuntSocket, cls).setUpClass()
 424
 425     @classmethod
 426     def tearDownClass(cls):
 427         super(TestIP6PuntSocket, cls).tearDownClass()
 428
 429     def setUp(self):
 430         super(TestIP6PuntSocket, self).setUp()
 431
 432         for i in self.pg_interfaces:
 433             i.config_ip6()
 434             i.resolve_ndp()
 435
 436     def tearDown(self):
 437         super(TestIP6PuntSocket, self).tearDown()
 438         for i in self.pg_interfaces:
 439             i.unconfig_ip6()
 440             i.admin_down()
 441
 442     def test_punt_socket_dump(self):
 443         """ Punt socket registration """
 444
 445         punts = self.vapi.punt_socket_dump(is_ip6=1)
 446         self.assertEqual(len(punts), 0)
 447
 448         #
 449         # configure a punt socket
 450         #
 451         self.vapi.punt_socket_register(1111, b"%s/socket_1111" %
 452                                        six.ensure_binary(self.tempdir),
 453                                        is_ip4=0)
 454         self.vapi.punt_socket_register(2222, b"%s/socket_2222" %
 455                                        six.ensure_binary(self.tempdir),
 456                                        is_ip4=0)
 457         punts = self.vapi.punt_socket_dump(is_ip6=1)
 458         self.assertEqual(len(punts), 2)
 459         self.assertEqual(punts[0].punt.l4_port, 1111)
 460         self.assertEqual(punts[1].punt.l4_port, 2222)
 461
 462         #
 463         # deregister a punt socket
 464         #
 465         self.vapi.punt_socket_deregister(1111, is_ip4=0)
 466         punts = self.vapi.punt_socket_dump(is_ip6=1)
 467         self.assertEqual(len(punts), 1)
 468
 469         #
 470         # configure a punt socket again
 471         #
 472         self.vapi.punt_socket_register(1111, b"%s/socket_1111" %
 473                                        six.ensure_binary(self.tempdir),
 474                                        is_ip4=0)
 475         punts = self.vapi.punt_socket_dump(is_ip6=1)
 476         self.assertEqual(len(punts), 2)
 477
 478         #
 479         # deregister all punt socket
 480         #
 481         self.vapi.punt_socket_deregister(1111, is_ip4=0)
 482         self.vapi.punt_socket_deregister(2222, is_ip4=0)
 483         self.vapi.punt_socket_deregister(3333, is_ip4=0)
 484         punts = self.vapi.punt_socket_dump(is_ip6=1)
 485         self.assertEqual(len(punts), 0)
 486
 487     def test_punt_socket_traffic_single_port_single_socket(self):
 488         """ Punt socket traffic single port single socket"""
 489
 490         port = self.ports[0]
 491
 492         p = (Ether(src=self.pg0.remote_mac,
 493                    dst=self.pg0.local_mac) /
 494              IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
 495              inet6.UDP(sport=9876, dport=port) /
 496              Raw('\xa5' * 100))
 497
 498         pkts = p * self.nr_packets
 499         self.portsCheck[port] = self.nr_packets
 500
 501         punts = self.vapi.punt_socket_dump(is_ip6=1)
 502         self.assertEqual(len(punts), 0)
 503
 504         #
 505         # expect ICMPv6 - destination unreachable for all packets
 506         #
 507         self.vapi.cli("clear trace")
 508         self.pg0.add_stream(pkts)
 509         self.pg_enable_capture(self.pg_interfaces)
 510         self.pg_start()
 511         # FIXME - when punt socket deregister is implemented
 512         # rx = self.pg0.get_capture(self.nr_packets)
 513         # for p in rx:
 514         #     self.assertEqual(int(p[IPv6].nh), 58)                # ICMPv6
 515         #     self.assertEqual(int(p[ICMPv6DestUnreach].code),4)  # unreachable
 516
 517         #
 518         # configure a punt socket
 519         #
 520         self.socket_client_create(b"%s/socket_%d" % (
 521             six.ensure_binary(self.tempdir), port))
 522         self.vapi.punt_socket_register(port, b"%s/socket_%d" % (
 523             six.ensure_binary(self.tempdir), port), is_ip4=0)
 524         punts = self.vapi.punt_socket_dump(is_ip6=1)
 525         self.assertEqual(len(punts), 1)
 526
 527         self.logger.debug("Sending %s packets to port %d",
 528                           str(self.portsCheck[port]), port)
 529         #
 530         # expect punt socket and no packets on pg0
 531         #
 532         self.vapi.cli("clear errors")
 533         self.vapi.cli("clear trace")
 534         self.pg0.add_stream(pkts)
 535         self.pg_enable_capture(self.pg_interfaces)
 536         self.pg_start()
 537         self.pg0.get_capture(0)
 538         self.logger.info(self.vapi.cli("show trace"))
 539         self.socket_client_close()
 540         self.assertEqual(self.portsCheck[port], 0)
 541
 542         #
 543         # remove punt socket. expect ICMP - dest. unreachable for all packets
 544         #
 545         self.vapi.punt_socket_deregister(port, is_ip4=0)
 546         punts = self.vapi.punt_socket_dump(is_ip6=1)
 547         self.assertEqual(len(punts), 0)
 548         self.pg0.add_stream(pkts)
 549         self.pg_enable_capture(self.pg_interfaces)
 550         self.pg_start()
 551         # FIXME - when punt socket deregister is implemented
 552         # self.pg0.get_capture(nr_packets)
 553
 554     def test_punt_socket_traffic_multi_port_multi_sockets(self):
 555         """ Punt socket traffic multi ports and multi sockets"""
 556
 557         for p in self.ports:
 558             self.portsCheck[p] = 0
 559
 560         #
 561         # create stream with random packets count per given ports
 562         #
 563         pkts = list()
 564         for _ in range(0, self.nr_packets):
 565             # choose port from port list
 566             p = random.choice(self.ports)
 567             pkts.append((
 568                 Ether(src=self.pg0.remote_mac,
 569                       dst=self.pg0.local_mac) /
 570                 IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
 571                 inet6.UDP(sport=9876, dport=p) /
 572                 Raw('\xa5' * 100)))
 573             self.portsCheck[p] += 1
 574         #
 575         # no punt socket
 576         #
 577         punts = self.vapi.punt_socket_dump(is_ip6=1)
 578         self.assertEqual(len(punts), 0)
 579
 580         #
 581         # configure a punt socket
 582         #
 583         for p in self.ports:
 584             self.socket_client_create(b"%s/socket_%d" % (
 585                 six.ensure_binary(self.tempdir), p))
 586             self.vapi.punt_socket_register(p, b"%s/socket_%d" % (
 587                 six.ensure_binary(self.tempdir), p), is_ip4=0)
 588         punts = self.vapi.punt_socket_dump(is_ip6=1)
 589         self.assertEqual(len(punts), len(self.ports))
 590
 591         for p in self.ports:
 592             self.logger.debug("Sending %s packets to port %d",
 593                               str(self.portsCheck[p]), p)
 594
 595         #
 596         # expect punt socket and no packets on pg0
 597         #
 598         self.vapi.cli("clear errors")
 599         self.vapi.cli("clear trace")
 600         self.pg0.add_stream(pkts)
 601         self.pg_enable_capture(self.pg_interfaces)
 602         self.pg_start()
 603         self.pg0.get_capture(0)
 604         self.logger.info(self.vapi.cli("show trace"))
 605         self.socket_client_close()
 606
 607         for p in self.ports:
 608             self.assertEqual(self.portsCheck[p], 0)
 609             self.vapi.punt_socket_deregister(p, is_ip4=0)
 610         punts = self.vapi.punt_socket_dump(is_ip6=1)
 611         self.assertEqual(len(punts), 0)
 612
 613     def test_punt_socket_traffic_multi_ports_single_socket(self):
 614         """ Punt socket traffic multi ports and single socket"""
 615
 616         for p in self.ports:
 617             self.portsCheck[p] = 0
 618
 619         #
 620         # create stream with random packets count per given ports
 621         #
 622         pkts = list()
 623         for _ in range(0, self.nr_packets):
 624             # choose port from port list
 625             p = random.choice(self.ports)
 626             pkts.append((
 627                 Ether(src=self.pg0.remote_mac,
 628                       dst=self.pg0.local_mac) /
 629                 IPv6(src=self.pg0.remote_ip6, dst=self.pg0.local_ip6) /
 630                 inet6.UDP(sport=9876, dport=p) /
 631                 Raw('\xa5' * 100)))
 632             self.portsCheck[p] += 1
 633
 634         #
 635         # no punt socket
 636         #
 637         punts = self.vapi.punt_socket_dump(is_ip6=1)
 638         self.assertEqual(len(punts), 0)
 639
 640         #
 641         # configure a punt socket
 642         #
 643         self.socket_client_create(b"%s/socket_multi" %
 644                                   six.ensure_binary(self.tempdir))
 645         for p in self.ports:
 646             self.vapi.punt_socket_register(p,
 647                                            b"%s/socket_multi" %
 648                                            six.ensure_binary(self.tempdir),
 649                                            is_ip4=0)
 650         punts = self.vapi.punt_socket_dump(is_ip6=1)
 651         self.assertEqual(len(punts), len(self.ports))
 652
 653         for p in self.ports:
 654             self.logger.debug("Send %s packets to port %d",
 655                               str(self.portsCheck[p]), p)
 656         #
 657         # expect punt socket and no packets on pg0
 658         #
 659         self.vapi.cli("clear errors")
 660         self.vapi.cli("clear trace")
 661         self.pg0.add_stream(pkts)
 662         self.pg_enable_capture(self.pg_interfaces)
 663         self.pg_start()
 664         self.pg0.get_capture(0)
 665         self.logger.info(self.vapi.cli("show trace"))
 666         self.socket_client_close()
 667
 668         for p in self.ports:
 669             self.assertEqual(self.portsCheck[p], 0)
 670             self.vapi.punt_socket_deregister(p, is_ip4=0)
 671         punts = self.vapi.punt_socket_dump(is_ip6=1)
 672         self.assertEqual(len(punts), 0)
 673
 674 if __name__ == '__main__':
 675     unittest.main(testRunner=VppTestRunner)