test/framework.py

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