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