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