test/framework.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import print_function
   4 import gc
   5 import sys
   6 import os
   7 import select
   8 import unittest
   9 import tempfile
  10 import time
  11 import faulthandler
  12 import random
  13 import copy
  14 import psutil
  15 import platform
  16 from collections import deque
  17 from threading import Thread, Event
  18 from inspect import getdoc, isclass
  19 from traceback import format_exception
  20 from logging import FileHandler, DEBUG, Formatter
  21 from scapy.packet import Raw
  22 from hook import StepHook, PollHook, VppDiedError
  23 from vpp_pg_interface import VppPGInterface
  24 from vpp_sub_interface import VppSubInterface
  25 from vpp_lo_interface import VppLoInterface
  26 from vpp_papi_provider import VppPapiProvider
  27 from vpp_papi.vpp_stats import VPPStats
  28 from log import RED, GREEN, YELLOW, double_line_delim, single_line_delim, \
  29     get_logger, colorize
  30 from vpp_object import VppObjectRegistry
  31 from util import ppp, is_core_present
  32 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
  33 from scapy.layers.inet6 import ICMPv6DestUnreach, ICMPv6EchoRequest
  34 from scapy.layers.inet6 import ICMPv6EchoReply
  35
  36 if os.name == 'posix' and sys.version_info[0] < 3:
  37     # using subprocess32 is recommended by python official documentation
  38     # @ https://docs.python.org/2/library/subprocess.html
  39     import subprocess32 as subprocess
  40 else:
  41     import subprocess
  42
  43 PASS = 0
  44 FAIL = 1
  45 ERROR = 2
  46 SKIP = 3
  47 TEST_RUN = 4
  48
  49 debug_framework = False
  50 if os.getenv('TEST_DEBUG', "0") == "1":
  51     debug_framework = True
  52     import debug_internal
  53
  54 """
  55   Test framework module.
  56
  57   The module provides a set of tools for constructing and running tests and
  58   representing the results.
  59 """
  60
  61
  62 class _PacketInfo(object):
  63     """Private class to create packet info object.
  64
  65     Help process information about the next packet.
  66     Set variables to default values.
  67     """
  68     #: Store the index of the packet.
  69     index = -1
  70     #: Store the index of the source packet generator interface of the packet.
  71     src = -1
  72     #: Store the index of the destination packet generator interface
  73     #: of the packet.
  74     dst = -1
  75     #: Store expected ip version
  76     ip = -1
  77     #: Store expected upper protocol
  78     proto = -1
  79     #: Store the copy of the former packet.
  80     data = None
  81
  82     def __eq__(self, other):
  83         index = self.index == other.index
  84         src = self.src == other.src
  85         dst = self.dst == other.dst
  86         data = self.data == other.data
  87         return index and src and dst and data
  88
  89
  90 def pump_output(testclass):
  91     """ pump output from vpp stdout/stderr to proper queues """
  92     stdout_fragment = ""
  93     stderr_fragment = ""
  94     while not testclass.pump_thread_stop_flag.is_set():
  95         readable = select.select([testclass.vpp.stdout.fileno(),
  96                                   testclass.vpp.stderr.fileno(),
  97                                   testclass.pump_thread_wakeup_pipe[0]],
  98                                  [], [])[0]
  99         if testclass.vpp.stdout.fileno() in readable:
 100             read = os.read(testclass.vpp.stdout.fileno(), 102400)
 101             if len(read) > 0:
 102                 split = read.splitlines(True)
 103                 if len(stdout_fragment) > 0:
 104                     split[0] = "%s%s" % (stdout_fragment, split[0])
 105                 if len(split) > 0 and split[-1].endswith("\n"):
 106                     limit = None
 107                 else:
 108                     limit = -1
 109                     stdout_fragment = split[-1]
 110                 testclass.vpp_stdout_deque.extend(split[:limit])
 111                 if not testclass.cache_vpp_output:
 112                     for line in split[:limit]:
 113                         testclass.logger.debug(
 114                             "VPP STDOUT: %s" % line.rstrip("\n"))
 115         if testclass.vpp.stderr.fileno() in readable:
 116             read = os.read(testclass.vpp.stderr.fileno(), 102400)
 117             if len(read) > 0:
 118                 split = read.splitlines(True)
 119                 if len(stderr_fragment) > 0:
 120                     split[0] = "%s%s" % (stderr_fragment, split[0])
 121                 if len(split) > 0 and split[-1].endswith(b"\n"):
 122                     limit = None
 123                 else:
 124                     limit = -1
 125                     stderr_fragment = split[-1]
 126                 testclass.vpp_stderr_deque.extend(split[:limit])
 127                 if not testclass.cache_vpp_output:
 128                     for line in split[:limit]:
 129                         testclass.logger.debug(
 130                             "VPP STDERR: %s" % line.rstrip("\n"))
 131                         # ignoring the dummy pipe here intentionally - the
 132                         # flag will take care of properly terminating the loop
 133
 134
 135 def is_skip_aarch64_set():
 136     return os.getenv('SKIP_AARCH64', 'n').lower() in ('yes', 'y', '1')
 137
 138
 139 def is_platform_aarch64():
 140     return platform.machine() == 'aarch64'
 141
 142
 143 def running_extended_tests():
 144     s = os.getenv("EXTENDED_TESTS", "n")
 145     return True if s.lower() in ("y", "yes", "1") else False
 146
 147
 148 def running_on_centos():
 149     os_id = os.getenv("OS_ID", "")
 150     return True if "centos" in os_id.lower() else False
 151
 152
 153 class KeepAliveReporter(object):
 154     """
 155     Singleton object which reports test start to parent process
 156     """
 157     _shared_state = {}
 158
 159     def __init__(self):
 160         self.__dict__ = self._shared_state
 161         self._pipe = None
 162
 163     @property
 164     def pipe(self):
 165         return self._pipe
 166
 167     @pipe.setter
 168     def pipe(self, pipe):
 169         if self._pipe is not None:
 170             raise Exception("Internal error - pipe should only be set once.")
 171         self._pipe = pipe
 172
 173     def send_keep_alive(self, test, desc=None):
 174         """
 175         Write current test tmpdir & desc to keep-alive pipe to signal liveness
 176         """
 177         if self.pipe is None:
 178             # if not running forked..
 179             return
 180
 181         if isclass(test):
 182             desc = '%s (%s)' % (desc, unittest.util.strclass(test))
 183         else:
 184             desc = test.id()
 185
 186         self.pipe.send((desc, test.vpp_bin, test.tempdir, test.vpp.pid))
 187
 188
 189 class VppTestCase(unittest.TestCase):
 190     """This subclass is a base class for VPP test cases that are implemented as
 191     classes. It provides methods to create and run test case.
 192     """
 193
 194     extra_vpp_punt_config = []
 195
 196     @property
 197     def packet_infos(self):
 198         """List of packet infos"""
 199         return self._packet_infos
 200
 201     @classmethod
 202     def get_packet_count_for_if_idx(cls, dst_if_index):
 203         """Get the number of packet info for specified destination if index"""
 204         if dst_if_index in cls._packet_count_for_dst_if_idx:
 205             return cls._packet_count_for_dst_if_idx[dst_if_index]
 206         else:
 207             return 0
 208
 209     @classmethod
 210     def instance(cls):
 211         """Return the instance of this testcase"""
 212         return cls.test_instance
 213
 214     @classmethod
 215     def set_debug_flags(cls, d):
 216         cls.debug_core = False
 217         cls.debug_gdb = False
 218         cls.debug_gdbserver = False
 219         if d is None:
 220             return
 221         dl = d.lower()
 222         if dl == "core":
 223             cls.debug_core = True
 224         elif dl == "gdb":
 225             cls.debug_gdb = True
 226         elif dl == "gdbserver":
 227             cls.debug_gdbserver = True
 228         else:
 229             raise Exception("Unrecognized DEBUG option: '%s'" % d)
 230
 231     @staticmethod
 232     def get_least_used_cpu():
 233         cpu_usage_list = [set(range(psutil.cpu_count()))]
 234         vpp_processes = [p for p in psutil.process_iter(attrs=['pid', 'name'])
 235                          if 'vpp_main' == p.info['name']]
 236         for vpp_process in vpp_processes:
 237             for cpu_usage_set in cpu_usage_list:
 238                 try:
 239                     cpu_num = vpp_process.cpu_num()
 240                     if cpu_num in cpu_usage_set:
 241                         cpu_usage_set_index = cpu_usage_list.index(
 242                             cpu_usage_set)
 243                         if cpu_usage_set_index == len(cpu_usage_list) - 1:
 244                             cpu_usage_list.append({cpu_num})
 245                         else:
 246                             cpu_usage_list[cpu_usage_set_index + 1].add(
 247                                 cpu_num)
 248                         cpu_usage_set.remove(cpu_num)
 249                         break
 250                 except psutil.NoSuchProcess:
 251                     pass
 252
 253         for cpu_usage_set in cpu_usage_list:
 254             if len(cpu_usage_set) > 0:
 255                 min_usage_set = cpu_usage_set
 256                 break
 257
 258         return random.choice(tuple(min_usage_set))
 259
 260     @classmethod
 261     def print_header(cls):
 262         if not hasattr(cls, '_header_printed'):
 263             print(double_line_delim)
 264             print(colorize(getdoc(cls).splitlines()[0], GREEN))
 265             print(double_line_delim)
 266             cls._header_printed = True
 267
 268     @classmethod
 269     def setUpConstants(cls):
 270         """ Set-up the test case class based on environment variables """
 271         s = os.getenv("STEP", "n")
 272         cls.step = True if s.lower() in ("y", "yes", "1") else False
 273         d = os.getenv("DEBUG", None)
 274         c = os.getenv("CACHE_OUTPUT", "1")
 275         cls.cache_vpp_output = False if c.lower() in ("n", "no", "0") else True
 276         cls.set_debug_flags(d)
 277         cls.vpp_bin = os.getenv('VPP_BIN', "vpp")
 278         cls.plugin_path = os.getenv('VPP_PLUGIN_PATH')
 279         cls.extern_plugin_path = os.getenv('EXTERN_PLUGINS')
 280         plugin_path = None
 281         if cls.plugin_path is not None:
 282             if cls.extern_plugin_path is not None:
 283                 plugin_path = "%s:%s" % (
 284                     cls.plugin_path, cls.extern_plugin_path)
 285             else:
 286                 plugin_path = cls.plugin_path
 287         elif cls.extern_plugin_path is not None:
 288             plugin_path = cls.extern_plugin_path
 289         debug_cli = ""
 290         if cls.step or cls.debug_gdb or cls.debug_gdbserver:
 291             debug_cli = "cli-listen localhost:5002"
 292         coredump_size = None
 293         size = os.getenv("COREDUMP_SIZE")
 294         if size is not None:
 295             coredump_size = "coredump-size %s" % size
 296         if coredump_size is None:
 297             coredump_size = "coredump-size unlimited"
 298
 299         cpu_core_number = cls.get_least_used_cpu()
 300
 301         cls.vpp_cmdline = [cls.vpp_bin, "unix",
 302                            "{", "nodaemon", debug_cli, "full-coredump",
 303                            coredump_size, "runtime-dir", cls.tempdir, "}",
 304                            "api-trace", "{", "on", "}", "api-segment", "{",
 305                            "prefix", cls.shm_prefix, "}", "cpu", "{",
 306                            "main-core", str(cpu_core_number), "}", "statseg",
 307                            "{", "socket-name", cls.stats_sock, "}", "plugins",
 308                            "{", "plugin", "dpdk_plugin.so", "{", "disable",
 309                            "}", "plugin", "unittest_plugin.so", "{", "enable",
 310                            "}", "}", ]
 311         if cls.extra_vpp_punt_config is not None:
 312             cls.vpp_cmdline.extend(cls.extra_vpp_punt_config)
 313         if plugin_path is not None:
 314             cls.vpp_cmdline.extend(["plugin_path", plugin_path])
 315         cls.logger.info("vpp_cmdline args: %s" % cls.vpp_cmdline)
 316         cls.logger.info("vpp_cmdline: %s" % " ".join(cls.vpp_cmdline))
 317
 318     @classmethod
 319     def wait_for_enter(cls):
 320         if cls.debug_gdbserver:
 321             print(double_line_delim)
 322             print("Spawned GDB server with PID: %d" % cls.vpp.pid)
 323         elif cls.debug_gdb:
 324             print(double_line_delim)
 325             print("Spawned VPP with PID: %d" % cls.vpp.pid)
 326         else:
 327             cls.logger.debug("Spawned VPP with PID: %d" % cls.vpp.pid)
 328             return
 329         print(single_line_delim)
 330         print("You can debug the VPP using e.g.:")
 331         if cls.debug_gdbserver:
 332             print("gdb " + cls.vpp_bin + " -ex 'target remote localhost:7777'")
 333             print("Now is the time to attach a gdb by running the above "
 334                   "command, set up breakpoints etc. and then resume VPP from "
 335                   "within gdb by issuing the 'continue' command")
 336         elif cls.debug_gdb:
 337             print("gdb " + cls.vpp_bin + " -ex 'attach %s'" % cls.vpp.pid)
 338             print("Now is the time to attach a gdb by running the above "
 339                   "command and set up breakpoints etc.")
 340         print(single_line_delim)
 341         raw_input("Press ENTER to continue running the testcase...")
 342
 343     @classmethod
 344     def run_vpp(cls):
 345         cmdline = cls.vpp_cmdline
 346
 347         if cls.debug_gdbserver:
 348             gdbserver = '/usr/bin/gdbserver'
 349             if not os.path.isfile(gdbserver) or \
 350                     not os.access(gdbserver, os.X_OK):
 351                 raise Exception("gdbserver binary '%s' does not exist or is "
 352                                 "not executable" % gdbserver)
 353
 354             cmdline = [gdbserver, 'localhost:7777'] + cls.vpp_cmdline
 355             cls.logger.info("Gdbserver cmdline is %s", " ".join(cmdline))
 356
 357         try:
 358             cls.vpp = subprocess.Popen(cmdline,
 359                                        stdout=subprocess.PIPE,
 360                                        stderr=subprocess.PIPE,
 361                                        bufsize=1)
 362         except subprocess.CalledProcessError as e:
 363             cls.logger.critical("Couldn't start vpp: %s" % e)
 364             raise
 365
 366         cls.wait_for_enter()
 367
 368     @classmethod
 369     def wait_for_stats_socket(cls):
 370         deadline = time.time() + 3
 371         ok = False
 372         while time.time() < deadline or \
 373                 cls.debug_gdb or cls.debug_gdbserver:
 374             if os.path.exists(cls.stats_sock):
 375                 ok = True
 376                 break
 377             time.sleep(0.8)
 378         if not ok:
 379             cls.logger.critical("Couldn't stat : {}".format(cls.stats_sock))
 380
 381     @classmethod
 382     def setUpClass(cls):
 383         """
 384         Perform class setup before running the testcase
 385         Remove shared memory files, start vpp and connect the vpp-api
 386         """
 387         gc.collect()  # run garbage collection first
 388         random.seed()
 389         cls.print_header()
 390         cls.logger = get_logger(cls.__name__)
 391         if hasattr(cls, 'parallel_handler'):
 392             cls.logger.addHandler(cls.parallel_handler)
 393             cls.logger.propagate = False
 394         cls.tempdir = tempfile.mkdtemp(
 395             prefix='vpp-unittest-%s-' % cls.__name__)
 396         cls.stats_sock = "%s/stats.sock" % cls.tempdir
 397         cls.file_handler = FileHandler("%s/log.txt" % cls.tempdir)
 398         cls.file_handler.setFormatter(
 399             Formatter(fmt='%(asctime)s,%(msecs)03d %(message)s',
 400                       datefmt="%H:%M:%S"))
 401         cls.file_handler.setLevel(DEBUG)
 402         cls.logger.addHandler(cls.file_handler)
 403         cls.shm_prefix = os.path.basename(cls.tempdir)
 404         os.chdir(cls.tempdir)
 405         cls.logger.info("Temporary dir is %s, shm prefix is %s",
 406                         cls.tempdir, cls.shm_prefix)
 407         cls.setUpConstants()
 408         cls.reset_packet_infos()
 409         cls._captures = []
 410         cls._zombie_captures = []
 411         cls.verbose = 0
 412         cls.vpp_dead = False
 413         cls.registry = VppObjectRegistry()
 414         cls.vpp_startup_failed = False
 415         cls.reporter = KeepAliveReporter()
 416         # need to catch exceptions here because if we raise, then the cleanup
 417         # doesn't get called and we might end with a zombie vpp
 418         try:
 419             cls.run_vpp()
 420             cls.reporter.send_keep_alive(cls, 'setUpClass')
 421             VppTestResult.current_test_case_info = TestCaseInfo(
 422                 cls.logger, cls.tempdir, cls.vpp.pid, cls.vpp_bin)
 423             cls.vpp_stdout_deque = deque()
 424             cls.vpp_stderr_deque = deque()
 425             cls.pump_thread_stop_flag = Event()
 426             cls.pump_thread_wakeup_pipe = os.pipe()
 427             cls.pump_thread = Thread(target=pump_output, args=(cls,))
 428             cls.pump_thread.daemon = True
 429             cls.pump_thread.start()
 430             if cls.debug_gdb or cls.debug_gdbserver:
 431                 read_timeout = 0
 432             else:
 433                 read_timeout = 5
 434             cls.vapi = VppPapiProvider(cls.shm_prefix, cls.shm_prefix, cls,
 435                                        read_timeout)
 436             if cls.step:
 437                 hook = StepHook(cls)
 438             else:
 439                 hook = PollHook(cls)
 440             cls.vapi.register_hook(hook)
 441             cls.wait_for_stats_socket()
 442             cls.statistics = VPPStats(socketname=cls.stats_sock)
 443             try:
 444                 hook.poll_vpp()
 445             except VppDiedError:
 446                 cls.vpp_startup_failed = True
 447                 cls.logger.critical(
 448                     "VPP died shortly after startup, check the"
 449                     " output to standard error for possible cause")
 450                 raise
 451             try:
 452                 cls.vapi.connect()
 453             except Exception:
 454                 try:
 455                     cls.vapi.disconnect()
 456                 except Exception:
 457                     pass
 458                 if cls.debug_gdbserver:
 459                     print(colorize("You're running VPP inside gdbserver but "
 460                                    "VPP-API connection failed, did you forget "
 461                                    "to 'continue' VPP from within gdb?", RED))
 462                 raise
 463         except Exception:
 464             try:
 465                 cls.quit()
 466             except Exception:
 467                 pass
 468             raise
 469
 470     @classmethod
 471     def quit(cls):
 472         """
 473         Disconnect vpp-api, kill vpp and cleanup shared memory files
 474         """
 475         if (cls.debug_gdbserver or cls.debug_gdb) and hasattr(cls, 'vpp'):
 476             cls.vpp.poll()
 477             if cls.vpp.returncode is None:
 478                 print(double_line_delim)
 479                 print("VPP or GDB server is still running")
 480                 print(single_line_delim)
 481                 raw_input("When done debugging, press ENTER to kill the "
 482                           "process and finish running the testcase...")
 483
 484         # first signal that we want to stop the pump thread, then wake it up
 485         if hasattr(cls, 'pump_thread_stop_flag'):
 486             cls.pump_thread_stop_flag.set()
 487         if hasattr(cls, 'pump_thread_wakeup_pipe'):
 488             os.write(cls.pump_thread_wakeup_pipe[1], b'ding dong wake up')
 489         if hasattr(cls, 'pump_thread'):
 490             cls.logger.debug("Waiting for pump thread to stop")
 491             cls.pump_thread.join()
 492         if hasattr(cls, 'vpp_stderr_reader_thread'):
 493             cls.logger.debug("Waiting for stdderr pump to stop")
 494             cls.vpp_stderr_reader_thread.join()
 495
 496         if hasattr(cls, 'vpp'):
 497             if hasattr(cls, 'vapi'):
 498                 cls.vapi.disconnect()
 499                 del cls.vapi
 500             cls.vpp.poll()
 501             if cls.vpp.returncode is None:
 502                 cls.logger.debug("Sending TERM to vpp")
 503                 cls.vpp.kill()
 504                 cls.logger.debug("Waiting for vpp to die")
 505                 cls.vpp.communicate()
 506             del cls.vpp
 507
 508         if cls.vpp_startup_failed:
 509             stdout_log = cls.logger.info
 510             stderr_log = cls.logger.critical
 511         else:
 512             stdout_log = cls.logger.info
 513             stderr_log = cls.logger.info
 514
 515         if hasattr(cls, 'vpp_stdout_deque'):
 516             stdout_log(single_line_delim)
 517             stdout_log('VPP output to stdout while running %s:', cls.__name__)
 518             stdout_log(single_line_delim)
 519             vpp_output = "".join(cls.vpp_stdout_deque)
 520             with open(cls.tempdir + '/vpp_stdout.txt', 'w') as f:
 521                 f.write(vpp_output)
 522             stdout_log('\n%s', vpp_output)
 523             stdout_log(single_line_delim)
 524
 525         if hasattr(cls, 'vpp_stderr_deque'):
 526             stderr_log(single_line_delim)
 527             stderr_log('VPP output to stderr while running %s:', cls.__name__)
 528             stderr_log(single_line_delim)
 529             vpp_output = "".join(str(cls.vpp_stderr_deque))
 530             with open(cls.tempdir + '/vpp_stderr.txt', 'w') as f:
 531                 f.write(vpp_output)
 532             stderr_log('\n%s', vpp_output)
 533             stderr_log(single_line_delim)
 534
 535     @classmethod
 536     def tearDownClass(cls):
 537         """ Perform final cleanup after running all tests in this test-case """
 538         cls.reporter.send_keep_alive(cls, 'tearDownClass')
 539         cls.quit()
 540         cls.file_handler.close()
 541         cls.reset_packet_infos()
 542         if debug_framework:
 543             debug_internal.on_tear_down_class(cls)
 544
 545     def tearDown(self):
 546         """ Show various debug prints after each test """
 547         self.logger.debug("--- tearDown() for %s.%s(%s) called ---" %
 548                           (self.__class__.__name__, self._testMethodName,
 549                            self._testMethodDoc))
 550         if not self.vpp_dead:
 551             self.logger.debug(self.vapi.cli("show trace"))
 552             self.logger.info(self.vapi.ppcli("show interface"))
 553             self.logger.info(self.vapi.ppcli("show hardware"))
 554             self.logger.info(self.statistics.set_errors_str())
 555             self.logger.info(self.vapi.ppcli("show run"))
 556             self.logger.info(self.vapi.ppcli("show log"))
 557             self.registry.remove_vpp_config(self.logger)
 558             # Save/Dump VPP api trace log
 559             api_trace = "vpp_api_trace.%s.log" % self._testMethodName
 560             tmp_api_trace = "/tmp/%s" % api_trace
 561             vpp_api_trace_log = "%s/%s" % (self.tempdir, api_trace)
 562             self.logger.info(self.vapi.ppcli("api trace save %s" % api_trace))
 563             self.logger.info("Moving %s to %s\n" % (tmp_api_trace,
 564                                                     vpp_api_trace_log))
 565             os.rename(tmp_api_trace, vpp_api_trace_log)
 566             self.logger.info(self.vapi.ppcli("api trace custom-dump %s" %
 567                                              vpp_api_trace_log))
 568         else:
 569             self.registry.unregister_all(self.logger)
 570
 571     def setUp(self):
 572         """ Clear trace before running each test"""
 573         self.reporter.send_keep_alive(self)
 574         self.logger.debug("--- setUp() for %s.%s(%s) called ---" %
 575                           (self.__class__.__name__, self._testMethodName,
 576                            self._testMethodDoc))
 577         if self.vpp_dead:
 578             raise Exception("VPP is dead when setting up the test")
 579         self.sleep(.1, "during setUp")
 580         self.vpp_stdout_deque.append(
 581             "--- test setUp() for %s.%s(%s) starts here ---\n" %
 582             (self.__class__.__name__, self._testMethodName,
 583              self._testMethodDoc))
 584         self.vpp_stderr_deque.append(
 585             "--- test setUp() for %s.%s(%s) starts here ---\n" %
 586             (self.__class__.__name__, self._testMethodName,
 587              self._testMethodDoc))
 588         self.vapi.cli("clear trace")
 589         # store the test instance inside the test class - so that objects
 590         # holding the class can access instance methods (like assertEqual)
 591         type(self).test_instance = self
 592
 593     @classmethod
 594     def pg_enable_capture(cls, interfaces=None):
 595         """
 596         Enable capture on packet-generator interfaces
 597
 598         :param interfaces: iterable interface indexes (if None,
 599                            use self.pg_interfaces)
 600
 601         """
 602         if interfaces is None:
 603             interfaces = cls.pg_interfaces
 604         for i in interfaces:
 605             i.enable_capture()
 606
 607     @classmethod
 608     def register_capture(cls, cap_name):
 609         """ Register a capture in the testclass """
 610         # add to the list of captures with current timestamp
 611         cls._captures.append((time.time(), cap_name))
 612         # filter out from zombies
 613         cls._zombie_captures = [(stamp, name)
 614                                 for (stamp, name) in cls._zombie_captures
 615                                 if name != cap_name]
 616
 617     @classmethod
 618     def pg_start(cls):
 619         """ Remove any zombie captures and enable the packet generator """
 620         # how long before capture is allowed to be deleted - otherwise vpp
 621         # crashes - 100ms seems enough (this shouldn't be needed at all)
 622         capture_ttl = 0.1
 623         now = time.time()
 624         for stamp, cap_name in cls._zombie_captures:
 625             wait = stamp + capture_ttl - now
 626             if wait > 0:
 627                 cls.sleep(wait, "before deleting capture %s" % cap_name)
 628                 now = time.time()
 629             cls.logger.debug("Removing zombie capture %s" % cap_name)
 630             cls.vapi.cli('packet-generator delete %s' % cap_name)
 631
 632         cls.vapi.cli("trace add pg-input 50")  # 50 is maximum
 633         cls.vapi.cli('packet-generator enable')
 634         cls._zombie_captures = cls._captures
 635         cls._captures = []
 636
 637     @classmethod
 638     def create_pg_interfaces(cls, interfaces):
 639         """
 640         Create packet-generator interfaces.
 641
 642         :param interfaces: iterable indexes of the interfaces.
 643         :returns: List of created interfaces.
 644
 645         """
 646         result = []
 647         for i in interfaces:
 648             intf = VppPGInterface(cls, i)
 649             setattr(cls, intf.name, intf)
 650             result.append(intf)
 651         cls.pg_interfaces = result
 652         return result
 653
 654     @classmethod
 655     def create_loopback_interfaces(cls, count):
 656         """
 657         Create loopback interfaces.
 658
 659         :param count: number of interfaces created.
 660         :returns: List of created interfaces.
 661         """
 662         result = [VppLoInterface(cls) for i in range(count)]
 663         for intf in result:
 664             setattr(cls, intf.name, intf)
 665         cls.lo_interfaces = result
 666         return result
 667
 668     @staticmethod
 669     def extend_packet(packet, size, padding=' '):
 670         """
 671         Extend packet to given size by padding with spaces or custom padding
 672         NOTE: Currently works only when Raw layer is present.
 673
 674         :param packet: packet
 675         :param size: target size
 676         :param padding: padding used to extend the payload
 677
 678         """
 679         packet_len = len(packet) + 4
 680         extend = size - packet_len
 681         if extend > 0:
 682             num = (extend / len(padding)) + 1
 683             packet[Raw].load += (padding * num)[:extend]
 684
 685     @classmethod
 686     def reset_packet_infos(cls):
 687         """ Reset the list of packet info objects and packet counts to zero """
 688         cls._packet_infos = {}
 689         cls._packet_count_for_dst_if_idx = {}
 690
 691     @classmethod
 692     def create_packet_info(cls, src_if, dst_if):
 693         """
 694         Create packet info object containing the source and destination indexes
 695         and add it to the testcase's packet info list
 696
 697         :param VppInterface src_if: source interface
 698         :param VppInterface dst_if: destination interface
 699
 700         :returns: _PacketInfo object
 701
 702         """
 703         info = _PacketInfo()
 704         info.index = len(cls._packet_infos)
 705         info.src = src_if.sw_if_index
 706         info.dst = dst_if.sw_if_index
 707         if isinstance(dst_if, VppSubInterface):
 708             dst_idx = dst_if.parent.sw_if_index
 709         else:
 710             dst_idx = dst_if.sw_if_index
 711         if dst_idx in cls._packet_count_for_dst_if_idx:
 712             cls._packet_count_for_dst_if_idx[dst_idx] += 1
 713         else:
 714             cls._packet_count_for_dst_if_idx[dst_idx] = 1
 715         cls._packet_infos[info.index] = info
 716         return info
 717
 718     @staticmethod
 719     def info_to_payload(info):
 720         """
 721         Convert _PacketInfo object to packet payload
 722
 723         :param info: _PacketInfo object
 724
 725         :returns: string containing serialized data from packet info
 726         """
 727         return "%d %d %d %d %d" % (info.index, info.src, info.dst,
 728                                    info.ip, info.proto)
 729
 730     @staticmethod
 731     def payload_to_info(payload):
 732         """
 733         Convert packet payload to _PacketInfo object
 734
 735         :param payload: packet payload
 736
 737         :returns: _PacketInfo object containing de-serialized data from payload
 738
 739         """
 740         numbers = payload.split()
 741         info = _PacketInfo()
 742         info.index = int(numbers[0])
 743         info.src = int(numbers[1])
 744         info.dst = int(numbers[2])
 745         info.ip = int(numbers[3])
 746         info.proto = int(numbers[4])
 747         return info
 748
 749     def get_next_packet_info(self, info):
 750         """
 751         Iterate over the packet info list stored in the testcase
 752         Start iteration with first element if info is None
 753         Continue based on index in info if info is specified
 754
 755         :param info: info or None
 756         :returns: next info in list or None if no more infos
 757         """
 758         if info is None:
 759             next_index = 0
 760         else:
 761             next_index = info.index + 1
 762         if next_index == len(self._packet_infos):
 763             return None
 764         else:
 765             return self._packet_infos[next_index]
 766
 767     def get_next_packet_info_for_interface(self, src_index, info):
 768         """
 769         Search the packet info list for the next packet info with same source
 770         interface index
 771
 772         :param src_index: source interface index to search for
 773         :param info: packet info - where to start the search
 774         :returns: packet info or None
 775
 776         """
 777         while True:
 778             info = self.get_next_packet_info(info)
 779             if info is None:
 780                 return None
 781             if info.src == src_index:
 782                 return info
 783
 784     def get_next_packet_info_for_interface2(self, src_index, dst_index, info):
 785         """
 786         Search the packet info list for the next packet info with same source
 787         and destination interface indexes
 788
 789         :param src_index: source interface index to search for
 790         :param dst_index: destination interface index to search for
 791         :param info: packet info - where to start the search
 792         :returns: packet info or None
 793
 794         """
 795         while True:
 796             info = self.get_next_packet_info_for_interface(src_index, info)
 797             if info is None:
 798                 return None
 799             if info.dst == dst_index:
 800                 return info
 801
 802     def assert_equal(self, real_value, expected_value, name_or_class=None):
 803         if name_or_class is None:
 804             self.assertEqual(real_value, expected_value)
 805             return
 806         try:
 807             msg = "Invalid %s: %d('%s') does not match expected value %d('%s')"
 808             msg = msg % (getdoc(name_or_class).strip(),
 809                          real_value, str(name_or_class(real_value)),
 810                          expected_value, str(name_or_class(expected_value)))
 811         except Exception:
 812             msg = "Invalid %s: %s does not match expected value %s" % (
 813                 name_or_class, real_value, expected_value)
 814
 815         self.assertEqual(real_value, expected_value, msg)
 816
 817     def assert_in_range(self,
 818                         real_value,
 819                         expected_min,
 820                         expected_max,
 821                         name=None):
 822         if name is None:
 823             msg = None
 824         else:
 825             msg = "Invalid %s: %s out of range <%s,%s>" % (
 826                 name, real_value, expected_min, expected_max)
 827         self.assertTrue(expected_min <= real_value <= expected_max, msg)
 828
 829     def assert_packet_checksums_valid(self, packet,
 830                                       ignore_zero_udp_checksums=True):
 831         received = packet.__class__(str(packet))
 832         self.logger.debug(
 833             ppp("Verifying packet checksums for packet:", received))
 834         udp_layers = ['UDP', 'UDPerror']
 835         checksum_fields = ['cksum', 'chksum']
 836         checksums = []
 837         counter = 0
 838         temp = received.__class__(str(received))
 839         while True:
 840             layer = temp.getlayer(counter)
 841             if layer:
 842                 for cf in checksum_fields:
 843                     if hasattr(layer, cf):
 844                         if ignore_zero_udp_checksums and \
 845                                         0 == getattr(layer, cf) and \
 846                                         layer.name in udp_layers:
 847                             continue
 848                         delattr(layer, cf)
 849                         checksums.append((counter, cf))
 850             else:
 851                 break
 852             counter = counter + 1
 853         if 0 == len(checksums):
 854             return
 855         temp = temp.__class__(str(temp))
 856         for layer, cf in checksums:
 857             calc_sum = getattr(temp[layer], cf)
 858             self.assert_equal(
 859                 getattr(received[layer], cf), calc_sum,
 860                 "packet checksum on layer #%d: %s" % (layer, temp[layer].name))
 861             self.logger.debug(
 862                 "Checksum field `%s` on `%s` layer has correct value `%s`" %
 863                 (cf, temp[layer].name, calc_sum))
 864
 865     def assert_checksum_valid(self, received_packet, layer,
 866                               field_name='chksum',
 867                               ignore_zero_checksum=False):
 868         """ Check checksum of received packet on given layer """
 869         received_packet_checksum = getattr(received_packet[layer], field_name)
 870         if ignore_zero_checksum and 0 == received_packet_checksum:
 871             return
 872         recalculated = received_packet.__class__(str(received_packet))
 873         delattr(recalculated[layer], field_name)
 874         recalculated = recalculated.__class__(str(recalculated))
 875         self.assert_equal(received_packet_checksum,
 876                           getattr(recalculated[layer], field_name),
 877                           "packet checksum on layer: %s" % layer)
 878
 879     def assert_ip_checksum_valid(self, received_packet,
 880                                  ignore_zero_checksum=False):
 881         self.assert_checksum_valid(received_packet, 'IP',
 882                                    ignore_zero_checksum=ignore_zero_checksum)
 883
 884     def assert_tcp_checksum_valid(self, received_packet,
 885                                   ignore_zero_checksum=False):
 886         self.assert_checksum_valid(received_packet, 'TCP',
 887                                    ignore_zero_checksum=ignore_zero_checksum)
 888
 889     def assert_udp_checksum_valid(self, received_packet,
 890                                   ignore_zero_checksum=True):
 891         self.assert_checksum_valid(received_packet, 'UDP',
 892                                    ignore_zero_checksum=ignore_zero_checksum)
 893
 894     def assert_embedded_icmp_checksum_valid(self, received_packet):
 895         if received_packet.haslayer(IPerror):
 896             self.assert_checksum_valid(received_packet, 'IPerror')
 897         if received_packet.haslayer(TCPerror):
 898             self.assert_checksum_valid(received_packet, 'TCPerror')
 899         if received_packet.haslayer(UDPerror):
 900             self.assert_checksum_valid(received_packet, 'UDPerror',
 901                                        ignore_zero_checksum=True)
 902         if received_packet.haslayer(ICMPerror):
 903             self.assert_checksum_valid(received_packet, 'ICMPerror')
 904
 905     def assert_icmp_checksum_valid(self, received_packet):
 906         self.assert_checksum_valid(received_packet, 'ICMP')
 907         self.assert_embedded_icmp_checksum_valid(received_packet)
 908
 909     def assert_icmpv6_checksum_valid(self, pkt):
 910         if pkt.haslayer(ICMPv6DestUnreach):
 911             self.assert_checksum_valid(pkt, 'ICMPv6DestUnreach', 'cksum')
 912             self.assert_embedded_icmp_checksum_valid(pkt)
 913         if pkt.haslayer(ICMPv6EchoRequest):
 914             self.assert_checksum_valid(pkt, 'ICMPv6EchoRequest', 'cksum')
 915         if pkt.haslayer(ICMPv6EchoReply):
 916             self.assert_checksum_valid(pkt, 'ICMPv6EchoReply', 'cksum')
 917
 918     def assert_packet_counter_equal(self, counter, expected_value):
 919         counters = self.vapi.cli("sh errors").split('\n')
 920         counter_value = -1
 921         for i in range(1, len(counters)):
 922             results = counters[i].split()
 923             if results[1] == counter:
 924                 counter_value = int(results[0])
 925                 break
 926         self.assert_equal(counter_value, expected_value,
 927                           "packet counter `%s'" % counter)
 928
 929     @classmethod
 930     def sleep(cls, timeout, remark=None):
 931         if hasattr(cls, 'logger'):
 932             cls.logger.debug("Starting sleep for %es (%s)", timeout, remark)
 933         before = time.time()
 934         time.sleep(timeout)
 935         after = time.time()
 936         if hasattr(cls, 'logger') and after - before > 2 * timeout:
 937             cls.logger.error("unexpected time.sleep() result - "
 938                              "slept for %es instead of ~%es!",
 939                              after - before, timeout)
 940         if hasattr(cls, 'logger'):
 941             cls.logger.debug(
 942                 "Finished sleep (%s) - slept %es (wanted %es)",
 943                 remark, after - before, timeout)
 944
 945     def send_and_assert_no_replies(self, intf, pkts, remark="", timeout=None):
 946         self.vapi.cli("clear trace")
 947         intf.add_stream(pkts)
 948         self.pg_enable_capture(self.pg_interfaces)
 949         self.pg_start()
 950         if not timeout:
 951             timeout = 1
 952         for i in self.pg_interfaces:
 953             i.get_capture(0, timeout=timeout)
 954             i.assert_nothing_captured(remark=remark)
 955             timeout = 0.1
 956
 957     def send_and_expect(self, input, pkts, output):
 958         self.vapi.cli("clear trace")
 959         input.add_stream(pkts)
 960         self.pg_enable_capture(self.pg_interfaces)
 961         self.pg_start()
 962         if isinstance(object, (list,)):
 963             rx = []
 964             for o in output:
 965                 rx.append(output.get_capture(len(pkts)))
 966         else:
 967             rx = output.get_capture(len(pkts))
 968         return rx
 969
 970     def send_and_expect_only(self, input, pkts, output, timeout=None):
 971         self.vapi.cli("clear trace")
 972         input.add_stream(pkts)
 973         self.pg_enable_capture(self.pg_interfaces)
 974         self.pg_start()
 975         if isinstance(object, (list,)):
 976             outputs = output
 977             rx = []
 978             for o in outputs:
 979                 rx.append(output.get_capture(len(pkts)))
 980         else:
 981             rx = output.get_capture(len(pkts))
 982             outputs = [output]
 983         if not timeout:
 984             timeout = 1
 985         for i in self.pg_interfaces:
 986             if i not in outputs:
 987                 i.get_capture(0, timeout=timeout)
 988                 i.assert_nothing_captured()
 989                 timeout = 0.1
 990
 991         return rx
 992
 993
 994 def get_testcase_doc_name(test):
 995     return getdoc(test.__class__).splitlines()[0]
 996
 997
 998 def get_test_description(descriptions, test):
 999     short_description = test.shortDescription()
1000     if descriptions and short_description:
1001         return short_description
1002     else:
1003         return str(test)
1004
1005
1006 class TestCaseInfo(object):
1007     def __init__(self, logger, tempdir, vpp_pid, vpp_bin_path):
1008         self.logger = logger
1009         self.tempdir = tempdir
1010         self.vpp_pid = vpp_pid
1011         self.vpp_bin_path = vpp_bin_path
1012         self.core_crash_test = None
1013
1014
1015 class VppTestResult(unittest.TestResult):
1016     """
1017     @property result_string
1018      String variable to store the test case result string.
1019     @property errors
1020      List variable containing 2-tuples of TestCase instances and strings
1021      holding formatted tracebacks. Each tuple represents a test which
1022      raised an unexpected exception.
1023     @property failures
1024      List variable containing 2-tuples of TestCase instances and strings
1025      holding formatted tracebacks. Each tuple represents a test where
1026      a failure was explicitly signalled using the TestCase.assert*()
1027      methods.
1028     """
1029
1030     failed_test_cases_info = set()
1031     core_crash_test_cases_info = set()
1032     current_test_case_info = None
1033
1034     def __init__(self, stream, descriptions, verbosity, runner):
1035         """
1036         :param stream File descriptor to store where to report test results.
1037             Set to the standard error stream by default.
1038         :param descriptions Boolean variable to store information if to use
1039             test case descriptions.
1040         :param verbosity Integer variable to store required verbosity level.
1041         """
1042         unittest.TestResult.__init__(self, stream, descriptions, verbosity)
1043         self.stream = stream
1044         self.descriptions = descriptions
1045         self.verbosity = verbosity
1046         self.result_string = None
1047         self.runner = runner
1048
1049     def addSuccess(self, test):
1050         """
1051         Record a test succeeded result
1052
1053         :param test:
1054
1055         """
1056         if self.current_test_case_info:
1057             self.current_test_case_info.logger.debug(
1058                 "--- addSuccess() %s.%s(%s) called" % (test.__class__.__name__,
1059                                                        test._testMethodName,
1060                                                        test._testMethodDoc))
1061         unittest.TestResult.addSuccess(self, test)
1062         self.result_string = colorize("OK", GREEN)
1063
1064         self.send_result_through_pipe(test, PASS)
1065
1066     def addSkip(self, test, reason):
1067         """
1068         Record a test skipped.
1069
1070         :param test:
1071         :param reason:
1072
1073         """
1074         if self.current_test_case_info:
1075             self.current_test_case_info.logger.debug(
1076                 "--- addSkip() %s.%s(%s) called, reason is %s" %
1077                 (test.__class__.__name__, test._testMethodName,
1078                  test._testMethodDoc, reason))
1079         unittest.TestResult.addSkip(self, test, reason)
1080         self.result_string = colorize("SKIP", YELLOW)
1081
1082         self.send_result_through_pipe(test, SKIP)
1083
1084     def symlink_failed(self):
1085         if self.current_test_case_info:
1086             try:
1087                 failed_dir = os.getenv('FAILED_DIR')
1088                 link_path = os.path.join(
1089                     failed_dir,
1090                     '%s-FAILED' %
1091                     os.path.basename(self.current_test_case_info.tempdir))
1092                 if self.current_test_case_info.logger:
1093                     self.current_test_case_info.logger.debug(
1094                         "creating a link to the failed test")
1095                     self.current_test_case_info.logger.debug(
1096                         "os.symlink(%s, %s)" %
1097                         (self.current_test_case_info.tempdir, link_path))
1098                 if os.path.exists(link_path):
1099                     if self.current_test_case_info.logger:
1100                         self.current_test_case_info.logger.debug(
1101                             'symlink already exists')
1102                 else:
1103                     os.symlink(self.current_test_case_info.tempdir, link_path)
1104
1105             except Exception as e:
1106                 if self.current_test_case_info.logger:
1107                     self.current_test_case_info.logger.error(e)
1108
1109     def send_result_through_pipe(self, test, result):
1110         if hasattr(self, 'test_framework_result_pipe'):
1111             pipe = self.test_framework_result_pipe
1112             if pipe:
1113                 pipe.send((test.id(), result))
1114
1115     def log_error(self, test, err, fn_name):
1116         if self.current_test_case_info:
1117             if isinstance(test, unittest.suite._ErrorHolder):
1118                 test_name = test.description
1119             else:
1120                 test_name = '%s.%s(%s)' % (test.__class__.__name__,
1121                                            test._testMethodName,
1122                                            test._testMethodDoc)
1123             self.current_test_case_info.logger.debug(
1124                 "--- %s() %s called, err is %s" %
1125                 (fn_name, test_name, err))
1126             self.current_test_case_info.logger.debug(
1127                 "formatted exception is:\n%s" %
1128                 "".join(format_exception(*err)))
1129
1130     def add_error(self, test, err, unittest_fn, error_type):
1131         if error_type == FAIL:
1132             self.log_error(test, err, 'addFailure')
1133             error_type_str = colorize("FAIL", RED)
1134         elif error_type == ERROR:
1135             self.log_error(test, err, 'addError')
1136             error_type_str = colorize("ERROR", RED)
1137         else:
1138             raise Exception('Error type %s cannot be used to record an '
1139                             'error or a failure' % error_type)
1140
1141         unittest_fn(self, test, err)
1142         if self.current_test_case_info:
1143             self.result_string = "%s [ temp dir used by test case: %s ]" % \
1144                                  (error_type_str,
1145                                   self.current_test_case_info.tempdir)
1146             self.symlink_failed()
1147             self.failed_test_cases_info.add(self.current_test_case_info)
1148             if is_core_present(self.current_test_case_info.tempdir):
1149                 if not self.current_test_case_info.core_crash_test:
1150                     if isinstance(test, unittest.suite._ErrorHolder):
1151                         test_name = str(test)
1152                     else:
1153                         test_name = "'{}' ({})".format(
1154                             get_testcase_doc_name(test), test.id())
1155                     self.current_test_case_info.core_crash_test = test_name
1156                 self.core_crash_test_cases_info.add(
1157                     self.current_test_case_info)
1158         else:
1159             self.result_string = '%s [no temp dir]' % error_type_str
1160
1161         self.send_result_through_pipe(test, error_type)
1162
1163     def addFailure(self, test, err):
1164         """
1165         Record a test failed result
1166
1167         :param test:
1168         :param err: error message
1169
1170         """
1171         self.add_error(test, err, unittest.TestResult.addFailure, FAIL)
1172
1173     def addError(self, test, err):
1174         """
1175         Record a test error result
1176
1177         :param test:
1178         :param err: error message
1179
1180         """
1181         self.add_error(test, err, unittest.TestResult.addError, ERROR)
1182
1183     def getDescription(self, test):
1184         """
1185         Get test description
1186
1187         :param test:
1188         :returns: test description
1189
1190         """
1191         return get_test_description(self.descriptions, test)
1192
1193     def startTest(self, test):
1194         """
1195         Start a test
1196
1197         :param test:
1198
1199         """
1200         test.print_header()
1201
1202         unittest.TestResult.startTest(self, test)
1203         if self.verbosity > 0:
1204             self.stream.writeln(
1205                 "Starting " + self.getDescription(test) + " ...")
1206             self.stream.writeln(single_line_delim)
1207
1208     def stopTest(self, test):
1209         """
1210         Called when the given test has been run
1211
1212         :param test:
1213
1214         """
1215         unittest.TestResult.stopTest(self, test)
1216         if self.verbosity > 0:
1217             self.stream.writeln(single_line_delim)
1218             self.stream.writeln("%-73s%s" % (self.getDescription(test),
1219                                              self.result_string))
1220             self.stream.writeln(single_line_delim)
1221         else:
1222             self.stream.writeln("%-73s%s" % (self.getDescription(test),
1223                                              self.result_string))
1224
1225         self.send_result_through_pipe(test, TEST_RUN)
1226
1227     def printErrors(self):
1228         """
1229         Print errors from running the test case
1230         """
1231         if len(self.errors) > 0 or len(self.failures) > 0:
1232             self.stream.writeln()
1233             self.printErrorList('ERROR', self.errors)
1234             self.printErrorList('FAIL', self.failures)
1235
1236         # ^^ that is the last output from unittest before summary
1237         if not self.runner.print_summary:
1238             devnull = unittest.runner._WritelnDecorator(open(os.devnull, 'w'))
1239             self.stream = devnull
1240             self.runner.stream = devnull
1241
1242     def printErrorList(self, flavour, errors):
1243         """
1244         Print error list to the output stream together with error type
1245         and test case description.
1246
1247         :param flavour: error type
1248         :param errors: iterable errors
1249
1250         """
1251         for test, err in errors:
1252             self.stream.writeln(double_line_delim)
1253             self.stream.writeln("%s: %s" %
1254                                 (flavour, self.getDescription(test)))
1255             self.stream.writeln(single_line_delim)
1256             self.stream.writeln("%s" % err)
1257
1258
1259 class VppTestRunner(unittest.TextTestRunner):
1260     """
1261     A basic test runner implementation which prints results to standard error.
1262     """
1263
1264     @property
1265     def resultclass(self):
1266         """Class maintaining the results of the tests"""
1267         return VppTestResult
1268
1269     def __init__(self, keep_alive_pipe=None, descriptions=True, verbosity=1,
1270                  result_pipe=None, failfast=False, buffer=False,
1271                  resultclass=None, print_summary=True):
1272         # ignore stream setting here, use hard-coded stdout to be in sync
1273         # with prints from VppTestCase methods ...
1274         super(VppTestRunner, self).__init__(sys.stdout, descriptions,
1275                                             verbosity, failfast, buffer,
1276                                             resultclass)
1277         KeepAliveReporter.pipe = keep_alive_pipe
1278
1279         self.orig_stream = self.stream
1280         self.resultclass.test_framework_result_pipe = result_pipe
1281
1282         self.print_summary = print_summary
1283
1284     def _makeResult(self):
1285         return self.resultclass(self.stream,
1286                                 self.descriptions,
1287                                 self.verbosity,
1288                                 self)
1289
1290     def run(self, test):
1291         """
1292         Run the tests
1293
1294         :param test:
1295
1296         """
1297         faulthandler.enable()  # emit stack trace to stderr if killed by signal
1298
1299         result = super(VppTestRunner, self).run(test)
1300         if not self.print_summary:
1301             self.stream = self.orig_stream
1302             result.stream = self.orig_stream
1303         return result
1304
1305
1306 class Worker(Thread):
1307     def __init__(self, args, logger, env={}):
1308         self.logger = logger
1309         self.args = args
1310         self.result = None
1311         self.env = copy.deepcopy(env)
1312         super(Worker, self).__init__()
1313
1314     def run(self):
1315         executable = self.args[0]
1316         self.logger.debug("Running executable w/args `%s'" % self.args)
1317         env = os.environ.copy()
1318         env.update(self.env)
1319         env["CK_LOG_FILE_NAME"] = "-"
1320         self.process = subprocess.Popen(
1321             self.args, shell=False, env=env, preexec_fn=os.setpgrp,
1322             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
1323         out, err = self.process.communicate()
1324         self.logger.debug("Finished running `%s'" % executable)
1325         self.logger.info("Return code is `%s'" % self.process.returncode)
1326         self.logger.info(single_line_delim)
1327         self.logger.info("Executable `%s' wrote to stdout:" % executable)
1328         self.logger.info(single_line_delim)
1329         self.logger.info(out)
1330         self.logger.info(single_line_delim)
1331         self.logger.info("Executable `%s' wrote to stderr:" % executable)
1332         self.logger.info(single_line_delim)
1333         self.logger.info(err)
1334         self.logger.info(single_line_delim)
1335         self.result = self.process.returncode