test/framework.py

   1 #!/usr/bin/env python3
   2
   3 from __future__ import print_function
   4 import gc
   5 import logging
   6 import sys
   7 import os
   8 import select
   9 import signal
  10 import subprocess
  11 import unittest
  12 import tempfile
  13 import time
  14 import faulthandler
  15 import random
  16 import copy
  17 import psutil
  18 import platform
  19 from collections import deque
  20 from threading import Thread, Event
  21 from inspect import getdoc, isclass
  22 from traceback import format_exception
  23 from logging import FileHandler, DEBUG, Formatter
  24 from enum import Enum
  25 from abc import ABC, abstractmethod
  26 from struct import pack, unpack
  27
  28 import scapy.compat
  29 from scapy.packet import Raw
  30 import hook as hookmodule
  31 from vpp_pg_interface import VppPGInterface
  32 from vpp_sub_interface import VppSubInterface
  33 from vpp_lo_interface import VppLoInterface
  34 from vpp_bvi_interface import VppBviInterface
  35 from vpp_papi_provider import VppPapiProvider
  36 from vpp_papi import VppEnum
  37 import vpp_papi
  38 from vpp_papi.vpp_stats import VPPStats
  39 from vpp_papi.vpp_transport_socket import VppTransportSocketIOError
  40 from log import RED, GREEN, YELLOW, double_line_delim, single_line_delim, \
  41     get_logger, colorize
  42 from vpp_object import VppObjectRegistry
  43 from util import ppp, is_core_present
  44 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
  45 from scapy.layers.inet6 import ICMPv6DestUnreach, ICMPv6EchoRequest
  46 from scapy.layers.inet6 import ICMPv6EchoReply
  47
  48 from cpu_config import available_cpus, num_cpus, max_vpp_cpus
  49
  50 logger = logging.getLogger(__name__)
  51
  52 # Set up an empty logger for the testcase that can be overridden as necessary
  53 null_logger = logging.getLogger('VppTestCase')
  54 null_logger.addHandler(logging.NullHandler())
  55
  56 PASS = 0
  57 FAIL = 1
  58 ERROR = 2
  59 SKIP = 3
  60 TEST_RUN = 4
  61 SKIP_CPU_SHORTAGE = 5
  62
  63
  64 class BoolEnvironmentVariable(object):
  65
  66     def __init__(self, env_var_name, default='n', true_values=None):
  67         self.name = env_var_name
  68         self.default = default
  69         self.true_values = true_values if true_values is not None else \
  70             ("y", "yes", "1")
  71
  72     def __bool__(self):
  73         return os.getenv(self.name, self.default).lower() in self.true_values
  74
  75     if sys.version_info[0] == 2:
  76         __nonzero__ = __bool__
  77
  78     def __repr__(self):
  79         return 'BoolEnvironmentVariable(%r, default=%r, true_values=%r)' % \
  80                (self.name, self.default, self.true_values)
  81
  82
  83 debug_framework = BoolEnvironmentVariable('TEST_DEBUG')
  84 if debug_framework:
  85     import debug_internal
  86
  87 """
  88   Test framework module.
  89
  90   The module provides a set of tools for constructing and running tests and
  91   representing the results.
  92 """
  93
  94
  95 class VppDiedError(Exception):
  96     """ exception for reporting that the subprocess has died."""
  97
  98     signals_by_value = {v: k for k, v in signal.__dict__.items() if
  99                         k.startswith('SIG') and not k.startswith('SIG_')}
 100
 101     def __init__(self, rv=None, testcase=None, method_name=None):
 102         self.rv = rv
 103         self.signal_name = None
 104         self.testcase = testcase
 105         self.method_name = method_name
 106
 107         try:
 108             self.signal_name = VppDiedError.signals_by_value[-rv]
 109         except (KeyError, TypeError):
 110             pass
 111
 112         if testcase is None and method_name is None:
 113             in_msg = ''
 114         else:
 115             in_msg = ' while running %s.%s' % (testcase, method_name)
 116
 117         if self.rv:
 118             msg = "VPP subprocess died unexpectedly%s with return code: %d%s."\
 119                 % (in_msg, self.rv, ' [%s]' %
 120                    (self.signal_name if
 121                     self.signal_name is not None else ''))
 122         else:
 123             msg = "VPP subprocess died unexpectedly%s." % in_msg
 124
 125         super(VppDiedError, self).__init__(msg)
 126
 127
 128 class _PacketInfo(object):
 129     """Private class to create packet info object.
 130
 131     Help process information about the next packet.
 132     Set variables to default values.
 133     """
 134     #: Store the index of the packet.
 135     index = -1
 136     #: Store the index of the source packet generator interface of the packet.
 137     src = -1
 138     #: Store the index of the destination packet generator interface
 139     #: of the packet.
 140     dst = -1
 141     #: Store expected ip version
 142     ip = -1
 143     #: Store expected upper protocol
 144     proto = -1
 145     #: Store the copy of the former packet.
 146     data = None
 147
 148     def __eq__(self, other):
 149         index = self.index == other.index
 150         src = self.src == other.src
 151         dst = self.dst == other.dst
 152         data = self.data == other.data
 153         return index and src and dst and data
 154
 155
 156 def pump_output(testclass):
 157     """ pump output from vpp stdout/stderr to proper queues """
 158     stdout_fragment = ""
 159     stderr_fragment = ""
 160     while not testclass.pump_thread_stop_flag.is_set():
 161         readable = select.select([testclass.vpp.stdout.fileno(),
 162                                   testclass.vpp.stderr.fileno(),
 163                                   testclass.pump_thread_wakeup_pipe[0]],
 164                                  [], [])[0]
 165         if testclass.vpp.stdout.fileno() in readable:
 166             read = os.read(testclass.vpp.stdout.fileno(), 102400)
 167             if len(read) > 0:
 168                 split = read.decode('ascii',
 169                                     errors='backslashreplace').splitlines(True)
 170                 if len(stdout_fragment) > 0:
 171                     split[0] = "%s%s" % (stdout_fragment, split[0])
 172                 if len(split) > 0 and split[-1].endswith("\n"):
 173                     limit = None
 174                 else:
 175                     limit = -1
 176                     stdout_fragment = split[-1]
 177                 testclass.vpp_stdout_deque.extend(split[:limit])
 178                 if not testclass.cache_vpp_output:
 179                     for line in split[:limit]:
 180                         testclass.logger.info(
 181                             "VPP STDOUT: %s" % line.rstrip("\n"))
 182         if testclass.vpp.stderr.fileno() in readable:
 183             read = os.read(testclass.vpp.stderr.fileno(), 102400)
 184             if len(read) > 0:
 185                 split = read.decode('ascii',
 186                                     errors='backslashreplace').splitlines(True)
 187                 if len(stderr_fragment) > 0:
 188                     split[0] = "%s%s" % (stderr_fragment, split[0])
 189                 if len(split) > 0 and split[-1].endswith("\n"):
 190                     limit = None
 191                 else:
 192                     limit = -1
 193                     stderr_fragment = split[-1]
 194
 195                 testclass.vpp_stderr_deque.extend(split[:limit])
 196                 if not testclass.cache_vpp_output:
 197                     for line in split[:limit]:
 198                         testclass.logger.error(
 199                             "VPP STDERR: %s" % line.rstrip("\n"))
 200                         # ignoring the dummy pipe here intentionally - the
 201                         # flag will take care of properly terminating the loop
 202
 203
 204 def _is_skip_aarch64_set():
 205     return BoolEnvironmentVariable('SKIP_AARCH64')
 206
 207
 208 is_skip_aarch64_set = _is_skip_aarch64_set()
 209
 210
 211 def _is_platform_aarch64():
 212     return platform.machine() == 'aarch64'
 213
 214
 215 is_platform_aarch64 = _is_platform_aarch64()
 216
 217
 218 def _running_extended_tests():
 219     return BoolEnvironmentVariable("EXTENDED_TESTS")
 220
 221
 222 running_extended_tests = _running_extended_tests()
 223
 224
 225 def _running_gcov_tests():
 226     return BoolEnvironmentVariable("GCOV_TESTS")
 227
 228
 229 running_gcov_tests = _running_gcov_tests()
 230
 231
 232 def get_environ_vpp_worker_count():
 233     worker_config = os.getenv("VPP_WORKER_CONFIG", None)
 234     if worker_config:
 235         elems = worker_config.split(" ")
 236         if elems[0] != "workers" or len(elems) != 2:
 237             raise ValueError("Wrong VPP_WORKER_CONFIG == '%s' value." %
 238                              worker_config)
 239         return int(elems[1])
 240     else:
 241         return 0
 242
 243
 244 environ_vpp_worker_count = get_environ_vpp_worker_count()
 245
 246
 247 class KeepAliveReporter(object):
 248     """
 249     Singleton object which reports test start to parent process
 250     """
 251     _shared_state = {}
 252
 253     def __init__(self):
 254         self.__dict__ = self._shared_state
 255         self._pipe = None
 256
 257     @property
 258     def pipe(self):
 259         return self._pipe
 260
 261     @pipe.setter
 262     def pipe(self, pipe):
 263         if self._pipe is not None:
 264             raise Exception("Internal error - pipe should only be set once.")
 265         self._pipe = pipe
 266
 267     def send_keep_alive(self, test, desc=None):
 268         """
 269         Write current test tmpdir & desc to keep-alive pipe to signal liveness
 270         """
 271         if self.pipe is None:
 272             # if not running forked..
 273             return
 274
 275         if isclass(test):
 276             desc = '%s (%s)' % (desc, unittest.util.strclass(test))
 277         else:
 278             desc = test.id()
 279
 280         self.pipe.send((desc, test.vpp_bin, test.tempdir, test.vpp.pid))
 281
 282
 283 class TestCaseTag(Enum):
 284     # marks the suites that must run at the end
 285     # using only a single test runner
 286     RUN_SOLO = 1
 287     # marks the suites broken on VPP multi-worker
 288     FIXME_VPP_WORKERS = 2
 289
 290
 291 def create_tag_decorator(e):
 292     def decorator(cls):
 293         try:
 294             cls.test_tags.append(e)
 295         except AttributeError:
 296             cls.test_tags = [e]
 297         return cls
 298     return decorator
 299
 300
 301 tag_run_solo = create_tag_decorator(TestCaseTag.RUN_SOLO)
 302 tag_fixme_vpp_workers = create_tag_decorator(TestCaseTag.FIXME_VPP_WORKERS)
 303
 304
 305 class DummyVpp:
 306     returncode = None
 307     pid = 0xcafebafe
 308
 309     def poll(self):
 310         pass
 311
 312     def terminate(self):
 313         pass
 314
 315
 316 class CPUInterface(ABC):
 317     cpus = []
 318     skipped_due_to_cpu_lack = False
 319
 320     @classmethod
 321     @abstractmethod
 322     def get_cpus_required(cls):
 323         pass
 324
 325     @classmethod
 326     def assign_cpus(cls, cpus):
 327         cls.cpus = cpus
 328
 329
 330 class VppTestCase(CPUInterface, unittest.TestCase):
 331     """This subclass is a base class for VPP test cases that are implemented as
 332     classes. It provides methods to create and run test case.
 333     """
 334
 335     extra_vpp_statseg_config = ""
 336     extra_vpp_punt_config = []
 337     extra_vpp_plugin_config = []
 338     logger = null_logger
 339     vapi_response_timeout = 5
 340
 341     @property
 342     def packet_infos(self):
 343         """List of packet infos"""
 344         return self._packet_infos
 345
 346     @classmethod
 347     def get_packet_count_for_if_idx(cls, dst_if_index):
 348         """Get the number of packet info for specified destination if index"""
 349         if dst_if_index in cls._packet_count_for_dst_if_idx:
 350             return cls._packet_count_for_dst_if_idx[dst_if_index]
 351         else:
 352             return 0
 353
 354     @classmethod
 355     def has_tag(cls, tag):
 356         """ if the test case has a given tag - return true """
 357         try:
 358             return tag in cls.test_tags
 359         except AttributeError:
 360             pass
 361         return False
 362
 363     @classmethod
 364     def is_tagged_run_solo(cls):
 365         """ if the test case class is timing-sensitive - return true """
 366         return cls.has_tag(TestCaseTag.RUN_SOLO)
 367
 368     @classmethod
 369     def instance(cls):
 370         """Return the instance of this testcase"""
 371         return cls.test_instance
 372
 373     @classmethod
 374     def set_debug_flags(cls, d):
 375         cls.gdbserver_port = 7777
 376         cls.debug_core = False
 377         cls.debug_gdb = False
 378         cls.debug_gdbserver = False
 379         cls.debug_all = False
 380         cls.debug_attach = False
 381         if d is None:
 382             return
 383         dl = d.lower()
 384         if dl == "core":
 385             cls.debug_core = True
 386         elif dl == "gdb" or dl == "gdb-all":
 387             cls.debug_gdb = True
 388         elif dl == "gdbserver" or dl == "gdbserver-all":
 389             cls.debug_gdbserver = True
 390         elif dl == "attach":
 391             cls.debug_attach = True
 392         else:
 393             raise Exception("Unrecognized DEBUG option: '%s'" % d)
 394         if dl == "gdb-all" or dl == "gdbserver-all":
 395             cls.debug_all = True
 396
 397     @classmethod
 398     def get_vpp_worker_count(cls):
 399         if not hasattr(cls, "vpp_worker_count"):
 400             if cls.has_tag(TestCaseTag.FIXME_VPP_WORKERS):
 401                 cls.vpp_worker_count = 0
 402             else:
 403                 cls.vpp_worker_count = environ_vpp_worker_count
 404         return cls.vpp_worker_count
 405
 406     @classmethod
 407     def get_cpus_required(cls):
 408         return 1 + cls.get_vpp_worker_count()
 409
 410     @classmethod
 411     def setUpConstants(cls):
 412         """ Set-up the test case class based on environment variables """
 413         cls.step = BoolEnvironmentVariable('STEP')
 414         # inverted case to handle '' == True
 415         c = os.getenv("CACHE_OUTPUT", "1")
 416         cls.cache_vpp_output = False if c.lower() in ("n", "no", "0") else True
 417         cls.vpp_bin = os.getenv('VPP_BIN', "vpp")
 418         extern_plugin_path = os.getenv('EXTERN_PLUGINS')
 419         debug_cli = ""
 420         if cls.step or cls.debug_gdb or cls.debug_gdbserver:
 421             debug_cli = "cli-listen localhost:5002"
 422         coredump_size = None
 423         size = os.getenv("COREDUMP_SIZE")
 424         if size is not None:
 425             coredump_size = "coredump-size %s" % size
 426         if coredump_size is None:
 427             coredump_size = "coredump-size unlimited"
 428
 429         default_variant = os.getenv("VARIANT")
 430         if default_variant is not None:
 431             default_variant = "defaults { %s 100 }" % default_variant
 432         else:
 433             default_variant = ""
 434
 435         api_fuzzing = os.getenv("API_FUZZ")
 436         if api_fuzzing is None:
 437             api_fuzzing = 'off'
 438
 439         cls.vpp_cmdline = [
 440             cls.vpp_bin,
 441             "unix", "{", "nodaemon", debug_cli, "full-coredump",
 442             coredump_size, "runtime-dir", cls.tempdir, "}",
 443             "api-trace", "{", "on", "}",
 444             "api-segment", "{", "prefix", cls.get_api_segment_prefix(), "}",
 445             "cpu", "{", "main-core", str(cls.cpus[0]), ]
 446         if extern_plugin_path is not None:
 447             cls.extra_vpp_plugin_config.append(
 448                 "add-path %s" % extern_plugin_path)
 449         if cls.get_vpp_worker_count():
 450             cls.vpp_cmdline.extend([
 451                 "corelist-workers", ",".join([str(x) for x in cls.cpus[1:]])])
 452         cls.vpp_cmdline.extend([
 453             "}",
 454             "physmem", "{", "max-size", "32m", "}",
 455             "statseg", "{", "socket-name", cls.get_stats_sock_path(),
 456             cls.extra_vpp_statseg_config, "}",
 457             "socksvr", "{", "socket-name", cls.get_api_sock_path(), "}",
 458             "node { ", default_variant, "}",
 459             "api-fuzz {", api_fuzzing, "}",
 460             "plugins", "{", "plugin", "dpdk_plugin.so", "{", "disable", "}",
 461             "plugin", "rdma_plugin.so", "{", "disable", "}",
 462             "plugin", "lisp_unittest_plugin.so", "{", "enable", "}",
 463             "plugin", "unittest_plugin.so", "{", "enable", "}"
 464         ] + cls.extra_vpp_plugin_config + ["}", ])
 465
 466         if cls.extra_vpp_punt_config is not None:
 467             cls.vpp_cmdline.extend(cls.extra_vpp_punt_config)
 468
 469         if not cls.debug_attach:
 470             cls.logger.info("vpp_cmdline args: %s" % cls.vpp_cmdline)
 471             cls.logger.info("vpp_cmdline: %s" % " ".join(cls.vpp_cmdline))
 472
 473     @classmethod
 474     def wait_for_enter(cls):
 475         if cls.debug_gdbserver:
 476             print(double_line_delim)
 477             print("Spawned GDB server with PID: %d" % cls.vpp.pid)
 478         elif cls.debug_gdb:
 479             print(double_line_delim)
 480             print("Spawned VPP with PID: %d" % cls.vpp.pid)
 481         else:
 482             cls.logger.debug("Spawned VPP with PID: %d" % cls.vpp.pid)
 483             return
 484         print(single_line_delim)
 485         print("You can debug VPP using:")
 486         if cls.debug_gdbserver:
 487             print("sudo gdb " + cls.vpp_bin +
 488                   " -ex 'target remote localhost:{port}'"
 489                   .format(port=cls.gdbserver_port))
 490             print("Now is the time to attach gdb by running the above "
 491                   "command, set up breakpoints etc., then resume VPP from "
 492                   "within gdb by issuing the 'continue' command")
 493             cls.gdbserver_port += 1
 494         elif cls.debug_gdb:
 495             print("sudo gdb " + cls.vpp_bin + " -ex 'attach %s'" % cls.vpp.pid)
 496             print("Now is the time to attach gdb by running the above "
 497                   "command and set up breakpoints etc., then resume VPP from"
 498                   " within gdb by issuing the 'continue' command")
 499         print(single_line_delim)
 500         input("Press ENTER to continue running the testcase...")
 501
 502     @classmethod
 503     def attach_vpp(cls):
 504         cls.vpp = DummyVpp()
 505
 506     @classmethod
 507     def run_vpp(cls):
 508         cls.logger.debug(f"Assigned cpus: {cls.cpus}")
 509         cmdline = cls.vpp_cmdline
 510
 511         if cls.debug_gdbserver:
 512             gdbserver = '/usr/bin/gdbserver'
 513             if not os.path.isfile(gdbserver) or\
 514                     not os.access(gdbserver, os.X_OK):
 515                 raise Exception("gdbserver binary '%s' does not exist or is "
 516                                 "not executable" % gdbserver)
 517
 518             cmdline = [gdbserver, 'localhost:{port}'
 519                        .format(port=cls.gdbserver_port)] + cls.vpp_cmdline
 520             cls.logger.info("Gdbserver cmdline is %s", " ".join(cmdline))
 521
 522         try:
 523             cls.vpp = subprocess.Popen(cmdline,
 524                                        stdout=subprocess.PIPE,
 525                                        stderr=subprocess.PIPE)
 526         except subprocess.CalledProcessError as e:
 527             cls.logger.critical("Subprocess returned with non-0 return code: ("
 528                                 "%s)", e.returncode)
 529             raise
 530         except OSError as e:
 531             cls.logger.critical("Subprocess returned with OS error: "
 532                                 "(%s) %s", e.errno, e.strerror)
 533             raise
 534         except Exception as e:
 535             cls.logger.exception("Subprocess returned unexpected from "
 536                                  "%s:", cmdline)
 537             raise
 538
 539         cls.wait_for_enter()
 540
 541     @classmethod
 542     def wait_for_coredump(cls):
 543         corefile = cls.tempdir + "/core"
 544         if os.path.isfile(corefile):
 545             cls.logger.error("Waiting for coredump to complete: %s", corefile)
 546             curr_size = os.path.getsize(corefile)
 547             deadline = time.time() + 60
 548             ok = False
 549             while time.time() < deadline:
 550                 cls.sleep(1)
 551                 size = curr_size
 552                 curr_size = os.path.getsize(corefile)
 553                 if size == curr_size:
 554                     ok = True
 555                     break
 556             if not ok:
 557                 cls.logger.error("Timed out waiting for coredump to complete:"
 558                                  " %s", corefile)
 559             else:
 560                 cls.logger.error("Coredump complete: %s, size %d",
 561                                  corefile, curr_size)
 562
 563     @classmethod
 564     def get_stats_sock_path(cls):
 565         return "%s/stats.sock" % cls.tempdir
 566
 567     @classmethod
 568     def get_api_sock_path(cls):
 569         return "%s/api.sock" % cls.tempdir
 570
 571     @classmethod
 572     def get_api_segment_prefix(cls):
 573         return os.path.basename(cls.tempdir)  # Only used for VAPI
 574
 575     @classmethod
 576     def get_tempdir(cls):
 577         if cls.debug_attach:
 578             return os.getenv("VPP_IN_GDB_TMP_DIR",
 579                              "/tmp/unittest-attach-gdb")
 580         else:
 581             return tempfile.mkdtemp(prefix='vpp-unittest-%s-' % cls.__name__)
 582
 583     @classmethod
 584     def setUpClass(cls):
 585         """
 586         Perform class setup before running the testcase
 587         Remove shared memory files, start vpp and connect the vpp-api
 588         """
 589         super(VppTestCase, cls).setUpClass()
 590         cls.logger = get_logger(cls.__name__)
 591         seed = os.environ["RND_SEED"]
 592         random.seed(seed)
 593         if hasattr(cls, 'parallel_handler'):
 594             cls.logger.addHandler(cls.parallel_handler)
 595             cls.logger.propagate = False
 596         d = os.getenv("DEBUG", None)
 597         cls.set_debug_flags(d)
 598         cls.tempdir = cls.get_tempdir()
 599         cls.file_handler = FileHandler("%s/log.txt" % cls.tempdir)
 600         cls.file_handler.setFormatter(
 601             Formatter(fmt='%(asctime)s,%(msecs)03d %(message)s',
 602                       datefmt="%H:%M:%S"))
 603         cls.file_handler.setLevel(DEBUG)
 604         cls.logger.addHandler(cls.file_handler)
 605         cls.logger.debug("--- setUpClass() for %s called ---" % cls.__name__)
 606         os.chdir(cls.tempdir)
 607         cls.logger.info("Temporary dir is %s, api socket is %s",
 608                         cls.tempdir, cls.get_api_sock_path())
 609         cls.logger.debug("Random seed is %s", seed)
 610         cls.setUpConstants()
 611         cls.reset_packet_infos()
 612         cls._pcaps = []
 613         cls._old_pcaps = []
 614         cls.verbose = 0
 615         cls.vpp_dead = False
 616         cls.registry = VppObjectRegistry()
 617         cls.vpp_startup_failed = False
 618         cls.reporter = KeepAliveReporter()
 619         # need to catch exceptions here because if we raise, then the cleanup
 620         # doesn't get called and we might end with a zombie vpp
 621         try:
 622             if cls.debug_attach:
 623                 cls.attach_vpp()
 624             else:
 625                 cls.run_vpp()
 626             cls.reporter.send_keep_alive(cls, 'setUpClass')
 627             VppTestResult.current_test_case_info = TestCaseInfo(
 628                 cls.logger, cls.tempdir, cls.vpp.pid, cls.vpp_bin)
 629             cls.vpp_stdout_deque = deque()
 630             cls.vpp_stderr_deque = deque()
 631             if not cls.debug_attach:
 632                 cls.pump_thread_stop_flag = Event()
 633                 cls.pump_thread_wakeup_pipe = os.pipe()
 634                 cls.pump_thread = Thread(target=pump_output, args=(cls,))
 635                 cls.pump_thread.daemon = True
 636                 cls.pump_thread.start()
 637             if cls.debug_gdb or cls.debug_gdbserver or cls.debug_attach:
 638                 cls.vapi_response_timeout = 0
 639             cls.vapi = VppPapiProvider(cls.__name__, cls,
 640                                        cls.vapi_response_timeout)
 641             if cls.step:
 642                 hook = hookmodule.StepHook(cls)
 643             else:
 644                 hook = hookmodule.PollHook(cls)
 645             cls.vapi.register_hook(hook)
 646             cls.statistics = VPPStats(socketname=cls.get_stats_sock_path())
 647             try:
 648                 hook.poll_vpp()
 649             except VppDiedError:
 650                 cls.vpp_startup_failed = True
 651                 cls.logger.critical(
 652                     "VPP died shortly after startup, check the"
 653                     " output to standard error for possible cause")
 654                 raise
 655             try:
 656                 cls.vapi.connect()
 657             except (vpp_papi.VPPIOError, Exception) as e:
 658                 cls.logger.debug("Exception connecting to vapi: %s" % e)
 659                 cls.vapi.disconnect()
 660
 661                 if cls.debug_gdbserver:
 662                     print(colorize("You're running VPP inside gdbserver but "
 663                                    "VPP-API connection failed, did you forget "
 664                                    "to 'continue' VPP from within gdb?", RED))
 665                 raise e
 666             if cls.debug_attach:
 667                 last_line = cls.vapi.cli("show thread").split("\n")[-2]
 668                 cls.vpp_worker_count = int(last_line.split(" ")[0])
 669                 print("Detected VPP with %s workers." % cls.vpp_worker_count)
 670         except vpp_papi.VPPRuntimeError as e:
 671             cls.logger.debug("%s" % e)
 672             cls.quit()
 673             raise e
 674         except Exception as e:
 675             cls.logger.debug("Exception connecting to VPP: %s" % e)
 676             cls.quit()
 677             raise e
 678
 679     @classmethod
 680     def _debug_quit(cls):
 681         if (cls.debug_gdbserver or cls.debug_gdb):
 682             try:
 683                 cls.vpp.poll()
 684
 685                 if cls.vpp.returncode is None:
 686                     print()
 687                     print(double_line_delim)
 688                     print("VPP or GDB server is still running")
 689                     print(single_line_delim)
 690                     input("When done debugging, press ENTER to kill the "
 691                           "process and finish running the testcase...")
 692             except AttributeError:
 693                 pass
 694
 695     @classmethod
 696     def quit(cls):
 697         """
 698         Disconnect vpp-api, kill vpp and cleanup shared memory files
 699         """
 700         cls._debug_quit()
 701
 702         # first signal that we want to stop the pump thread, then wake it up
 703         if hasattr(cls, 'pump_thread_stop_flag'):
 704             cls.pump_thread_stop_flag.set()
 705         if hasattr(cls, 'pump_thread_wakeup_pipe'):
 706             os.write(cls.pump_thread_wakeup_pipe[1], b'ding dong wake up')
 707         if hasattr(cls, 'pump_thread'):
 708             cls.logger.debug("Waiting for pump thread to stop")
 709             cls.pump_thread.join()
 710         if hasattr(cls, 'vpp_stderr_reader_thread'):
 711             cls.logger.debug("Waiting for stderr pump to stop")
 712             cls.vpp_stderr_reader_thread.join()
 713
 714         if hasattr(cls, 'vpp'):
 715             if hasattr(cls, 'vapi'):
 716                 cls.logger.debug(cls.vapi.vpp.get_stats())
 717                 cls.logger.debug("Disconnecting class vapi client on %s",
 718                                  cls.__name__)
 719                 cls.vapi.disconnect()
 720                 cls.logger.debug("Deleting class vapi attribute on %s",
 721                                  cls.__name__)
 722                 del cls.vapi
 723             cls.vpp.poll()
 724             if not cls.debug_attach and cls.vpp.returncode is None:
 725                 cls.wait_for_coredump()
 726                 cls.logger.debug("Sending TERM to vpp")
 727                 cls.vpp.terminate()
 728                 cls.logger.debug("Waiting for vpp to die")
 729                 try:
 730                     outs, errs = cls.vpp.communicate(timeout=5)
 731                 except subprocess.TimeoutExpired:
 732                     cls.vpp.kill()
 733                     outs, errs = cls.vpp.communicate()
 734             cls.logger.debug("Deleting class vpp attribute on %s",
 735                              cls.__name__)
 736             if not cls.debug_attach:
 737                 cls.vpp.stdout.close()
 738                 cls.vpp.stderr.close()
 739             del cls.vpp
 740
 741         if cls.vpp_startup_failed:
 742             stdout_log = cls.logger.info
 743             stderr_log = cls.logger.critical
 744         else:
 745             stdout_log = cls.logger.info
 746             stderr_log = cls.logger.info
 747
 748         if hasattr(cls, 'vpp_stdout_deque'):
 749             stdout_log(single_line_delim)
 750             stdout_log('VPP output to stdout while running %s:', cls.__name__)
 751             stdout_log(single_line_delim)
 752             vpp_output = "".join(cls.vpp_stdout_deque)
 753             with open(cls.tempdir + '/vpp_stdout.txt', 'w') as f:
 754                 f.write(vpp_output)
 755             stdout_log('\n%s', vpp_output)
 756             stdout_log(single_line_delim)
 757
 758         if hasattr(cls, 'vpp_stderr_deque'):
 759             stderr_log(single_line_delim)
 760             stderr_log('VPP output to stderr while running %s:', cls.__name__)
 761             stderr_log(single_line_delim)
 762             vpp_output = "".join(cls.vpp_stderr_deque)
 763             with open(cls.tempdir + '/vpp_stderr.txt', 'w') as f:
 764                 f.write(vpp_output)
 765             stderr_log('\n%s', vpp_output)
 766             stderr_log(single_line_delim)
 767
 768     @classmethod
 769     def tearDownClass(cls):
 770         """ Perform final cleanup after running all tests in this test-case """
 771         cls.logger.debug("--- tearDownClass() for %s called ---" %
 772                          cls.__name__)
 773         cls.reporter.send_keep_alive(cls, 'tearDownClass')
 774         cls.quit()
 775         cls.file_handler.close()
 776         cls.reset_packet_infos()
 777         if debug_framework:
 778             debug_internal.on_tear_down_class(cls)
 779
 780     def show_commands_at_teardown(self):
 781         """ Allow subclass specific teardown logging additions."""
 782         self.logger.info("--- No test specific show commands provided. ---")
 783
 784     def tearDown(self):
 785         """ Show various debug prints after each test """
 786         self.logger.debug("--- tearDown() for %s.%s(%s) called ---" %
 787                           (self.__class__.__name__, self._testMethodName,
 788                            self._testMethodDoc))
 789
 790         try:
 791             if not self.vpp_dead:
 792                 self.logger.debug(self.vapi.cli("show trace max 1000"))
 793                 self.logger.info(self.vapi.ppcli("show interface"))
 794                 self.logger.info(self.vapi.ppcli("show hardware"))
 795                 self.logger.info(self.statistics.set_errors_str())
 796                 self.logger.info(self.vapi.ppcli("show run"))
 797                 self.logger.info(self.vapi.ppcli("show log"))
 798                 self.logger.info(self.vapi.ppcli("show bihash"))
 799                 self.logger.info("Logging testcase specific show commands.")
 800                 self.show_commands_at_teardown()
 801                 self.registry.remove_vpp_config(self.logger)
 802             # Save/Dump VPP api trace log
 803             m = self._testMethodName
 804             api_trace = "vpp_api_trace.%s.%d.log" % (m, self.vpp.pid)
 805             tmp_api_trace = "/tmp/%s" % api_trace
 806             vpp_api_trace_log = "%s/%s" % (self.tempdir, api_trace)
 807             self.logger.info(self.vapi.ppcli("api trace save %s" % api_trace))
 808             self.logger.info("Moving %s to %s\n" % (tmp_api_trace,
 809                                                     vpp_api_trace_log))
 810             os.rename(tmp_api_trace, vpp_api_trace_log)
 811         except VppTransportSocketIOError:
 812             self.logger.debug("VppTransportSocketIOError: Vpp dead. "
 813                               "Cannot log show commands.")
 814             self.vpp_dead = True
 815         else:
 816             self.registry.unregister_all(self.logger)
 817
 818     def setUp(self):
 819         """ Clear trace before running each test"""
 820         super(VppTestCase, self).setUp()
 821         self.reporter.send_keep_alive(self)
 822         if self.vpp_dead:
 823             raise VppDiedError(rv=None, testcase=self.__class__.__name__,
 824                                method_name=self._testMethodName)
 825         self.sleep(.1, "during setUp")
 826         self.vpp_stdout_deque.append(
 827             "--- test setUp() for %s.%s(%s) starts here ---\n" %
 828             (self.__class__.__name__, self._testMethodName,
 829              self._testMethodDoc))
 830         self.vpp_stderr_deque.append(
 831             "--- test setUp() for %s.%s(%s) starts here ---\n" %
 832             (self.__class__.__name__, self._testMethodName,
 833              self._testMethodDoc))
 834         self.vapi.cli("clear trace")
 835         # store the test instance inside the test class - so that objects
 836         # holding the class can access instance methods (like assertEqual)
 837         type(self).test_instance = self
 838
 839     @classmethod
 840     def pg_enable_capture(cls, interfaces=None):
 841         """
 842         Enable capture on packet-generator interfaces
 843
 844         :param interfaces: iterable interface indexes (if None,
 845                            use self.pg_interfaces)
 846
 847         """
 848         if interfaces is None:
 849             interfaces = cls.pg_interfaces
 850         for i in interfaces:
 851             i.enable_capture()
 852
 853     @classmethod
 854     def register_pcap(cls, intf, worker):
 855         """ Register a pcap in the testclass """
 856         # add to the list of captures with current timestamp
 857         cls._pcaps.append((intf, worker))
 858
 859     @classmethod
 860     def get_vpp_time(cls):
 861         # processes e.g. "Time now 2.190522, Wed, 11 Mar 2020 17:29:54 GMT"
 862         # returns float("2.190522")
 863         timestr = cls.vapi.cli('show clock')
 864         head, sep, tail = timestr.partition(',')
 865         head, sep, tail = head.partition('Time now')
 866         return float(tail)
 867
 868     @classmethod
 869     def sleep_on_vpp_time(cls, sec):
 870         """ Sleep according to time in VPP world """
 871         # On a busy system with many processes
 872         # we might end up with VPP time being slower than real world
 873         # So take that into account when waiting for VPP to do something
 874         start_time = cls.get_vpp_time()
 875         while cls.get_vpp_time() - start_time < sec:
 876             cls.sleep(0.1)
 877
 878     @classmethod
 879     def pg_start(cls, trace=True):
 880         """ Enable the PG, wait till it is done, then clean up """
 881         for (intf, worker) in cls._old_pcaps:
 882             intf.rename_old_pcap_file(intf.get_in_path(worker),
 883                                       intf.in_history_counter)
 884         cls._old_pcaps = []
 885         if trace:
 886             cls.vapi.cli("clear trace")
 887             cls.vapi.cli("trace add pg-input 1000")
 888         cls.vapi.cli('packet-generator enable')
 889         # PG, when starts, runs to completion -
 890         # so let's avoid a race condition,
 891         # and wait a little till it's done.
 892         # Then clean it up  - and then be gone.
 893         deadline = time.time() + 300
 894         while cls.vapi.cli('show packet-generator').find("Yes") != -1:
 895             cls.sleep(0.01)  # yield
 896             if time.time() > deadline:
 897                 cls.logger.error("Timeout waiting for pg to stop")
 898                 break
 899         for intf, worker in cls._pcaps:
 900             cls.vapi.cli('packet-generator delete %s' %
 901                          intf.get_cap_name(worker))
 902         cls._old_pcaps = cls._pcaps
 903         cls._pcaps = []
 904
 905     @classmethod
 906     def create_pg_interfaces_internal(cls, interfaces, gso=0, gso_size=0,
 907                                       mode=None):
 908         """
 909         Create packet-generator interfaces.
 910
 911         :param interfaces: iterable indexes of the interfaces.
 912         :returns: List of created interfaces.
 913
 914         """
 915         result = []
 916         for i in interfaces:
 917             intf = VppPGInterface(cls, i, gso, gso_size, mode)
 918             setattr(cls, intf.name, intf)
 919             result.append(intf)
 920         cls.pg_interfaces = result
 921         return result
 922
 923     @classmethod
 924     def create_pg_ip4_interfaces(cls, interfaces, gso=0, gso_size=0):
 925         pgmode = VppEnum.vl_api_pg_interface_mode_t
 926         return cls.create_pg_interfaces_internal(interfaces, gso, gso_size,
 927                                                  pgmode.PG_API_MODE_IP4)
 928
 929     @classmethod
 930     def create_pg_ip6_interfaces(cls, interfaces, gso=0, gso_size=0):
 931         pgmode = VppEnum.vl_api_pg_interface_mode_t
 932         return cls.create_pg_interfaces_internal(interfaces, gso, gso_size,
 933                                                  pgmode.PG_API_MODE_IP6)
 934
 935     @classmethod
 936     def create_pg_interfaces(cls, interfaces, gso=0, gso_size=0):
 937         pgmode = VppEnum.vl_api_pg_interface_mode_t
 938         return cls.create_pg_interfaces_internal(interfaces, gso, gso_size,
 939                                                  pgmode.PG_API_MODE_ETHERNET)
 940
 941     @classmethod
 942     def create_pg_ethernet_interfaces(cls, interfaces, gso=0, gso_size=0):
 943         pgmode = VppEnum.vl_api_pg_interface_mode_t
 944         return cls.create_pg_interfaces_internal(interfaces, gso, gso_size,
 945                                                  pgmode.PG_API_MODE_ETHERNET)
 946
 947     @classmethod
 948     def create_loopback_interfaces(cls, count):
 949         """
 950         Create loopback interfaces.
 951
 952         :param count: number of interfaces created.
 953         :returns: List of created interfaces.
 954         """
 955         result = [VppLoInterface(cls) for i in range(count)]
 956         for intf in result:
 957             setattr(cls, intf.name, intf)
 958         cls.lo_interfaces = result
 959         return result
 960
 961     @classmethod
 962     def create_bvi_interfaces(cls, count):
 963         """
 964         Create BVI interfaces.
 965
 966         :param count: number of interfaces created.
 967         :returns: List of created interfaces.
 968         """
 969         result = [VppBviInterface(cls) for i in range(count)]
 970         for intf in result:
 971             setattr(cls, intf.name, intf)
 972         cls.bvi_interfaces = result
 973         return result
 974
 975     @staticmethod
 976     def extend_packet(packet, size, padding=' '):
 977         """
 978         Extend packet to given size by padding with spaces or custom padding
 979         NOTE: Currently works only when Raw layer is present.
 980
 981         :param packet: packet
 982         :param size: target size
 983         :param padding: padding used to extend the payload
 984
 985         """
 986         packet_len = len(packet) + 4
 987         extend = size - packet_len
 988         if extend > 0:
 989             num = (extend // len(padding)) + 1
 990             packet[Raw].load += (padding * num)[:extend].encode("ascii")
 991
 992     @classmethod
 993     def reset_packet_infos(cls):
 994         """ Reset the list of packet info objects and packet counts to zero """
 995         cls._packet_infos = {}
 996         cls._packet_count_for_dst_if_idx = {}
 997
 998     @classmethod
 999     def create_packet_info(cls, src_if, dst_if):
1000         """
1001         Create packet info object containing the source and destination indexes
1002         and add it to the testcase's packet info list
1003
1004         :param VppInterface src_if: source interface
1005         :param VppInterface dst_if: destination interface
1006
1007         :returns: _PacketInfo object
1008
1009         """
1010         info = _PacketInfo()
1011         info.index = len(cls._packet_infos)
1012         info.src = src_if.sw_if_index
1013         info.dst = dst_if.sw_if_index
1014         if isinstance(dst_if, VppSubInterface):
1015             dst_idx = dst_if.parent.sw_if_index
1016         else:
1017             dst_idx = dst_if.sw_if_index
1018         if dst_idx in cls._packet_count_for_dst_if_idx:
1019             cls._packet_count_for_dst_if_idx[dst_idx] += 1
1020         else:
1021             cls._packet_count_for_dst_if_idx[dst_idx] = 1
1022         cls._packet_infos[info.index] = info
1023         return info
1024
1025     @staticmethod
1026     def info_to_payload(info):
1027         """
1028         Convert _PacketInfo object to packet payload
1029
1030         :param info: _PacketInfo object
1031
1032         :returns: string containing serialized data from packet info
1033         """
1034
1035         # retrieve payload, currently 18 bytes (4 x ints + 1 short)
1036         return pack('iiiih', info.index, info.src,
1037                     info.dst, info.ip, info.proto)
1038
1039     @staticmethod
1040     def payload_to_info(payload, payload_field='load'):
1041         """
1042         Convert packet payload to _PacketInfo object
1043
1044         :param payload: packet payload
1045         :type payload:  <class 'scapy.packet.Raw'>
1046         :param payload_field: packet fieldname of payload "load" for
1047                 <class 'scapy.packet.Raw'>
1048         :type payload_field: str
1049         :returns: _PacketInfo object containing de-serialized data from payload
1050
1051         """
1052
1053         # retrieve payload, currently 18 bytes (4 x ints + 1 short)
1054         payload_b = getattr(payload, payload_field)[:18]
1055
1056         info = _PacketInfo()
1057         info.index, info.src, info.dst, info.ip, info.proto \
1058             = unpack('iiiih', payload_b)
1059
1060         # some SRv6 TCs depend on get an exception if bad values are detected
1061         if info.index > 0x4000:
1062             raise ValueError('Index value is invalid')
1063
1064         return info
1065
1066     def get_next_packet_info(self, info):
1067         """
1068         Iterate over the packet info list stored in the testcase
1069         Start iteration with first element if info is None
1070         Continue based on index in info if info is specified
1071
1072         :param info: info or None
1073         :returns: next info in list or None if no more infos
1074         """
1075         if info is None:
1076             next_index = 0
1077         else:
1078             next_index = info.index + 1
1079         if next_index == len(self._packet_infos):
1080             return None
1081         else:
1082             return self._packet_infos[next_index]
1083
1084     def get_next_packet_info_for_interface(self, src_index, info):
1085         """
1086         Search the packet info list for the next packet info with same source
1087         interface index
1088
1089         :param src_index: source interface index to search for
1090         :param info: packet info - where to start the search
1091         :returns: packet info or None
1092
1093         """
1094         while True:
1095             info = self.get_next_packet_info(info)
1096             if info is None:
1097                 return None
1098             if info.src == src_index:
1099                 return info
1100
1101     def get_next_packet_info_for_interface2(self, src_index, dst_index, info):
1102         """
1103         Search the packet info list for the next packet info with same source
1104         and destination interface indexes
1105
1106         :param src_index: source interface index to search for
1107         :param dst_index: destination interface index to search for
1108         :param info: packet info - where to start the search
1109         :returns: packet info or None
1110
1111         """
1112         while True:
1113             info = self.get_next_packet_info_for_interface(src_index, info)
1114             if info is None:
1115                 return None
1116             if info.dst == dst_index:
1117                 return info
1118
1119     def assert_equal(self, real_value, expected_value, name_or_class=None):
1120         if name_or_class is None:
1121             self.assertEqual(real_value, expected_value)
1122             return
1123         try:
1124             msg = "Invalid %s: %d('%s') does not match expected value %d('%s')"
1125             msg = msg % (getdoc(name_or_class).strip(),
1126                          real_value, str(name_or_class(real_value)),
1127                          expected_value, str(name_or_class(expected_value)))
1128         except Exception:
1129             msg = "Invalid %s: %s does not match expected value %s" % (
1130                 name_or_class, real_value, expected_value)
1131
1132         self.assertEqual(real_value, expected_value, msg)
1133
1134     def assert_in_range(self,
1135                         real_value,
1136                         expected_min,
1137                         expected_max,
1138                         name=None):
1139         if name is None:
1140             msg = None
1141         else:
1142             msg = "Invalid %s: %s out of range <%s,%s>" % (
1143                 name, real_value, expected_min, expected_max)
1144         self.assertTrue(expected_min <= real_value <= expected_max, msg)
1145
1146     def assert_packet_checksums_valid(self, packet,
1147                                       ignore_zero_udp_checksums=True):
1148         received = packet.__class__(scapy.compat.raw(packet))
1149         udp_layers = ['UDP', 'UDPerror']
1150         checksum_fields = ['cksum', 'chksum']
1151         checksums = []
1152         counter = 0
1153         temp = received.__class__(scapy.compat.raw(received))
1154         while True:
1155             layer = temp.getlayer(counter)
1156             if layer:
1157                 layer = layer.copy()
1158                 layer.remove_payload()
1159                 for cf in checksum_fields:
1160                     if hasattr(layer, cf):
1161                         if ignore_zero_udp_checksums and \
1162                                 0 == getattr(layer, cf) and \
1163                                 layer.name in udp_layers:
1164                             continue
1165                         delattr(temp.getlayer(counter), cf)
1166                         checksums.append((counter, cf))
1167             else:
1168                 break
1169             counter = counter + 1
1170         if 0 == len(checksums):
1171             return
1172         temp = temp.__class__(scapy.compat.raw(temp))
1173         for layer, cf in checksums:
1174             calc_sum = getattr(temp[layer], cf)
1175             self.assert_equal(
1176                 getattr(received[layer], cf), calc_sum,
1177                 "packet checksum on layer #%d: %s" % (layer, temp[layer].name))
1178             self.logger.debug(
1179                 "Checksum field `%s` on `%s` layer has correct value `%s`" %
1180                 (cf, temp[layer].name, calc_sum))
1181
1182     def assert_checksum_valid(self, received_packet, layer,
1183                               field_name='chksum',
1184                               ignore_zero_checksum=False):
1185         """ Check checksum of received packet on given layer """
1186         received_packet_checksum = getattr(received_packet[layer], field_name)
1187         if ignore_zero_checksum and 0 == received_packet_checksum:
1188             return
1189         recalculated = received_packet.__class__(
1190             scapy.compat.raw(received_packet))
1191         delattr(recalculated[layer], field_name)
1192         recalculated = recalculated.__class__(scapy.compat.raw(recalculated))
1193         self.assert_equal(received_packet_checksum,
1194                           getattr(recalculated[layer], field_name),
1195                           "packet checksum on layer: %s" % layer)
1196
1197     def assert_ip_checksum_valid(self, received_packet,
1198                                  ignore_zero_checksum=False):
1199         self.assert_checksum_valid(received_packet, 'IP',
1200                                    ignore_zero_checksum=ignore_zero_checksum)
1201
1202     def assert_tcp_checksum_valid(self, received_packet,
1203                                   ignore_zero_checksum=False):
1204         self.assert_checksum_valid(received_packet, 'TCP',
1205                                    ignore_zero_checksum=ignore_zero_checksum)
1206
1207     def assert_udp_checksum_valid(self, received_packet,
1208                                   ignore_zero_checksum=True):
1209         self.assert_checksum_valid(received_packet, 'UDP',
1210                                    ignore_zero_checksum=ignore_zero_checksum)
1211
1212     def assert_embedded_icmp_checksum_valid(self, received_packet):
1213         if received_packet.haslayer(IPerror):
1214             self.assert_checksum_valid(received_packet, 'IPerror')
1215         if received_packet.haslayer(TCPerror):
1216             self.assert_checksum_valid(received_packet, 'TCPerror')
1217         if received_packet.haslayer(UDPerror):
1218             self.assert_checksum_valid(received_packet, 'UDPerror',
1219                                        ignore_zero_checksum=True)
1220         if received_packet.haslayer(ICMPerror):
1221             self.assert_checksum_valid(received_packet, 'ICMPerror')
1222
1223     def assert_icmp_checksum_valid(self, received_packet):
1224         self.assert_checksum_valid(received_packet, 'ICMP')
1225         self.assert_embedded_icmp_checksum_valid(received_packet)
1226
1227     def assert_icmpv6_checksum_valid(self, pkt):
1228         if pkt.haslayer(ICMPv6DestUnreach):
1229             self.assert_checksum_valid(pkt, 'ICMPv6DestUnreach', 'cksum')
1230             self.assert_embedded_icmp_checksum_valid(pkt)
1231         if pkt.haslayer(ICMPv6EchoRequest):
1232             self.assert_checksum_valid(pkt, 'ICMPv6EchoRequest', 'cksum')
1233         if pkt.haslayer(ICMPv6EchoReply):
1234             self.assert_checksum_valid(pkt, 'ICMPv6EchoReply', 'cksum')
1235
1236     def get_packet_counter(self, counter):
1237         if counter.startswith("/"):
1238             counter_value = self.statistics.get_counter(counter)
1239         else:
1240             counters = self.vapi.cli("sh errors").split('\n')
1241             counter_value = 0
1242             for i in range(1, len(counters) - 1):
1243                 results = counters[i].split()
1244                 if results[1] == counter:
1245                     counter_value = int(results[0])
1246                     break
1247         return counter_value
1248
1249     def assert_packet_counter_equal(self, counter, expected_value):
1250         counter_value = self.get_packet_counter(counter)
1251         self.assert_equal(counter_value, expected_value,
1252                           "packet counter `%s'" % counter)
1253
1254     def assert_error_counter_equal(self, counter, expected_value):
1255         counter_value = self.statistics[counter].sum()
1256         self.assert_equal(counter_value, expected_value,
1257                           "error counter `%s'" % counter)
1258
1259     @classmethod
1260     def sleep(cls, timeout, remark=None):
1261
1262         # /* Allow sleep(0) to maintain win32 semantics, and as decreed
1263         #  * by Guido, only the main thread can be interrupted.
1264         # */
1265         # https://github.com/python/cpython/blob/6673decfa0fb078f60587f5cb5e98460eea137c2/Modules/timemodule.c#L1892  # noqa
1266         if timeout == 0:
1267             # yield quantum
1268             if hasattr(os, 'sched_yield'):
1269                 os.sched_yield()
1270             else:
1271                 time.sleep(0)
1272             return
1273
1274         cls.logger.debug("Starting sleep for %es (%s)", timeout, remark)
1275         before = time.time()
1276         time.sleep(timeout)
1277         after = time.time()
1278         if after - before > 2 * timeout:
1279             cls.logger.error("unexpected self.sleep() result - "
1280                              "slept for %es instead of ~%es!",
1281                              after - before, timeout)
1282
1283         cls.logger.debug(
1284             "Finished sleep (%s) - slept %es (wanted %es)",
1285             remark, after - before, timeout)
1286
1287     def pg_send(self, intf, pkts, worker=None, trace=True):
1288         intf.add_stream(pkts, worker=worker)
1289         self.pg_enable_capture(self.pg_interfaces)
1290         self.pg_start(trace=trace)
1291
1292     def send_and_assert_no_replies(self, intf, pkts, remark="", timeout=None):
1293         self.pg_send(intf, pkts)
1294         if not timeout:
1295             timeout = 1
1296         for i in self.pg_interfaces:
1297             i.get_capture(0, timeout=timeout)
1298             i.assert_nothing_captured(remark=remark)
1299             timeout = 0.1
1300
1301     def send_and_expect(self, intf, pkts, output, n_rx=None, worker=None,
1302                         trace=True):
1303         if not n_rx:
1304             n_rx = len(pkts)
1305         self.pg_send(intf, pkts, worker=worker, trace=trace)
1306         rx = output.get_capture(n_rx)
1307         if trace:
1308             self.logger.debug(self.vapi.cli("show trace"))
1309         return rx
1310
1311     def send_and_expect_only(self, intf, pkts, output, timeout=None):
1312         self.pg_send(intf, pkts)
1313         rx = output.get_capture(len(pkts))
1314         outputs = [output]
1315         if not timeout:
1316             timeout = 1
1317         for i in self.pg_interfaces:
1318             if i not in outputs:
1319                 i.get_capture(0, timeout=timeout)
1320                 i.assert_nothing_captured()
1321                 timeout = 0.1
1322
1323         return rx
1324
1325
1326 def get_testcase_doc_name(test):
1327     return getdoc(test.__class__).splitlines()[0]
1328
1329
1330 def get_test_description(descriptions, test):
1331     short_description = test.shortDescription()
1332     if descriptions and short_description:
1333         return short_description
1334     else:
1335         return str(test)
1336
1337
1338 class TestCaseInfo(object):
1339     def __init__(self, logger, tempdir, vpp_pid, vpp_bin_path):
1340         self.logger = logger
1341         self.tempdir = tempdir
1342         self.vpp_pid = vpp_pid
1343         self.vpp_bin_path = vpp_bin_path
1344         self.core_crash_test = None
1345
1346
1347 class VppTestResult(unittest.TestResult):
1348     """
1349     @property result_string
1350      String variable to store the test case result string.
1351     @property errors
1352      List variable containing 2-tuples of TestCase instances and strings
1353      holding formatted tracebacks. Each tuple represents a test which
1354      raised an unexpected exception.
1355     @property failures
1356      List variable containing 2-tuples of TestCase instances and strings
1357      holding formatted tracebacks. Each tuple represents a test where
1358      a failure was explicitly signalled using the TestCase.assert*()
1359      methods.
1360     """
1361
1362     failed_test_cases_info = set()
1363     core_crash_test_cases_info = set()
1364     current_test_case_info = None
1365
1366     def __init__(self, stream=None, descriptions=None, verbosity=None,
1367                  runner=None):
1368         """
1369         :param stream File descriptor to store where to report test results.
1370             Set to the standard error stream by default.
1371         :param descriptions Boolean variable to store information if to use
1372             test case descriptions.
1373         :param verbosity Integer variable to store required verbosity level.
1374         """
1375         super(VppTestResult, self).__init__(stream, descriptions, verbosity)
1376         self.stream = stream
1377         self.descriptions = descriptions
1378         self.verbosity = verbosity
1379         self.result_string = None
1380         self.runner = runner
1381         self.printed = []
1382
1383     def addSuccess(self, test):
1384         """
1385         Record a test succeeded result
1386
1387         :param test:
1388
1389         """
1390         if self.current_test_case_info:
1391             self.current_test_case_info.logger.debug(
1392                 "--- addSuccess() %s.%s(%s) called" % (test.__class__.__name__,
1393                                                        test._testMethodName,
1394                                                        test._testMethodDoc))
1395         unittest.TestResult.addSuccess(self, test)
1396         self.result_string = colorize("OK", GREEN)
1397
1398         self.send_result_through_pipe(test, PASS)
1399
1400     def addSkip(self, test, reason):
1401         """
1402         Record a test skipped.
1403
1404         :param test:
1405         :param reason:
1406
1407         """
1408         if self.current_test_case_info:
1409             self.current_test_case_info.logger.debug(
1410                 "--- addSkip() %s.%s(%s) called, reason is %s" %
1411                 (test.__class__.__name__, test._testMethodName,
1412                  test._testMethodDoc, reason))
1413         unittest.TestResult.addSkip(self, test, reason)
1414         self.result_string = colorize("SKIP", YELLOW)
1415
1416         if reason == "not enough cpus":
1417             self.send_result_through_pipe(test, SKIP_CPU_SHORTAGE)
1418         else:
1419             self.send_result_through_pipe(test, SKIP)
1420
1421     def symlink_failed(self):
1422         if self.current_test_case_info:
1423             try:
1424                 failed_dir = os.getenv('FAILED_DIR')
1425                 link_path = os.path.join(
1426                     failed_dir,
1427                     '%s-FAILED' %
1428                     os.path.basename(self.current_test_case_info.tempdir))
1429
1430                 self.current_test_case_info.logger.debug(
1431                     "creating a link to the failed test")
1432                 self.current_test_case_info.logger.debug(
1433                     "os.symlink(%s, %s)" %
1434                     (self.current_test_case_info.tempdir, link_path))
1435                 if os.path.exists(link_path):
1436                     self.current_test_case_info.logger.debug(
1437                         'symlink already exists')
1438                 else:
1439                     os.symlink(self.current_test_case_info.tempdir, link_path)
1440
1441             except Exception as e:
1442                 self.current_test_case_info.logger.error(e)
1443
1444     def send_result_through_pipe(self, test, result):
1445         if hasattr(self, 'test_framework_result_pipe'):
1446             pipe = self.test_framework_result_pipe
1447             if pipe:
1448                 pipe.send((test.id(), result))
1449
1450     def log_error(self, test, err, fn_name):
1451         if self.current_test_case_info:
1452             if isinstance(test, unittest.suite._ErrorHolder):
1453                 test_name = test.description
1454             else:
1455                 test_name = '%s.%s(%s)' % (test.__class__.__name__,
1456                                            test._testMethodName,
1457                                            test._testMethodDoc)
1458             self.current_test_case_info.logger.debug(
1459                 "--- %s() %s called, err is %s" %
1460                 (fn_name, test_name, err))
1461             self.current_test_case_info.logger.debug(
1462                 "formatted exception is:\n%s" %
1463                 "".join(format_exception(*err)))
1464
1465     def add_error(self, test, err, unittest_fn, error_type):
1466         if error_type == FAIL:
1467             self.log_error(test, err, 'addFailure')
1468             error_type_str = colorize("FAIL", RED)
1469         elif error_type == ERROR:
1470             self.log_error(test, err, 'addError')
1471             error_type_str = colorize("ERROR", RED)
1472         else:
1473             raise Exception('Error type %s cannot be used to record an '
1474                             'error or a failure' % error_type)
1475
1476         unittest_fn(self, test, err)
1477         if self.current_test_case_info:
1478             self.result_string = "%s [ temp dir used by test case: %s ]" % \
1479                                  (error_type_str,
1480                                   self.current_test_case_info.tempdir)
1481             self.symlink_failed()
1482             self.failed_test_cases_info.add(self.current_test_case_info)
1483             if is_core_present(self.current_test_case_info.tempdir):
1484                 if not self.current_test_case_info.core_crash_test:
1485                     if isinstance(test, unittest.suite._ErrorHolder):
1486                         test_name = str(test)
1487                     else:
1488                         test_name = "'{!s}' ({!s})".format(
1489                             get_testcase_doc_name(test), test.id())
1490                     self.current_test_case_info.core_crash_test = test_name
1491                 self.core_crash_test_cases_info.add(
1492                     self.current_test_case_info)
1493         else:
1494             self.result_string = '%s [no temp dir]' % error_type_str
1495
1496         self.send_result_through_pipe(test, error_type)
1497
1498     def addFailure(self, test, err):
1499         """
1500         Record a test failed result
1501
1502         :param test:
1503         :param err: error message
1504
1505         """
1506         self.add_error(test, err, unittest.TestResult.addFailure, FAIL)
1507
1508     def addError(self, test, err):
1509         """
1510         Record a test error result
1511
1512         :param test:
1513         :param err: error message
1514
1515         """
1516         self.add_error(test, err, unittest.TestResult.addError, ERROR)
1517
1518     def getDescription(self, test):
1519         """
1520         Get test description
1521
1522         :param test:
1523         :returns: test description
1524
1525         """
1526         return get_test_description(self.descriptions, test)
1527
1528     def startTest(self, test):
1529         """
1530         Start a test
1531
1532         :param test:
1533
1534         """
1535
1536         def print_header(test):
1537             if test.__class__ in self.printed:
1538                 return
1539
1540             test_doc = getdoc(test)
1541             if not test_doc:
1542                 raise Exception("No doc string for test '%s'" % test.id())
1543
1544             test_title = test_doc.splitlines()[0].rstrip()
1545             test_title = colorize(test_title, GREEN)
1546             if test.is_tagged_run_solo():
1547                 test_title = colorize(f"SOLO RUN: {test_title}", YELLOW)
1548
1549             # This block may overwrite the colorized title above,
1550             # but we want this to stand out and be fixed
1551             if test.has_tag(TestCaseTag.FIXME_VPP_WORKERS):
1552                 test_title = colorize(
1553                     f"FIXME with VPP workers: {test_title}", RED)
1554
1555             if hasattr(test, 'vpp_worker_count'):
1556                 if test.vpp_worker_count == 0:
1557                     test_title += " [main thread only]"
1558                 elif test.vpp_worker_count == 1:
1559                     test_title += " [1 worker thread]"
1560                 else:
1561                     test_title += f" [{test.vpp_worker_count} worker threads]"
1562
1563             if test.__class__.skipped_due_to_cpu_lack:
1564                 test_title = colorize(
1565                     f"{test_title} [skipped - not enough cpus, "
1566                     f"required={test.__class__.get_cpus_required()}, "
1567                     f"available={max_vpp_cpus}]", YELLOW)
1568
1569             print(double_line_delim)
1570             print(test_title)
1571             print(double_line_delim)
1572             self.printed.append(test.__class__)
1573
1574         print_header(test)
1575         self.start_test = time.time()
1576         unittest.TestResult.startTest(self, test)
1577         if self.verbosity > 0:
1578             self.stream.writeln(
1579                 "Starting " + self.getDescription(test) + " ...")
1580             self.stream.writeln(single_line_delim)
1581
1582     def stopTest(self, test):
1583         """
1584         Called when the given test has been run
1585
1586         :param test:
1587
1588         """
1589         unittest.TestResult.stopTest(self, test)
1590
1591         if self.verbosity > 0:
1592             self.stream.writeln(single_line_delim)
1593             self.stream.writeln("%-73s%s" % (self.getDescription(test),
1594                                              self.result_string))
1595             self.stream.writeln(single_line_delim)
1596         else:
1597             self.stream.writeln("%-68s %4.2f %s" %
1598                                 (self.getDescription(test),
1599                                  time.time() - self.start_test,
1600                                  self.result_string))
1601
1602         self.send_result_through_pipe(test, TEST_RUN)
1603
1604     def printErrors(self):
1605         """
1606         Print errors from running the test case
1607         """
1608         if len(self.errors) > 0 or len(self.failures) > 0:
1609             self.stream.writeln()
1610             self.printErrorList('ERROR', self.errors)
1611             self.printErrorList('FAIL', self.failures)
1612
1613         # ^^ that is the last output from unittest before summary
1614         if not self.runner.print_summary:
1615             devnull = unittest.runner._WritelnDecorator(open(os.devnull, 'w'))
1616             self.stream = devnull
1617             self.runner.stream = devnull
1618
1619     def printErrorList(self, flavour, errors):
1620         """
1621         Print error list to the output stream together with error type
1622         and test case description.
1623
1624         :param flavour: error type
1625         :param errors: iterable errors
1626
1627         """
1628         for test, err in errors:
1629             self.stream.writeln(double_line_delim)
1630             self.stream.writeln("%s: %s" %
1631                                 (flavour, self.getDescription(test)))
1632             self.stream.writeln(single_line_delim)
1633             self.stream.writeln("%s" % err)
1634
1635
1636 class VppTestRunner(unittest.TextTestRunner):
1637     """
1638     A basic test runner implementation which prints results to standard error.
1639     """
1640
1641     @property
1642     def resultclass(self):
1643         """Class maintaining the results of the tests"""
1644         return VppTestResult
1645
1646     def __init__(self, keep_alive_pipe=None, descriptions=True, verbosity=1,
1647                  result_pipe=None, failfast=False, buffer=False,
1648                  resultclass=None, print_summary=True, **kwargs):
1649         # ignore stream setting here, use hard-coded stdout to be in sync
1650         # with prints from VppTestCase methods ...
1651         super(VppTestRunner, self).__init__(sys.stdout, descriptions,
1652                                             verbosity, failfast, buffer,
1653                                             resultclass, **kwargs)
1654         KeepAliveReporter.pipe = keep_alive_pipe
1655
1656         self.orig_stream = self.stream
1657         self.resultclass.test_framework_result_pipe = result_pipe
1658
1659         self.print_summary = print_summary
1660
1661     def _makeResult(self):
1662         return self.resultclass(self.stream,
1663                                 self.descriptions,
1664                                 self.verbosity,
1665                                 self)
1666
1667     def run(self, test):
1668         """
1669         Run the tests
1670
1671         :param test:
1672
1673         """
1674         faulthandler.enable()  # emit stack trace to stderr if killed by signal
1675
1676         result = super(VppTestRunner, self).run(test)
1677         if not self.print_summary:
1678             self.stream = self.orig_stream
1679             result.stream = self.orig_stream
1680         return result
1681
1682
1683 class Worker(Thread):
1684     def __init__(self, executable_args, logger, env=None, *args, **kwargs):
1685         super(Worker, self).__init__(*args, **kwargs)
1686         self.logger = logger
1687         self.args = executable_args
1688         if hasattr(self, 'testcase') and self.testcase.debug_all:
1689             if self.testcase.debug_gdbserver:
1690                 self.args = ['/usr/bin/gdbserver', 'localhost:{port}'
1691                              .format(port=self.testcase.gdbserver_port)] + args
1692             elif self.testcase.debug_gdb and hasattr(self, 'wait_for_gdb'):
1693                 self.args.append(self.wait_for_gdb)
1694         self.app_bin = executable_args[0]
1695         self.app_name = os.path.basename(self.app_bin)
1696         if hasattr(self, 'role'):
1697             self.app_name += ' {role}'.format(role=self.role)
1698         self.process = None
1699         self.result = None
1700         env = {} if env is None else env
1701         self.env = copy.deepcopy(env)
1702
1703     def wait_for_enter(self):
1704         if not hasattr(self, 'testcase'):
1705             return
1706         if self.testcase.debug_all and self.testcase.debug_gdbserver:
1707             print()
1708             print(double_line_delim)
1709             print("Spawned GDB Server for '{app}' with PID: {pid}"
1710                   .format(app=self.app_name, pid=self.process.pid))
1711         elif self.testcase.debug_all and self.testcase.debug_gdb:
1712             print()
1713             print(double_line_delim)
1714             print("Spawned '{app}' with PID: {pid}"
1715                   .format(app=self.app_name, pid=self.process.pid))
1716         else:
1717             return
1718         print(single_line_delim)
1719         print("You can debug '{app}' using:".format(app=self.app_name))
1720         if self.testcase.debug_gdbserver:
1721             print("sudo gdb " + self.app_bin +
1722                   " -ex 'target remote localhost:{port}'"
1723                   .format(port=self.testcase.gdbserver_port))
1724             print("Now is the time to attach gdb by running the above "
1725                   "command, set up breakpoints etc., then resume from "
1726                   "within gdb by issuing the 'continue' command")
1727             self.testcase.gdbserver_port += 1
1728         elif self.testcase.debug_gdb:
1729             print("sudo gdb " + self.app_bin +
1730                   " -ex 'attach {pid}'".format(pid=self.process.pid))
1731             print("Now is the time to attach gdb by running the above "
1732                   "command and set up breakpoints etc., then resume from"
1733                   " within gdb by issuing the 'continue' command")
1734         print(single_line_delim)
1735         input("Press ENTER to continue running the testcase...")
1736
1737     def run(self):
1738         executable = self.args[0]
1739         if not os.path.exists(executable) or not os.access(
1740                 executable, os.F_OK | os.X_OK):
1741             # Exit code that means some system file did not exist,
1742             # could not be opened, or had some other kind of error.
1743             self.result = os.EX_OSFILE
1744             raise EnvironmentError(
1745                 "executable '%s' is not found or executable." % executable)
1746         self.logger.debug("Running executable '{app}': '{cmd}'"
1747                           .format(app=self.app_name,
1748                                   cmd=' '.join(self.args)))
1749         env = os.environ.copy()
1750         env.update(self.env)
1751         env["CK_LOG_FILE_NAME"] = "-"
1752         self.process = subprocess.Popen(
1753             ['stdbuf', '-o0', '-e0'] + self.args, shell=False, env=env,
1754             preexec_fn=os.setpgrp, stdout=subprocess.PIPE,
1755             stderr=subprocess.PIPE)
1756         self.wait_for_enter()
1757         out, err = self.process.communicate()
1758         self.logger.debug("Finished running `{app}'".format(app=self.app_name))
1759         self.logger.info("Return code is `%s'" % self.process.returncode)
1760         self.logger.info(single_line_delim)
1761         self.logger.info("Executable `{app}' wrote to stdout:"
1762                          .format(app=self.app_name))
1763         self.logger.info(single_line_delim)
1764         self.logger.info(out.decode('utf-8'))
1765         self.logger.info(single_line_delim)
1766         self.logger.info("Executable `{app}' wrote to stderr:"
1767                          .format(app=self.app_name))
1768         self.logger.info(single_line_delim)
1769         self.logger.info(err.decode('utf-8'))
1770         self.logger.info(single_line_delim)
1771         self.result = self.process.returncode
1772
1773
1774 if __name__ == '__main__':
1775     pass