test/framework.py

   1 #!/usr/bin/env python
   2
   3 from abc import *
   4 import os
   5 import sys
   6 import subprocess
   7 import unittest
   8 import tempfile
   9 import resource
  10 from time import sleep
  11 from inspect import getdoc
  12 from hook import PollHook
  13 from vpp_pg_interface import VppPGInterface
  14 from vpp_papi_provider import VppPapiProvider
  15
  16 from scapy.packet import Raw
  17
  18 from logging import *
  19
  20 """
  21   Test framework module.
  22
  23   The module provides a set of tools for constructing and running tests and
  24   representing the results.
  25 """
  26
  27 handler = StreamHandler(sys.stdout)
  28 getLogger().addHandler(handler)
  29 try:
  30     verbose = int(os.getenv("V", 0))
  31 except:
  32     verbose = 0
  33 # 40 = ERROR, 30 = WARNING, 20 = INFO, 10 = DEBUG, 0 = NOTSET (all messages)
  34 getLogger().setLevel(40 - 10 * verbose)
  35 getLogger("scapy.runtime").addHandler(handler)
  36 getLogger("scapy.runtime").setLevel(ERROR)
  37
  38 # Static variables to store color formatting strings.
  39 #
  40 # These variables (RED, GREEN, YELLOW and LPURPLE) are used to configure
  41 # the color of the text to be printed in the terminal. Variable COLOR_RESET
  42 # is used to revert the text color to the default one.
  43 if hasattr(sys.stdout, 'isatty') and sys.stdout.isatty():
  44     RED = '\033[91m'
  45     GREEN = '\033[92m'
  46     YELLOW = '\033[93m'
  47     LPURPLE = '\033[94m'
  48     COLOR_RESET = '\033[0m'
  49 else:
  50     RED = ''
  51     GREEN = ''
  52     YELLOW = ''
  53     LPURPLE = ''
  54     COLOR_RESET = ''
  55
  56
  57 """ @var formatting delimiter consisting of '=' characters """
  58 double_line_delim = '=' * 70
  59 """ @var formatting delimiter consisting of '-' characters """
  60 single_line_delim = '-' * 70
  61
  62
  63 class _PacketInfo(object):
  64     """Private class to create packet info object.
  65
  66     Help process information about the next packet.
  67     Set variables to default values.
  68     @property index
  69       Integer variable to store the index of the packet.
  70     @property src
  71       Integer variable to store the index of the source packet generator
  72       interface of the packet.
  73     @property dst
  74       Integer variable to store the index of the destination packet generator
  75       interface of the packet.
  76     @property data
  77       Object variable to store the copy of the former packet.
  78
  79
  80     """
  81     index = -1
  82     src = -1
  83     dst = -1
  84     data = None
  85
  86
  87 class VppTestCase(unittest.TestCase):
  88     """
  89     Subclass of the python unittest.TestCase class.
  90
  91     This subclass is a base class for test cases that are implemented as classes
  92     It provides methods to create and run test case.
  93
  94     """
  95
  96     @property
  97     def packet_infos(self):
  98         """List of packet infos"""
  99         return self._packet_infos
 100
 101     @packet_infos.setter
 102     def packet_infos(self, value):
 103         self._packet_infos = value
 104
 105     @classmethod
 106     def instance(cls):
 107         """Return the instance of this testcase"""
 108         return cls.test_instance
 109
 110     @classmethod
 111     def setUpConstants(cls):
 112         """ Set-up the test case class based on environment variables """
 113         try:
 114             cls.interactive = True if int(os.getenv("I")) > 0 else False
 115         except:
 116             cls.interactive = False
 117         if cls.interactive and resource.getrlimit(resource.RLIMIT_CORE)[0] <= 0:
 118             # give a heads up if this is actually useless
 119             critical("WARNING: core size limit is set 0, core files will NOT "
 120                      "be created")
 121         cls.vpp_bin = os.getenv('VPP_TEST_BIN', "vpp")
 122         cls.plugin_path = os.getenv('VPP_TEST_PLUGIN_PATH')
 123         cls.vpp_cmdline = [cls.vpp_bin, "unix", "nodaemon",
 124                            "api-segment", "{", "prefix", cls.shm_prefix, "}"]
 125         if cls.plugin_path is not None:
 126             cls.vpp_cmdline.extend(["plugin_path", cls.plugin_path])
 127         info("vpp_cmdline: %s" % cls.vpp_cmdline)
 128
 129     @classmethod
 130     def setUpClass(cls):
 131         """
 132         Perform class setup before running the testcase
 133         Remove shared memory files, start vpp and connect the vpp-api
 134         """
 135         cls.tempdir = tempfile.mkdtemp(
 136             prefix='vpp-unittest-' + cls.__name__ + '-')
 137         cls.shm_prefix = cls.tempdir.split("/")[-1]
 138         os.chdir(cls.tempdir)
 139         info("Temporary dir is %s, shm prefix is %s",
 140              cls.tempdir, cls.shm_prefix)
 141         cls.setUpConstants()
 142         cls.pg_streams = []
 143         cls.packet_infos = {}
 144         cls.verbose = 0
 145         print(double_line_delim)
 146         print(YELLOW + getdoc(cls) + COLOR_RESET)
 147         print(double_line_delim)
 148         # need to catch exceptions here because if we raise, then the cleanup
 149         # doesn't get called and we might end with a zombie vpp
 150         try:
 151             cls.vpp = subprocess.Popen(cls.vpp_cmdline, stderr=subprocess.PIPE)
 152             debug("Spawned VPP with PID: %d" % cls.vpp.pid)
 153             cls.vpp_dead = False
 154             cls.vapi = VppPapiProvider(cls.shm_prefix, cls.shm_prefix)
 155             cls.vapi.register_hook(PollHook(cls))
 156             cls.vapi.connect()
 157         except:
 158             cls.vpp.terminate()
 159             del cls.vpp
 160
 161     @classmethod
 162     def quit(cls):
 163         """
 164         Disconnect vpp-api, kill vpp and cleanup shared memory files
 165         """
 166         if hasattr(cls, 'vpp'):
 167             cls.vapi.disconnect()
 168             cls.vpp.poll()
 169             if cls.vpp.returncode is None:
 170                 cls.vpp.terminate()
 171             del cls.vpp
 172
 173     @classmethod
 174     def tearDownClass(cls):
 175         """ Perform final cleanup after running all tests in this test-case """
 176         cls.quit()
 177
 178     def tearDown(self):
 179         """ Show various debug prints after each test """
 180         if not self.vpp_dead:
 181             info(self.vapi.cli("show int"))
 182             info(self.vapi.cli("show trace"))
 183             info(self.vapi.cli("show hardware"))
 184             info(self.vapi.cli("show error"))
 185             info(self.vapi.cli("show run"))
 186
 187     def setUp(self):
 188         """ Clear trace before running each test"""
 189         self.vapi.cli("clear trace")
 190         # store the test instance inside the test class - so that objects
 191         # holding the class can access instance methods (like assertEqual)
 192         type(self).test_instance = self
 193
 194     @classmethod
 195     def pg_enable_capture(cls, interfaces):
 196         """
 197         Enable capture on packet-generator interfaces
 198
 199         :param interfaces: iterable interface indexes
 200
 201         """
 202         for i in interfaces:
 203             i.enable_capture()
 204
 205     @classmethod
 206     def pg_start(cls):
 207         """
 208         Enable the packet-generator and send all prepared packet streams
 209         Remove the packet streams afterwards
 210         """
 211         cls.vapi.cli("trace add pg-input 50")  # 50 is maximum
 212         cls.vapi.cli('packet-generator enable')
 213         sleep(1)  # give VPP some time to process the packets
 214         for stream in cls.pg_streams:
 215             cls.vapi.cli('packet-generator delete %s' % stream)
 216         cls.pg_streams = []
 217
 218     @classmethod
 219     def create_pg_interfaces(cls, interfaces):
 220         """
 221         Create packet-generator interfaces
 222
 223         :param interfaces: iterable indexes of the interfaces
 224
 225         """
 226         result = []
 227         for i in interfaces:
 228             intf = VppPGInterface(cls, i)
 229             setattr(cls, intf.name, intf)
 230             result.append(intf)
 231         cls.pg_interfaces = result
 232         return result
 233
 234     @staticmethod
 235     def extend_packet(packet, size):
 236         """
 237         Extend packet to given size by padding with spaces
 238         NOTE: Currently works only when Raw layer is present.
 239
 240         :param packet: packet
 241         :param size: target size
 242
 243         """
 244         packet_len = len(packet) + 4
 245         extend = size - packet_len
 246         if extend > 0:
 247             packet[Raw].load += ' ' * extend
 248
 249     def add_packet_info_to_list(self, info):
 250         """
 251         Add packet info to the testcase's packet info list
 252
 253         :param info: packet info
 254
 255         """
 256         info.index = len(self.packet_infos)
 257         self.packet_infos[info.index] = info
 258
 259     def create_packet_info(self, src_pg_index, dst_pg_index):
 260         """
 261         Create packet info object containing the source and destination indexes
 262         and add it to the testcase's packet info list
 263
 264         :param src_pg_index: source packet-generator index
 265         :param dst_pg_index: destination packet-generator index
 266
 267         :returns: _PacketInfo object
 268
 269         """
 270         info = _PacketInfo()
 271         self.add_packet_info_to_list(info)
 272         info.src = src_pg_index
 273         info.dst = dst_pg_index
 274         return info
 275
 276     @staticmethod
 277     def info_to_payload(info):
 278         """
 279         Convert _PacketInfo object to packet payload
 280
 281         :param info: _PacketInfo object
 282
 283         :returns: string containing serialized data from packet info
 284         """
 285         return "%d %d %d" % (info.index, info.src, info.dst)
 286
 287     @staticmethod
 288     def payload_to_info(payload):
 289         """
 290         Convert packet payload to _PacketInfo object
 291
 292         :param payload: packet payload
 293
 294         :returns: _PacketInfo object containing de-serialized data from payload
 295
 296         """
 297         numbers = payload.split()
 298         info = _PacketInfo()
 299         info.index = int(numbers[0])
 300         info.src = int(numbers[1])
 301         info.dst = int(numbers[2])
 302         return info
 303
 304     def get_next_packet_info(self, info):
 305         """
 306         Iterate over the packet info list stored in the testcase
 307         Start iteration with first element if info is None
 308         Continue based on index in info if info is specified
 309
 310         :param info: info or None
 311         :returns: next info in list or None if no more infos
 312         """
 313         if info is None:
 314             next_index = 0
 315         else:
 316             next_index = info.index + 1
 317         if next_index == len(self.packet_infos):
 318             return None
 319         else:
 320             return self.packet_infos[next_index]
 321
 322     def get_next_packet_info_for_interface(self, src_index, info):
 323         """
 324         Search the packet info list for the next packet info with same source
 325         interface index
 326
 327         :param src_index: source interface index to search for
 328         :param info: packet info - where to start the search
 329         :returns: packet info or None
 330
 331         """
 332         while True:
 333             info = self.get_next_packet_info(info)
 334             if info is None:
 335                 return None
 336             if info.src == src_index:
 337                 return info
 338
 339     def get_next_packet_info_for_interface2(self, src_index, dst_index, info):
 340         """
 341         Search the packet info list for the next packet info with same source
 342         and destination interface indexes
 343
 344         :param src_index: source interface index to search for
 345         :param dst_index: destination interface index to search for
 346         :param info: packet info - where to start the search
 347         :returns: packet info or None
 348
 349         """
 350         while True:
 351             info = self.get_next_packet_info_for_interface(src_index, info)
 352             if info is None:
 353                 return None
 354             if info.dst == dst_index:
 355                 return info
 356
 357
 358 class VppTestResult(unittest.TestResult):
 359     """
 360     @property result_string
 361      String variable to store the test case result string.
 362     @property errors
 363      List variable containing 2-tuples of TestCase instances and strings
 364      holding formatted tracebacks. Each tuple represents a test which
 365      raised an unexpected exception.
 366     @property failures
 367      List variable containing 2-tuples of TestCase instances and strings
 368      holding formatted tracebacks. Each tuple represents a test where
 369      a failure was explicitly signalled using the TestCase.assert*()
 370      methods.
 371     """
 372
 373     def __init__(self, stream, descriptions, verbosity):
 374         """
 375         :param stream File descriptor to store where to report test results. Set
 376             to the standard error stream by default.
 377         :param descriptions Boolean variable to store information if to use test
 378             case descriptions.
 379         :param verbosity Integer variable to store required verbosity level.
 380         """
 381         unittest.TestResult.__init__(self, stream, descriptions, verbosity)
 382         self.stream = stream
 383         self.descriptions = descriptions
 384         self.verbosity = verbosity
 385         self.result_string = None
 386
 387     def addSuccess(self, test):
 388         """
 389         Record a test succeeded result
 390
 391         :param test:
 392
 393         """
 394         unittest.TestResult.addSuccess(self, test)
 395         self.result_string = GREEN + "OK" + COLOR_RESET
 396
 397     def addSkip(self, test, reason):
 398         """
 399         Record a test skipped.
 400
 401         :param test:
 402         :param reason:
 403
 404         """
 405         unittest.TestResult.addSkip(self, test, reason)
 406         self.result_string = YELLOW + "SKIP" + COLOR_RESET
 407
 408     def addFailure(self, test, err):
 409         """
 410         Record a test failed result
 411
 412         :param test:
 413         :param err: error message
 414
 415         """
 416         unittest.TestResult.addFailure(self, test, err)
 417         if hasattr(test, 'tempdir'):
 418             self.result_string = RED + "FAIL" + COLOR_RESET + \
 419                 ' [ temp dir used by test case: ' + test.tempdir + ' ]'
 420         else:
 421             self.result_string = RED + "FAIL" + COLOR_RESET + ' [no temp dir]'
 422
 423     def addError(self, test, err):
 424         """
 425         Record a test error result
 426
 427         :param test:
 428         :param err: error message
 429
 430         """
 431         unittest.TestResult.addError(self, test, err)
 432         if hasattr(test, 'tempdir'):
 433             self.result_string = RED + "ERROR" + COLOR_RESET + \
 434                 ' [ temp dir used by test case: ' + test.tempdir + ' ]'
 435         else:
 436             self.result_string = RED + "ERROR" + COLOR_RESET + ' [no temp dir]'
 437
 438     def getDescription(self, test):
 439         """
 440         Get test description
 441
 442         :param test:
 443         :returns: test description
 444
 445         """
 446         # TODO: if none print warning not raise exception
 447         short_description = test.shortDescription()
 448         if self.descriptions and short_description:
 449             return short_description
 450         else:
 451             return str(test)
 452
 453     def startTest(self, test):
 454         """
 455         Start a test
 456
 457         :param test:
 458
 459         """
 460         unittest.TestResult.startTest(self, test)
 461         if self.verbosity > 0:
 462             self.stream.writeln(
 463                 "Starting " + self.getDescription(test) + " ...")
 464             self.stream.writeln(single_line_delim)
 465
 466     def stopTest(self, test):
 467         """
 468         Stop a test
 469
 470         :param test:
 471
 472         """
 473         unittest.TestResult.stopTest(self, test)
 474         if self.verbosity > 0:
 475             self.stream.writeln(single_line_delim)
 476             self.stream.writeln("%-60s%s" %
 477                                 (self.getDescription(test), self.result_string))
 478             self.stream.writeln(single_line_delim)
 479         else:
 480             self.stream.writeln("%-60s%s" %
 481                                 (self.getDescription(test), self.result_string))
 482
 483     def printErrors(self):
 484         """
 485         Print errors from running the test case
 486         """
 487         self.stream.writeln()
 488         self.printErrorList('ERROR', self.errors)
 489         self.printErrorList('FAIL', self.failures)
 490
 491     def printErrorList(self, flavour, errors):
 492         """
 493         Print error list to the output stream together with error type
 494         and test case description.
 495
 496         :param flavour: error type
 497         :param errors: iterable errors
 498
 499         """
 500         for test, err in errors:
 501             self.stream.writeln(double_line_delim)
 502             self.stream.writeln("%s: %s" %
 503                                 (flavour, self.getDescription(test)))
 504             self.stream.writeln(single_line_delim)
 505             self.stream.writeln("%s" % err)
 506
 507
 508 class VppTestRunner(unittest.TextTestRunner):
 509     """
 510     A basic test runner implementation which prints results on standard error.
 511     """
 512     @property
 513     def resultclass(self):
 514         """Class maintaining the results of the tests"""
 515         return VppTestResult
 516
 517     def run(self, test):
 518         """
 519         Run the tests
 520
 521         :param test:
 522
 523         """
 524         print("Running tests using custom test runner")  # debug message
 525         return super(VppTestRunner, self).run(test)