test/asf/asfframework.py

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