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 resource
  12 from collections import deque
  13 from threading import Thread, Event
  14 from inspect import getdoc
  15 from traceback import format_exception
  16 from logging import FileHandler, DEBUG, Formatter
  17 from scapy.packet import Raw
  18 from hook import StepHook, PollHook
  19 from vpp_pg_interface import VppPGInterface
  20 from vpp_sub_interface import VppSubInterface
  21 from vpp_lo_interface import VppLoInterface
  22 from vpp_papi_provider import VppPapiProvider
  23 from log import *
  24 from vpp_object import VppObjectRegistry
  25 if os.name == 'posix' and sys.version_info[0] < 3:
  26     # using subprocess32 is recommended by python official documentation
  27     # @ https://docs.python.org/2/library/subprocess.html
  28     import subprocess32 as subprocess
  29 else:
  30     import subprocess
  31
  32 """
  33   Test framework module.
  34
  35   The module provides a set of tools for constructing and running tests and
  36   representing the results.
  37 """
  38
  39
  40 class _PacketInfo(object):
  41     """Private class to create packet info object.
  42
  43     Help process information about the next packet.
  44     Set variables to default values.
  45     """
  46     #: Store the index of the packet.
  47     index = -1
  48     #: Store the index of the source packet generator interface of the packet.
  49     src = -1
  50     #: Store the index of the destination packet generator interface
  51     #: of the packet.
  52     dst = -1
  53     #: Store the copy of the former packet.
  54     data = None
  55
  56     def __eq__(self, other):
  57         index = self.index == other.index
  58         src = self.src == other.src
  59         dst = self.dst == other.dst
  60         data = self.data == other.data
  61         return index and src and dst and data
  62
  63
  64 def pump_output(testclass):
  65     """ pump output from vpp stdout/stderr to proper queues """
  66     while not testclass.pump_thread_stop_flag.wait(0):
  67         readable = select.select([testclass.vpp.stdout.fileno(),
  68                                   testclass.vpp.stderr.fileno(),
  69                                   testclass.pump_thread_wakeup_pipe[0]],
  70                                  [], [])[0]
  71         if testclass.vpp.stdout.fileno() in readable:
  72             read = os.read(testclass.vpp.stdout.fileno(), 1024)
  73             testclass.vpp_stdout_deque.append(read)
  74         if testclass.vpp.stderr.fileno() in readable:
  75             read = os.read(testclass.vpp.stderr.fileno(), 1024)
  76             testclass.vpp_stderr_deque.append(read)
  77         # ignoring the dummy pipe here intentionally - the flag will take care
  78         # of properly terminating the loop
  79
  80
  81 class VppTestCase(unittest.TestCase):
  82     """This subclass is a base class for VPP test cases that are implemented as
  83     classes. It provides methods to create and run test case.
  84     """
  85
  86     @property
  87     def packet_infos(self):
  88         """List of packet infos"""
  89         return self._packet_infos
  90
  91     @classmethod
  92     def get_packet_count_for_if_idx(cls, dst_if_index):
  93         """Get the number of packet info for specified destination if index"""
  94         if dst_if_index in cls._packet_count_for_dst_if_idx:
  95             return cls._packet_count_for_dst_if_idx[dst_if_index]
  96         else:
  97             return 0
  98
  99     @classmethod
 100     def instance(cls):
 101         """Return the instance of this testcase"""
 102         return cls.test_instance
 103
 104     @classmethod
 105     def set_debug_flags(cls, d):
 106         cls.debug_core = False
 107         cls.debug_gdb = False
 108         cls.debug_gdbserver = False
 109         if d is None:
 110             return
 111         dl = d.lower()
 112         if dl == "core":
 113             cls.debug_core = True
 114         elif dl == "gdb":
 115             cls.debug_gdb = True
 116         elif dl == "gdbserver":
 117             cls.debug_gdbserver = True
 118         else:
 119             raise Exception("Unrecognized DEBUG option: '%s'" % d)
 120
 121     @classmethod
 122     def setUpConstants(cls):
 123         """ Set-up the test case class based on environment variables """
 124         try:
 125             s = os.getenv("STEP")
 126             cls.step = True if s.lower() in ("y", "yes", "1") else False
 127         except:
 128             cls.step = False
 129         try:
 130             d = os.getenv("DEBUG")
 131         except:
 132             d = None
 133         cls.set_debug_flags(d)
 134         cls.vpp_bin = os.getenv('VPP_TEST_BIN', "vpp")
 135         cls.plugin_path = os.getenv('VPP_TEST_PLUGIN_PATH')
 136         debug_cli = ""
 137         if cls.step or cls.debug_gdb or cls.debug_gdbserver:
 138             debug_cli = "cli-listen localhost:5002"
 139         coredump_size = None
 140         try:
 141             size = os.getenv("COREDUMP_SIZE")
 142             if size is not None:
 143                 coredump_size = "coredump-size %s" % size
 144         except:
 145             pass
 146         if coredump_size is None:
 147             coredump_size = "coredump-size unlimited"
 148         cls.vpp_cmdline = [cls.vpp_bin, "unix",
 149                            "{", "nodaemon", debug_cli, coredump_size, "}",
 150                            "api-trace", "{", "on", "}",
 151                            "api-segment", "{", "prefix", cls.shm_prefix, "}"]
 152         if cls.plugin_path is not None:
 153             cls.vpp_cmdline.extend(["plugin_path", cls.plugin_path])
 154         cls.logger.info("vpp_cmdline: %s" % cls.vpp_cmdline)
 155
 156     @classmethod
 157     def wait_for_enter(cls):
 158         if cls.debug_gdbserver:
 159             print(double_line_delim)
 160             print("Spawned GDB server with PID: %d" % cls.vpp.pid)
 161         elif cls.debug_gdb:
 162             print(double_line_delim)
 163             print("Spawned VPP with PID: %d" % cls.vpp.pid)
 164         else:
 165             cls.logger.debug("Spawned VPP with PID: %d" % cls.vpp.pid)
 166             return
 167         print(single_line_delim)
 168         print("You can debug the VPP using e.g.:")
 169         if cls.debug_gdbserver:
 170             print("gdb " + cls.vpp_bin + " -ex 'target remote localhost:7777'")
 171             print("Now is the time to attach a gdb by running the above "
 172                   "command, set up breakpoints etc. and then resume VPP from "
 173                   "within gdb by issuing the 'continue' command")
 174         elif cls.debug_gdb:
 175             print("gdb " + cls.vpp_bin + " -ex 'attach %s'" % cls.vpp.pid)
 176             print("Now is the time to attach a gdb by running the above "
 177                   "command and set up breakpoints etc.")
 178         print(single_line_delim)
 179         raw_input("Press ENTER to continue running the testcase...")
 180
 181     @classmethod
 182     def run_vpp(cls):
 183         cmdline = cls.vpp_cmdline
 184
 185         if cls.debug_gdbserver:
 186             gdbserver = '/usr/bin/gdbserver'
 187             if not os.path.isfile(gdbserver) or \
 188                     not os.access(gdbserver, os.X_OK):
 189                 raise Exception("gdbserver binary '%s' does not exist or is "
 190                                 "not executable" % gdbserver)
 191
 192             cmdline = [gdbserver, 'localhost:7777'] + cls.vpp_cmdline
 193             cls.logger.info("Gdbserver cmdline is %s", " ".join(cmdline))
 194
 195         try:
 196             cls.vpp = subprocess.Popen(cmdline,
 197                                        stdout=subprocess.PIPE,
 198                                        stderr=subprocess.PIPE,
 199                                        bufsize=1)
 200         except Exception as e:
 201             cls.logger.critical("Couldn't start vpp: %s" % e)
 202             raise
 203
 204         cls.wait_for_enter()
 205
 206     @classmethod
 207     def setUpClass(cls):
 208         """
 209         Perform class setup before running the testcase
 210         Remove shared memory files, start vpp and connect the vpp-api
 211         """
 212         gc.collect()  # run garbage collection first
 213         cls.logger = getLogger(cls.__name__)
 214         cls.tempdir = tempfile.mkdtemp(
 215             prefix='vpp-unittest-' + cls.__name__ + '-')
 216         file_handler = FileHandler("%s/log.txt" % cls.tempdir)
 217         file_handler.setFormatter(
 218             Formatter(fmt='%(asctime)s,%(msecs)03d %(message)s',
 219                       datefmt="%H:%M:%S"))
 220         file_handler.setLevel(DEBUG)
 221         cls.logger.addHandler(file_handler)
 222         cls.shm_prefix = cls.tempdir.split("/")[-1]
 223         os.chdir(cls.tempdir)
 224         cls.logger.info("Temporary dir is %s, shm prefix is %s",
 225                         cls.tempdir, cls.shm_prefix)
 226         cls.setUpConstants()
 227         cls.reset_packet_infos()
 228         cls._captures = []
 229         cls._zombie_captures = []
 230         cls.verbose = 0
 231         cls.vpp_dead = False
 232         cls.registry = VppObjectRegistry()
 233         print(double_line_delim)
 234         print(colorize(getdoc(cls).splitlines()[0], YELLOW))
 235         print(double_line_delim)
 236         # need to catch exceptions here because if we raise, then the cleanup
 237         # doesn't get called and we might end with a zombie vpp
 238         try:
 239             cls.run_vpp()
 240             cls.vpp_stdout_deque = deque()
 241             cls.vpp_stderr_deque = deque()
 242             cls.pump_thread_stop_flag = Event()
 243             cls.pump_thread_wakeup_pipe = os.pipe()
 244             cls.pump_thread = Thread(target=pump_output, args=(cls,))
 245             cls.pump_thread.daemon = True
 246             cls.pump_thread.start()
 247             cls.vapi = VppPapiProvider(cls.shm_prefix, cls.shm_prefix, cls)
 248             if cls.step:
 249                 hook = StepHook(cls)
 250             else:
 251                 hook = PollHook(cls)
 252             cls.vapi.register_hook(hook)
 253             cls.sleep(0.1, "after vpp startup, before initial poll")
 254             hook.poll_vpp()
 255             try:
 256                 cls.vapi.connect()
 257             except:
 258                 if cls.debug_gdbserver:
 259                     print(colorize("You're running VPP inside gdbserver but "
 260                                    "VPP-API connection failed, did you forget "
 261                                    "to 'continue' VPP from within gdb?", RED))
 262                 raise
 263         except:
 264             t, v, tb = sys.exc_info()
 265             try:
 266                 cls.quit()
 267             except:
 268                 pass
 269             raise t, v, tb
 270
 271     @classmethod
 272     def quit(cls):
 273         """
 274         Disconnect vpp-api, kill vpp and cleanup shared memory files
 275         """
 276         if (cls.debug_gdbserver or cls.debug_gdb) and hasattr(cls, 'vpp'):
 277             cls.vpp.poll()
 278             if cls.vpp.returncode is None:
 279                 print(double_line_delim)
 280                 print("VPP or GDB server is still running")
 281                 print(single_line_delim)
 282                 raw_input("When done debugging, press ENTER to kill the "
 283                           "process and finish running the testcase...")
 284
 285         os.write(cls.pump_thread_wakeup_pipe[1], 'ding dong wake up')
 286         cls.pump_thread_stop_flag.set()
 287         if hasattr(cls, 'pump_thread'):
 288             cls.logger.debug("Waiting for pump thread to stop")
 289             cls.pump_thread.join()
 290         if hasattr(cls, 'vpp_stderr_reader_thread'):
 291             cls.logger.debug("Waiting for stdderr pump to stop")
 292             cls.vpp_stderr_reader_thread.join()
 293
 294         if hasattr(cls, 'vpp'):
 295             if hasattr(cls, 'vapi'):
 296                 cls.vapi.disconnect()
 297                 del cls.vapi
 298             cls.vpp.poll()
 299             if cls.vpp.returncode is None:
 300                 cls.logger.debug("Sending TERM to vpp")
 301                 cls.vpp.terminate()
 302                 cls.logger.debug("Waiting for vpp to die")
 303                 cls.vpp.communicate()
 304             del cls.vpp
 305
 306         if hasattr(cls, 'vpp_stdout_deque'):
 307             cls.logger.info(single_line_delim)
 308             cls.logger.info('VPP output to stdout while running %s:',
 309                             cls.__name__)
 310             cls.logger.info(single_line_delim)
 311             f = open(cls.tempdir + '/vpp_stdout.txt', 'w')
 312             vpp_output = "".join(cls.vpp_stdout_deque)
 313             f.write(vpp_output)
 314             cls.logger.info('\n%s', vpp_output)
 315             cls.logger.info(single_line_delim)
 316
 317         if hasattr(cls, 'vpp_stderr_deque'):
 318             cls.logger.info(single_line_delim)
 319             cls.logger.info('VPP output to stderr while running %s:',
 320                             cls.__name__)
 321             cls.logger.info(single_line_delim)
 322             f = open(cls.tempdir + '/vpp_stderr.txt', 'w')
 323             vpp_output = "".join(cls.vpp_stderr_deque)
 324             f.write(vpp_output)
 325             cls.logger.info('\n%s', vpp_output)
 326             cls.logger.info(single_line_delim)
 327
 328     @classmethod
 329     def tearDownClass(cls):
 330         """ Perform final cleanup after running all tests in this test-case """
 331         cls.quit()
 332
 333     def tearDown(self):
 334         """ Show various debug prints after each test """
 335         self.logger.debug("--- tearDown() for %s.%s(%s) called ---" %
 336                           (self.__class__.__name__, self._testMethodName,
 337                            self._testMethodDoc))
 338         if not self.vpp_dead:
 339             self.logger.debug(self.vapi.cli("show trace"))
 340             self.logger.info(self.vapi.ppcli("show int"))
 341             self.logger.info(self.vapi.ppcli("show hardware"))
 342             self.logger.info(self.vapi.ppcli("show error"))
 343             self.logger.info(self.vapi.ppcli("show run"))
 344             self.registry.remove_vpp_config(self.logger)
 345             # Save/Dump VPP api trace log
 346             api_trace = "vpp_api_trace.%s.log" % self._testMethodName
 347             tmp_api_trace = "/tmp/%s" % api_trace
 348             vpp_api_trace_log = "%s/%s" % (self.tempdir, api_trace)
 349             self.logger.info(self.vapi.ppcli("api trace save %s" % api_trace))
 350             self.logger.info("Moving %s to %s\n" % (tmp_api_trace,
 351                                                     vpp_api_trace_log))
 352             os.rename(tmp_api_trace, vpp_api_trace_log)
 353             self.logger.info(self.vapi.ppcli("api trace dump %s" %
 354                                              vpp_api_trace_log))
 355         else:
 356             self.registry.unregister_all(self.logger)
 357
 358     def setUp(self):
 359         """ Clear trace before running each test"""
 360         self.logger.debug("--- setUp() for %s.%s(%s) called ---" %
 361                           (self.__class__.__name__, self._testMethodName,
 362                            self._testMethodDoc))
 363         if self.vpp_dead:
 364             raise Exception("VPP is dead when setting up the test")
 365         self.sleep(.1, "during setUp")
 366         self.vpp_stdout_deque.append(
 367             "--- test setUp() for %s.%s(%s) starts here ---\n" %
 368             (self.__class__.__name__, self._testMethodName,
 369              self._testMethodDoc))
 370         self.vpp_stderr_deque.append(
 371             "--- test setUp() for %s.%s(%s) starts here ---\n" %
 372             (self.__class__.__name__, self._testMethodName,
 373              self._testMethodDoc))
 374         self.vapi.cli("clear trace")
 375         # store the test instance inside the test class - so that objects
 376         # holding the class can access instance methods (like assertEqual)
 377         type(self).test_instance = self
 378
 379     @classmethod
 380     def pg_enable_capture(cls, interfaces):
 381         """
 382         Enable capture on packet-generator interfaces
 383
 384         :param interfaces: iterable interface indexes
 385
 386         """
 387         for i in interfaces:
 388             i.enable_capture()
 389
 390     @classmethod
 391     def register_capture(cls, cap_name):
 392         """ Register a capture in the testclass """
 393         # add to the list of captures with current timestamp
 394         cls._captures.append((time.time(), cap_name))
 395         # filter out from zombies
 396         cls._zombie_captures = [(stamp, name)
 397                                 for (stamp, name) in cls._zombie_captures
 398                                 if name != cap_name]
 399
 400     @classmethod
 401     def pg_start(cls):
 402         """ Remove any zombie captures and enable the packet generator """
 403         # how long before capture is allowed to be deleted - otherwise vpp
 404         # crashes - 100ms seems enough (this shouldn't be needed at all)
 405         capture_ttl = 0.1
 406         now = time.time()
 407         for stamp, cap_name in cls._zombie_captures:
 408             wait = stamp + capture_ttl - now
 409             if wait > 0:
 410                 cls.sleep(wait, "before deleting capture %s" % cap_name)
 411                 now = time.time()
 412             cls.logger.debug("Removing zombie capture %s" % cap_name)
 413             cls.vapi.cli('packet-generator delete %s' % cap_name)
 414
 415         cls.vapi.cli("trace add pg-input 50")  # 50 is maximum
 416         cls.vapi.cli('packet-generator enable')
 417         cls._zombie_captures = cls._captures
 418         cls._captures = []
 419
 420     @classmethod
 421     def create_pg_interfaces(cls, interfaces):
 422         """
 423         Create packet-generator interfaces.
 424
 425         :param interfaces: iterable indexes of the interfaces.
 426         :returns: List of created interfaces.
 427
 428         """
 429         result = []
 430         for i in interfaces:
 431             intf = VppPGInterface(cls, i)
 432             setattr(cls, intf.name, intf)
 433             result.append(intf)
 434         cls.pg_interfaces = result
 435         return result
 436
 437     @classmethod
 438     def create_loopback_interfaces(cls, interfaces):
 439         """
 440         Create loopback interfaces.
 441
 442         :param interfaces: iterable indexes of the interfaces.
 443         :returns: List of created interfaces.
 444         """
 445         result = []
 446         for i in interfaces:
 447             intf = VppLoInterface(cls, i)
 448             setattr(cls, intf.name, intf)
 449             result.append(intf)
 450         cls.lo_interfaces = result
 451         return result
 452
 453     @staticmethod
 454     def extend_packet(packet, size):
 455         """
 456         Extend packet to given size by padding with spaces
 457         NOTE: Currently works only when Raw layer is present.
 458
 459         :param packet: packet
 460         :param size: target size
 461
 462         """
 463         packet_len = len(packet) + 4
 464         extend = size - packet_len
 465         if extend > 0:
 466             packet[Raw].load += ' ' * extend
 467
 468     @classmethod
 469     def reset_packet_infos(cls):
 470         """ Reset the list of packet info objects and packet counts to zero """
 471         cls._packet_infos = {}
 472         cls._packet_count_for_dst_if_idx = {}
 473
 474     @classmethod
 475     def create_packet_info(cls, src_if, dst_if):
 476         """
 477         Create packet info object containing the source and destination indexes
 478         and add it to the testcase's packet info list
 479
 480         :param VppInterface src_if: source interface
 481         :param VppInterface dst_if: destination interface
 482
 483         :returns: _PacketInfo object
 484
 485         """
 486         info = _PacketInfo()
 487         info.index = len(cls._packet_infos)
 488         info.src = src_if.sw_if_index
 489         info.dst = dst_if.sw_if_index
 490         if isinstance(dst_if, VppSubInterface):
 491             dst_idx = dst_if.parent.sw_if_index
 492         else:
 493             dst_idx = dst_if.sw_if_index
 494         if dst_idx in cls._packet_count_for_dst_if_idx:
 495             cls._packet_count_for_dst_if_idx[dst_idx] += 1
 496         else:
 497             cls._packet_count_for_dst_if_idx[dst_idx] = 1
 498         cls._packet_infos[info.index] = info
 499         return info
 500
 501     @staticmethod
 502     def info_to_payload(info):
 503         """
 504         Convert _PacketInfo object to packet payload
 505
 506         :param info: _PacketInfo object
 507
 508         :returns: string containing serialized data from packet info
 509         """
 510         return "%d %d %d" % (info.index, info.src, info.dst)
 511
 512     @staticmethod
 513     def payload_to_info(payload):
 514         """
 515         Convert packet payload to _PacketInfo object
 516
 517         :param payload: packet payload
 518
 519         :returns: _PacketInfo object containing de-serialized data from payload
 520
 521         """
 522         numbers = payload.split()
 523         info = _PacketInfo()
 524         info.index = int(numbers[0])
 525         info.src = int(numbers[1])
 526         info.dst = int(numbers[2])
 527         return info
 528
 529     def get_next_packet_info(self, info):
 530         """
 531         Iterate over the packet info list stored in the testcase
 532         Start iteration with first element if info is None
 533         Continue based on index in info if info is specified
 534
 535         :param info: info or None
 536         :returns: next info in list or None if no more infos
 537         """
 538         if info is None:
 539             next_index = 0
 540         else:
 541             next_index = info.index + 1
 542         if next_index == len(self._packet_infos):
 543             return None
 544         else:
 545             return self._packet_infos[next_index]
 546
 547     def get_next_packet_info_for_interface(self, src_index, info):
 548         """
 549         Search the packet info list for the next packet info with same source
 550         interface index
 551
 552         :param src_index: source interface index to search for
 553         :param info: packet info - where to start the search
 554         :returns: packet info or None
 555
 556         """
 557         while True:
 558             info = self.get_next_packet_info(info)
 559             if info is None:
 560                 return None
 561             if info.src == src_index:
 562                 return info
 563
 564     def get_next_packet_info_for_interface2(self, src_index, dst_index, info):
 565         """
 566         Search the packet info list for the next packet info with same source
 567         and destination interface indexes
 568
 569         :param src_index: source interface index to search for
 570         :param dst_index: destination interface index to search for
 571         :param info: packet info - where to start the search
 572         :returns: packet info or None
 573
 574         """
 575         while True:
 576             info = self.get_next_packet_info_for_interface(src_index, info)
 577             if info is None:
 578                 return None
 579             if info.dst == dst_index:
 580                 return info
 581
 582     def assert_equal(self, real_value, expected_value, name_or_class=None):
 583         if name_or_class is None:
 584             self.assertEqual(real_value, expected_value)
 585             return
 586         try:
 587             msg = "Invalid %s: %d('%s') does not match expected value %d('%s')"
 588             msg = msg % (getdoc(name_or_class).strip(),
 589                          real_value, str(name_or_class(real_value)),
 590                          expected_value, str(name_or_class(expected_value)))
 591         except:
 592             msg = "Invalid %s: %s does not match expected value %s" % (
 593                 name_or_class, real_value, expected_value)
 594
 595         self.assertEqual(real_value, expected_value, msg)
 596
 597     def assert_in_range(self,
 598                         real_value,
 599                         expected_min,
 600                         expected_max,
 601                         name=None):
 602         if name is None:
 603             msg = None
 604         else:
 605             msg = "Invalid %s: %s out of range <%s,%s>" % (
 606                 name, real_value, expected_min, expected_max)
 607         self.assertTrue(expected_min <= real_value <= expected_max, msg)
 608
 609     @classmethod
 610     def sleep(cls, timeout, remark=None):
 611         if hasattr(cls, 'logger'):
 612             cls.logger.debug("Sleeping for %ss (%s)" % (timeout, remark))
 613         time.sleep(timeout)
 614
 615
 616 class VppTestResult(unittest.TestResult):
 617     """
 618     @property result_string
 619      String variable to store the test case result string.
 620     @property errors
 621      List variable containing 2-tuples of TestCase instances and strings
 622      holding formatted tracebacks. Each tuple represents a test which
 623      raised an unexpected exception.
 624     @property failures
 625      List variable containing 2-tuples of TestCase instances and strings
 626      holding formatted tracebacks. Each tuple represents a test where
 627      a failure was explicitly signalled using the TestCase.assert*()
 628      methods.
 629     """
 630
 631     def __init__(self, stream, descriptions, verbosity):
 632         """
 633         :param stream File descriptor to store where to report test results.
 634             Set to the standard error stream by default.
 635         :param descriptions Boolean variable to store information if to use
 636             test case descriptions.
 637         :param verbosity Integer variable to store required verbosity level.
 638         """
 639         unittest.TestResult.__init__(self, stream, descriptions, verbosity)
 640         self.stream = stream
 641         self.descriptions = descriptions
 642         self.verbosity = verbosity
 643         self.result_string = None
 644
 645     def addSuccess(self, test):
 646         """
 647         Record a test succeeded result
 648
 649         :param test:
 650
 651         """
 652         if hasattr(test, 'logger'):
 653             test.logger.debug("--- addSuccess() %s.%s(%s) called"
 654                               % (test.__class__.__name__,
 655                                  test._testMethodName,
 656                                  test._testMethodDoc))
 657         unittest.TestResult.addSuccess(self, test)
 658         self.result_string = colorize("OK", GREEN)
 659
 660     def addSkip(self, test, reason):
 661         """
 662         Record a test skipped.
 663
 664         :param test:
 665         :param reason:
 666
 667         """
 668         if hasattr(test, 'logger'):
 669             test.logger.debug("--- addSkip() %s.%s(%s) called, reason is %s"
 670                               % (test.__class__.__name__,
 671                                  test._testMethodName,
 672                                  test._testMethodDoc,
 673                                  reason))
 674         unittest.TestResult.addSkip(self, test, reason)
 675         self.result_string = colorize("SKIP", YELLOW)
 676
 677     def addFailure(self, test, err):
 678         """
 679         Record a test failed result
 680
 681         :param test:
 682         :param err: error message
 683
 684         """
 685         if hasattr(test, 'logger'):
 686             test.logger.debug("--- addFailure() %s.%s(%s) called, err is %s"
 687                               % (test.__class__.__name__,
 688                                  test._testMethodName,
 689                                  test._testMethodDoc, err))
 690             test.logger.debug("formatted exception is:\n%s" %
 691                               "".join(format_exception(*err)))
 692         unittest.TestResult.addFailure(self, test, err)
 693         if hasattr(test, 'tempdir'):
 694             self.result_string = colorize("FAIL", RED) + \
 695                 ' [ temp dir used by test case: ' + test.tempdir + ' ]'
 696         else:
 697             self.result_string = colorize("FAIL", RED) + ' [no temp dir]'
 698
 699     def addError(self, test, err):
 700         """
 701         Record a test error result
 702
 703         :param test:
 704         :param err: error message
 705
 706         """
 707         if hasattr(test, 'logger'):
 708             test.logger.debug("--- addError() %s.%s(%s) called, err is %s"
 709                               % (test.__class__.__name__,
 710                                  test._testMethodName,
 711                                  test._testMethodDoc, err))
 712             test.logger.debug("formatted exception is:\n%s" %
 713                               "".join(format_exception(*err)))
 714         unittest.TestResult.addError(self, test, err)
 715         if hasattr(test, 'tempdir'):
 716             self.result_string = colorize("ERROR", RED) + \
 717                 ' [ temp dir used by test case: ' + test.tempdir + ' ]'
 718         else:
 719             self.result_string = colorize("ERROR", RED) + ' [no temp dir]'
 720
 721     def getDescription(self, test):
 722         """
 723         Get test description
 724
 725         :param test:
 726         :returns: test description
 727
 728         """
 729         # TODO: if none print warning not raise exception
 730         short_description = test.shortDescription()
 731         if self.descriptions and short_description:
 732             return short_description
 733         else:
 734             return str(test)
 735
 736     def startTest(self, test):
 737         """
 738         Start a test
 739
 740         :param test:
 741
 742         """
 743         unittest.TestResult.startTest(self, test)
 744         if self.verbosity > 0:
 745             self.stream.writeln(
 746                 "Starting " + self.getDescription(test) + " ...")
 747             self.stream.writeln(single_line_delim)
 748
 749     def stopTest(self, test):
 750         """
 751         Stop a test
 752
 753         :param test:
 754
 755         """
 756         unittest.TestResult.stopTest(self, test)
 757         if self.verbosity > 0:
 758             self.stream.writeln(single_line_delim)
 759             self.stream.writeln("%-73s%s" % (self.getDescription(test),
 760                                              self.result_string))
 761             self.stream.writeln(single_line_delim)
 762         else:
 763             self.stream.writeln("%-73s%s" % (self.getDescription(test),
 764                                              self.result_string))
 765
 766     def printErrors(self):
 767         """
 768         Print errors from running the test case
 769         """
 770         self.stream.writeln()
 771         self.printErrorList('ERROR', self.errors)
 772         self.printErrorList('FAIL', self.failures)
 773
 774     def printErrorList(self, flavour, errors):
 775         """
 776         Print error list to the output stream together with error type
 777         and test case description.
 778
 779         :param flavour: error type
 780         :param errors: iterable errors
 781
 782         """
 783         for test, err in errors:
 784             self.stream.writeln(double_line_delim)
 785             self.stream.writeln("%s: %s" %
 786                                 (flavour, self.getDescription(test)))
 787             self.stream.writeln(single_line_delim)
 788             self.stream.writeln("%s" % err)
 789
 790
 791 class VppTestRunner(unittest.TextTestRunner):
 792     """
 793     A basic test runner implementation which prints results to standard error.
 794     """
 795     @property
 796     def resultclass(self):
 797         """Class maintaining the results of the tests"""
 798         return VppTestResult
 799
 800     def __init__(self, stream=sys.stderr, descriptions=True, verbosity=1,
 801                  failfast=False, buffer=False, resultclass=None):
 802         # ignore stream setting here, use hard-coded stdout to be in sync
 803         # with prints from VppTestCase methods ...
 804         super(VppTestRunner, self).__init__(sys.stdout, descriptions,
 805                                             verbosity, failfast, buffer,
 806                                             resultclass)
 807
 808     test_option = "TEST"
 809
 810     def parse_test_option(self):
 811         try:
 812             f = os.getenv(self.test_option)
 813         except:
 814             f = None
 815         filter_file_name = None
 816         filter_class_name = None
 817         filter_func_name = None
 818         if f:
 819             if '.' in f:
 820                 parts = f.split('.')
 821                 if len(parts) > 3:
 822                     raise Exception("Unrecognized %s option: %s" %
 823                                     (self.test_option, f))
 824                 if len(parts) > 2:
 825                     if parts[2] not in ('*', ''):
 826                         filter_func_name = parts[2]
 827                 if parts[1] not in ('*', ''):
 828                     filter_class_name = parts[1]
 829                 if parts[0] not in ('*', ''):
 830                     if parts[0].startswith('test_'):
 831                         filter_file_name = parts[0]
 832                     else:
 833                         filter_file_name = 'test_%s' % parts[0]
 834             else:
 835                 if f.startswith('test_'):
 836                     filter_file_name = f
 837                 else:
 838                     filter_file_name = 'test_%s' % f
 839         return filter_file_name, filter_class_name, filter_func_name
 840
 841     def filter_tests(self, tests, filter_file, filter_class, filter_func):
 842         result = unittest.suite.TestSuite()
 843         for t in tests:
 844             if isinstance(t, unittest.suite.TestSuite):
 845                 # this is a bunch of tests, recursively filter...
 846                 x = self.filter_tests(t, filter_file, filter_class,
 847                                       filter_func)
 848                 if x.countTestCases() > 0:
 849                     result.addTest(x)
 850             elif isinstance(t, unittest.TestCase):
 851                 # this is a single test
 852                 parts = t.id().split('.')
 853                 # t.id() for common cases like this:
 854                 # test_classifier.TestClassifier.test_acl_ip
 855                 # apply filtering only if it is so
 856                 if len(parts) == 3:
 857                     if filter_file and filter_file != parts[0]:
 858                         continue
 859                     if filter_class and filter_class != parts[1]:
 860                         continue
 861                     if filter_func and filter_func != parts[2]:
 862                         continue
 863                 result.addTest(t)
 864             else:
 865                 # unexpected object, don't touch it
 866                 result.addTest(t)
 867         return result
 868
 869     def run(self, test):
 870         """
 871         Run the tests
 872
 873         :param test:
 874
 875         """
 876         gc.disable()  # disable garbage collection, we'll do that manually
 877         print("Running tests using custom test runner")  # debug message
 878         filter_file, filter_class, filter_func = self.parse_test_option()
 879         print("Active filters: file=%s, class=%s, function=%s" % (
 880             filter_file, filter_class, filter_func))
 881         filtered = self.filter_tests(test, filter_file, filter_class,
 882                                      filter_func)
 883         print("%s out of %s tests match specified filters" % (
 884             filtered.countTestCases(), test.countTestCases()))
 885         return super(VppTestRunner, self).run(filtered)