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
 356     def setUp(self):
 357         """ Clear trace before running each test"""
 358         self.logger.debug("--- setUp() for %s.%s(%s) called ---" %
 359                           (self.__class__.__name__, self._testMethodName,
 360                            self._testMethodDoc))
 361         if self.vpp_dead:
 362             raise Exception("VPP is dead when setting up the test")
 363         self.sleep(.1, "during setUp")
 364         self.vpp_stdout_deque.append(
 365             "--- test setUp() for %s.%s(%s) starts here ---\n" %
 366             (self.__class__.__name__, self._testMethodName,
 367              self._testMethodDoc))
 368         self.vpp_stderr_deque.append(
 369             "--- test setUp() for %s.%s(%s) starts here ---\n" %
 370             (self.__class__.__name__, self._testMethodName,
 371              self._testMethodDoc))
 372         self.vapi.cli("clear trace")
 373         # store the test instance inside the test class - so that objects
 374         # holding the class can access instance methods (like assertEqual)
 375         type(self).test_instance = self
 376
 377     @classmethod
 378     def pg_enable_capture(cls, interfaces):
 379         """
 380         Enable capture on packet-generator interfaces
 381
 382         :param interfaces: iterable interface indexes
 383
 384         """
 385         for i in interfaces:
 386             i.enable_capture()
 387
 388     @classmethod
 389     def register_capture(cls, cap_name):
 390         """ Register a capture in the testclass """
 391         # add to the list of captures with current timestamp
 392         cls._captures.append((time.time(), cap_name))
 393         # filter out from zombies
 394         cls._zombie_captures = [(stamp, name)
 395                                 for (stamp, name) in cls._zombie_captures
 396                                 if name != cap_name]
 397
 398     @classmethod
 399     def pg_start(cls):
 400         """ Remove any zombie captures and enable the packet generator """
 401         # how long before capture is allowed to be deleted - otherwise vpp
 402         # crashes - 100ms seems enough (this shouldn't be needed at all)
 403         capture_ttl = 0.1
 404         now = time.time()
 405         for stamp, cap_name in cls._zombie_captures:
 406             wait = stamp + capture_ttl - now
 407             if wait > 0:
 408                 cls.sleep(wait, "before deleting capture %s" % cap_name)
 409                 now = time.time()
 410             cls.logger.debug("Removing zombie capture %s" % cap_name)
 411             cls.vapi.cli('packet-generator delete %s' % cap_name)
 412
 413         cls.vapi.cli("trace add pg-input 50")  # 50 is maximum
 414         cls.vapi.cli('packet-generator enable')
 415         cls._zombie_captures = cls._captures
 416         cls._captures = []
 417
 418     @classmethod
 419     def create_pg_interfaces(cls, interfaces):
 420         """
 421         Create packet-generator interfaces.
 422
 423         :param interfaces: iterable indexes of the interfaces.
 424         :returns: List of created interfaces.
 425
 426         """
 427         result = []
 428         for i in interfaces:
 429             intf = VppPGInterface(cls, i)
 430             setattr(cls, intf.name, intf)
 431             result.append(intf)
 432         cls.pg_interfaces = result
 433         return result
 434
 435     @classmethod
 436     def create_loopback_interfaces(cls, interfaces):
 437         """
 438         Create loopback interfaces.
 439
 440         :param interfaces: iterable indexes of the interfaces.
 441         :returns: List of created interfaces.
 442         """
 443         result = []
 444         for i in interfaces:
 445             intf = VppLoInterface(cls, i)
 446             setattr(cls, intf.name, intf)
 447             result.append(intf)
 448         cls.lo_interfaces = result
 449         return result
 450
 451     @staticmethod
 452     def extend_packet(packet, size):
 453         """
 454         Extend packet to given size by padding with spaces
 455         NOTE: Currently works only when Raw layer is present.
 456
 457         :param packet: packet
 458         :param size: target size
 459
 460         """
 461         packet_len = len(packet) + 4
 462         extend = size - packet_len
 463         if extend > 0:
 464             packet[Raw].load += ' ' * extend
 465
 466     @classmethod
 467     def reset_packet_infos(cls):
 468         """ Reset the list of packet info objects and packet counts to zero """
 469         cls._packet_infos = {}
 470         cls._packet_count_for_dst_if_idx = {}
 471
 472     @classmethod
 473     def create_packet_info(cls, src_if, dst_if):
 474         """
 475         Create packet info object containing the source and destination indexes
 476         and add it to the testcase's packet info list
 477
 478         :param VppInterface src_if: source interface
 479         :param VppInterface dst_if: destination interface
 480
 481         :returns: _PacketInfo object
 482
 483         """
 484         info = _PacketInfo()
 485         info.index = len(cls._packet_infos)
 486         info.src = src_if.sw_if_index
 487         info.dst = dst_if.sw_if_index
 488         if isinstance(dst_if, VppSubInterface):
 489             dst_idx = dst_if.parent.sw_if_index
 490         else:
 491             dst_idx = dst_if.sw_if_index
 492         if dst_idx in cls._packet_count_for_dst_if_idx:
 493             cls._packet_count_for_dst_if_idx[dst_idx] += 1
 494         else:
 495             cls._packet_count_for_dst_if_idx[dst_idx] = 1
 496         cls._packet_infos[info.index] = info
 497         return info
 498
 499     @staticmethod
 500     def info_to_payload(info):
 501         """
 502         Convert _PacketInfo object to packet payload
 503
 504         :param info: _PacketInfo object
 505
 506         :returns: string containing serialized data from packet info
 507         """
 508         return "%d %d %d" % (info.index, info.src, info.dst)
 509
 510     @staticmethod
 511     def payload_to_info(payload):
 512         """
 513         Convert packet payload to _PacketInfo object
 514
 515         :param payload: packet payload
 516
 517         :returns: _PacketInfo object containing de-serialized data from payload
 518
 519         """
 520         numbers = payload.split()
 521         info = _PacketInfo()
 522         info.index = int(numbers[0])
 523         info.src = int(numbers[1])
 524         info.dst = int(numbers[2])
 525         return info
 526
 527     def get_next_packet_info(self, info):
 528         """
 529         Iterate over the packet info list stored in the testcase
 530         Start iteration with first element if info is None
 531         Continue based on index in info if info is specified
 532
 533         :param info: info or None
 534         :returns: next info in list or None if no more infos
 535         """
 536         if info is None:
 537             next_index = 0
 538         else:
 539             next_index = info.index + 1
 540         if next_index == len(self._packet_infos):
 541             return None
 542         else:
 543             return self._packet_infos[next_index]
 544
 545     def get_next_packet_info_for_interface(self, src_index, info):
 546         """
 547         Search the packet info list for the next packet info with same source
 548         interface index
 549
 550         :param src_index: source interface index to search for
 551         :param info: packet info - where to start the search
 552         :returns: packet info or None
 553
 554         """
 555         while True:
 556             info = self.get_next_packet_info(info)
 557             if info is None:
 558                 return None
 559             if info.src == src_index:
 560                 return info
 561
 562     def get_next_packet_info_for_interface2(self, src_index, dst_index, info):
 563         """
 564         Search the packet info list for the next packet info with same source
 565         and destination interface indexes
 566
 567         :param src_index: source interface index to search for
 568         :param dst_index: destination interface index to search for
 569         :param info: packet info - where to start the search
 570         :returns: packet info or None
 571
 572         """
 573         while True:
 574             info = self.get_next_packet_info_for_interface(src_index, info)
 575             if info is None:
 576                 return None
 577             if info.dst == dst_index:
 578                 return info
 579
 580     def assert_equal(self, real_value, expected_value, name_or_class=None):
 581         if name_or_class is None:
 582             self.assertEqual(real_value, expected_value)
 583             return
 584         try:
 585             msg = "Invalid %s: %d('%s') does not match expected value %d('%s')"
 586             msg = msg % (getdoc(name_or_class).strip(),
 587                          real_value, str(name_or_class(real_value)),
 588                          expected_value, str(name_or_class(expected_value)))
 589         except:
 590             msg = "Invalid %s: %s does not match expected value %s" % (
 591                 name_or_class, real_value, expected_value)
 592
 593         self.assertEqual(real_value, expected_value, msg)
 594
 595     def assert_in_range(self,
 596                         real_value,
 597                         expected_min,
 598                         expected_max,
 599                         name=None):
 600         if name is None:
 601             msg = None
 602         else:
 603             msg = "Invalid %s: %s out of range <%s,%s>" % (
 604                 name, real_value, expected_min, expected_max)
 605         self.assertTrue(expected_min <= real_value <= expected_max, msg)
 606
 607     @classmethod
 608     def sleep(cls, timeout, remark=None):
 609         if hasattr(cls, 'logger'):
 610             cls.logger.debug("Sleeping for %ss (%s)" % (timeout, remark))
 611         time.sleep(timeout)
 612
 613
 614 class VppTestResult(unittest.TestResult):
 615     """
 616     @property result_string
 617      String variable to store the test case result string.
 618     @property errors
 619      List variable containing 2-tuples of TestCase instances and strings
 620      holding formatted tracebacks. Each tuple represents a test which
 621      raised an unexpected exception.
 622     @property failures
 623      List variable containing 2-tuples of TestCase instances and strings
 624      holding formatted tracebacks. Each tuple represents a test where
 625      a failure was explicitly signalled using the TestCase.assert*()
 626      methods.
 627     """
 628
 629     def __init__(self, stream, descriptions, verbosity):
 630         """
 631         :param stream File descriptor to store where to report test results.
 632             Set to the standard error stream by default.
 633         :param descriptions Boolean variable to store information if to use
 634             test case descriptions.
 635         :param verbosity Integer variable to store required verbosity level.
 636         """
 637         unittest.TestResult.__init__(self, stream, descriptions, verbosity)
 638         self.stream = stream
 639         self.descriptions = descriptions
 640         self.verbosity = verbosity
 641         self.result_string = None
 642
 643     def addSuccess(self, test):
 644         """
 645         Record a test succeeded result
 646
 647         :param test:
 648
 649         """
 650         if hasattr(test, 'logger'):
 651             test.logger.debug("--- addSuccess() %s.%s(%s) called"
 652                               % (test.__class__.__name__,
 653                                  test._testMethodName,
 654                                  test._testMethodDoc))
 655         unittest.TestResult.addSuccess(self, test)
 656         self.result_string = colorize("OK", GREEN)
 657
 658     def addSkip(self, test, reason):
 659         """
 660         Record a test skipped.
 661
 662         :param test:
 663         :param reason:
 664
 665         """
 666         if hasattr(test, 'logger'):
 667             test.logger.debug("--- addSkip() %s.%s(%s) called, reason is %s"
 668                               % (test.__class__.__name__,
 669                                  test._testMethodName,
 670                                  test._testMethodDoc,
 671                                  reason))
 672         unittest.TestResult.addSkip(self, test, reason)
 673         self.result_string = colorize("SKIP", YELLOW)
 674
 675     def addFailure(self, test, err):
 676         """
 677         Record a test failed result
 678
 679         :param test:
 680         :param err: error message
 681
 682         """
 683         if hasattr(test, 'logger'):
 684             test.logger.debug("--- addFailure() %s.%s(%s) called, err is %s"
 685                               % (test.__class__.__name__,
 686                                  test._testMethodName,
 687                                  test._testMethodDoc, err))
 688             test.logger.debug("formatted exception is:\n%s" %
 689                               "".join(format_exception(*err)))
 690         unittest.TestResult.addFailure(self, test, err)
 691         if hasattr(test, 'tempdir'):
 692             self.result_string = colorize("FAIL", RED) + \
 693                 ' [ temp dir used by test case: ' + test.tempdir + ' ]'
 694         else:
 695             self.result_string = colorize("FAIL", RED) + ' [no temp dir]'
 696
 697     def addError(self, test, err):
 698         """
 699         Record a test error result
 700
 701         :param test:
 702         :param err: error message
 703
 704         """
 705         if hasattr(test, 'logger'):
 706             test.logger.debug("--- addError() %s.%s(%s) called, err is %s"
 707                               % (test.__class__.__name__,
 708                                  test._testMethodName,
 709                                  test._testMethodDoc, err))
 710             test.logger.debug("formatted exception is:\n%s" %
 711                               "".join(format_exception(*err)))
 712         unittest.TestResult.addError(self, test, err)
 713         if hasattr(test, 'tempdir'):
 714             self.result_string = colorize("ERROR", RED) + \
 715                 ' [ temp dir used by test case: ' + test.tempdir + ' ]'
 716         else:
 717             self.result_string = colorize("ERROR", RED) + ' [no temp dir]'
 718
 719     def getDescription(self, test):
 720         """
 721         Get test description
 722
 723         :param test:
 724         :returns: test description
 725
 726         """
 727         # TODO: if none print warning not raise exception
 728         short_description = test.shortDescription()
 729         if self.descriptions and short_description:
 730             return short_description
 731         else:
 732             return str(test)
 733
 734     def startTest(self, test):
 735         """
 736         Start a test
 737
 738         :param test:
 739
 740         """
 741         unittest.TestResult.startTest(self, test)
 742         if self.verbosity > 0:
 743             self.stream.writeln(
 744                 "Starting " + self.getDescription(test) + " ...")
 745             self.stream.writeln(single_line_delim)
 746
 747     def stopTest(self, test):
 748         """
 749         Stop a test
 750
 751         :param test:
 752
 753         """
 754         unittest.TestResult.stopTest(self, test)
 755         if self.verbosity > 0:
 756             self.stream.writeln(single_line_delim)
 757             self.stream.writeln("%-73s%s" % (self.getDescription(test),
 758                                              self.result_string))
 759             self.stream.writeln(single_line_delim)
 760         else:
 761             self.stream.writeln("%-73s%s" % (self.getDescription(test),
 762                                              self.result_string))
 763
 764     def printErrors(self):
 765         """
 766         Print errors from running the test case
 767         """
 768         self.stream.writeln()
 769         self.printErrorList('ERROR', self.errors)
 770         self.printErrorList('FAIL', self.failures)
 771
 772     def printErrorList(self, flavour, errors):
 773         """
 774         Print error list to the output stream together with error type
 775         and test case description.
 776
 777         :param flavour: error type
 778         :param errors: iterable errors
 779
 780         """
 781         for test, err in errors:
 782             self.stream.writeln(double_line_delim)
 783             self.stream.writeln("%s: %s" %
 784                                 (flavour, self.getDescription(test)))
 785             self.stream.writeln(single_line_delim)
 786             self.stream.writeln("%s" % err)
 787
 788
 789 class VppTestRunner(unittest.TextTestRunner):
 790     """
 791     A basic test runner implementation which prints results to standard error.
 792     """
 793     @property
 794     def resultclass(self):
 795         """Class maintaining the results of the tests"""
 796         return VppTestResult
 797
 798     def __init__(self, stream=sys.stderr, descriptions=True, verbosity=1,
 799                  failfast=False, buffer=False, resultclass=None):
 800         # ignore stream setting here, use hard-coded stdout to be in sync
 801         # with prints from VppTestCase methods ...
 802         super(VppTestRunner, self).__init__(sys.stdout, descriptions,
 803                                             verbosity, failfast, buffer,
 804                                             resultclass)
 805
 806     test_option = "TEST"
 807
 808     def parse_test_option(self):
 809         try:
 810             f = os.getenv(self.test_option)
 811         except:
 812             f = None
 813         filter_file_name = None
 814         filter_class_name = None
 815         filter_func_name = None
 816         if f:
 817             if '.' in f:
 818                 parts = f.split('.')
 819                 if len(parts) > 3:
 820                     raise Exception("Unrecognized %s option: %s" %
 821                                     (self.test_option, f))
 822                 if len(parts) > 2:
 823                     if parts[2] not in ('*', ''):
 824                         filter_func_name = parts[2]
 825                 if parts[1] not in ('*', ''):
 826                     filter_class_name = parts[1]
 827                 if parts[0] not in ('*', ''):
 828                     if parts[0].startswith('test_'):
 829                         filter_file_name = parts[0]
 830                     else:
 831                         filter_file_name = 'test_%s' % parts[0]
 832             else:
 833                 if f.startswith('test_'):
 834                     filter_file_name = f
 835                 else:
 836                     filter_file_name = 'test_%s' % f
 837         return filter_file_name, filter_class_name, filter_func_name
 838
 839     def filter_tests(self, tests, filter_file, filter_class, filter_func):
 840         result = unittest.suite.TestSuite()
 841         for t in tests:
 842             if isinstance(t, unittest.suite.TestSuite):
 843                 # this is a bunch of tests, recursively filter...
 844                 x = self.filter_tests(t, filter_file, filter_class,
 845                                       filter_func)
 846                 if x.countTestCases() > 0:
 847                     result.addTest(x)
 848             elif isinstance(t, unittest.TestCase):
 849                 # this is a single test
 850                 parts = t.id().split('.')
 851                 # t.id() for common cases like this:
 852                 # test_classifier.TestClassifier.test_acl_ip
 853                 # apply filtering only if it is so
 854                 if len(parts) == 3:
 855                     if filter_file and filter_file != parts[0]:
 856                         continue
 857                     if filter_class and filter_class != parts[1]:
 858                         continue
 859                     if filter_func and filter_func != parts[2]:
 860                         continue
 861                 result.addTest(t)
 862             else:
 863                 # unexpected object, don't touch it
 864                 result.addTest(t)
 865         return result
 866
 867     def run(self, test):
 868         """
 869         Run the tests
 870
 871         :param test:
 872
 873         """
 874         gc.disable()  # disable garbage collection, we'll do that manually
 875         print("Running tests using custom test runner")  # debug message
 876         filter_file, filter_class, filter_func = self.parse_test_option()
 877         print("Active filters: file=%s, class=%s, function=%s" % (
 878             filter_file, filter_class, filter_func))
 879         filtered = self.filter_tests(test, filter_file, filter_class,
 880                                      filter_func)
 881         print("%s out of %s tests match specified filters" % (
 882             filtered.countTestCases(), test.countTestCases()))
 883         return super(VppTestRunner, self).run(filtered)