test/run_tests.py

   1 #!/usr/bin/env python
   2
   3 import sys
   4 import os
   5 import select
   6 import unittest
   7 import argparse
   8 import importlib
   9 from multiprocessing import Process, Pipe
  10 from framework import VppTestRunner
  11 from debug import spawn_gdb
  12 from log import global_logger
  13
  14
  15 def add_from_dir(suite, directory):
  16     do_insert = True
  17     for _f in os.listdir(directory):
  18         f = "%s/%s" % (directory, _f)
  19         if os.path.isdir(f):
  20             add_from_dir(suite, f)
  21             continue
  22         if not os.path.isfile(f):
  23             continue
  24         if do_insert:
  25             sys.path.insert(0, directory)
  26             do_insert = False
  27         if not _f.startswith("test_") or not _f.endswith(".py"):
  28             continue
  29         name = "".join(f.split("/")[-1].split(".")[:-1])
  30         if name in sys.modules:
  31             raise Exception("Duplicate test module `%s' found!" % name)
  32         module = importlib.import_module(name)
  33         for name, cls in module.__dict__.items():
  34             if not isinstance(cls, type):
  35                 continue
  36             if not issubclass(cls, unittest.TestCase):
  37                 continue
  38             if name == "VppTestCase":
  39                 continue
  40             for method in dir(cls):
  41                 if not callable(getattr(cls, method)):
  42                     continue
  43                 if method.startswith("test_"):
  44                     suite.addTest(cls(method))
  45
  46
  47 def test_runner_wrapper(suite, keep_alive_pipe, result_pipe):
  48     result = not VppTestRunner(
  49         pipe=keep_alive_pipe,
  50         verbosity=verbose,
  51         failfast=failfast).run(suite).wasSuccessful()
  52     result_pipe.send(result)
  53     result_pipe.close()
  54     keep_alive_pipe.close()
  55
  56
  57 def run_forked(suite):
  58     keep_alive_parent_end, keep_alive_child_end = Pipe(duplex=False)
  59     result_parent_end, result_child_end = Pipe(duplex=False)
  60
  61     child = Process(target=test_runner_wrapper,
  62                     args=(suite, keep_alive_child_end, result_child_end))
  63     child.start()
  64     last_test_temp_dir = None
  65     last_test_vpp_binary = None
  66     last_test = None
  67     result = None
  68     while result is None:
  69         readable = select.select([keep_alive_parent_end.fileno(),
  70                                   result_parent_end.fileno(),
  71                                   ],
  72                                  [], [], test_timeout)[0]
  73         if result_parent_end.fileno() in readable:
  74             result = result_parent_end.recv()
  75         elif keep_alive_parent_end.fileno() in readable:
  76             while keep_alive_parent_end.poll():
  77                 last_test, last_test_vpp_binary, last_test_temp_dir =\
  78                     keep_alive_parent_end.recv()
  79         else:
  80             global_logger.critical("Timeout while waiting for child test "
  81                                    "runner process (last test running was "
  82                                    "`%s' in `%s')!" %
  83                                    (last_test, last_test_temp_dir))
  84             if last_test_temp_dir and last_test_vpp_binary:
  85                 core_path = "%s/core" % last_test_temp_dir
  86                 if os.path.isfile(core_path):
  87                     global_logger.error("Core-file exists in test temporary "
  88                                         "directory: %s!" % core_path)
  89                     if d and d.lower() == "core":
  90                         spawn_gdb(last_test_vpp_binary, core_path,
  91                                   global_logger)
  92             child.terminate()
  93             result = -1
  94     keep_alive_parent_end.close()
  95     result_parent_end.close()
  96     return result
  97
  98
  99 if __name__ == '__main__':
 100
 101     try:
 102         verbose = int(os.getenv("V", 0))
 103     except:
 104         verbose = 0
 105
 106     default_test_timeout = 600  # 10 minutes
 107     try:
 108         test_timeout = int(os.getenv("TIMEOUT", default_test_timeout))
 109     except:
 110         test_timeout = default_test_timeout
 111
 112     try:
 113         debug = os.getenv("DEBUG")
 114     except:
 115         debug = None
 116
 117     parser = argparse.ArgumentParser(description="VPP unit tests")
 118     parser.add_argument("-f", "--failfast", action='count',
 119                         help="fast failure flag")
 120     parser.add_argument("-d", "--dir", action='append', type=str,
 121                         help="directory containing test files "
 122                              "(may be specified multiple times)")
 123     args = parser.parse_args()
 124     failfast = True if args.failfast == 1 else False
 125
 126     suite = unittest.TestSuite()
 127     for d in args.dir:
 128         global_logger.info("Adding tests from directory tree %s" % d)
 129         add_from_dir(suite, d)
 130
 131     if debug is None or debug.lower() not in ["gdb", "gdbserver"]:
 132         sys.exit(run_forked(suite))
 133
 134     # don't fork if debugging..
 135     sys.exit(not VppTestRunner(verbosity=verbose,
 136                                failfast=failfast).run(suite).wasSuccessful())