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