test/test_punt.py

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