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(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 handle_core(vpp_binary, core_path):
  58     try:
  59         d = os.getenv("DEBUG")
  60     except:
  61         d = None
  62     if d and d.lower() == "core":
  63         spawn_gdb(vpp_binary, core_path, global_logger)
  64
  65
  66 if __name__ == '__main__':
  67     try:
  68         verbose = int(os.getenv("V", 0))
  69     except:
  70         verbose = 0
  71
  72     default_test_timeout = 600  # 10 minutes
  73     try:
  74         test_timeout = int(os.getenv("TIMEOUT", default_test_timeout))
  75     except:
  76         test_timeout = default_test_timeout
  77
  78     parser = argparse.ArgumentParser(description="VPP unit tests")
  79     parser.add_argument("-f", "--failfast", action='count',
  80                         help="fast failure flag")
  81     parser.add_argument("-d", "--dir", action='append', type=str,
  82                         help="directory containing test files "
  83                              "(may be specified multiple times)")
  84     args = parser.parse_args()
  85     failfast = True if args.failfast == 1 else False
  86
  87     suite = unittest.TestSuite()
  88     for d in args.dir:
  89         global_logger.info("Adding tests from directory tree %s" % d)
  90         add_from_dir(suite, d)
  91     keep_alive_parent_end, keep_alive_child_end = Pipe(duplex=False)
  92     result_parent_end, result_child_end = Pipe(duplex=False)
  93
  94     p = Process(target=test_runner_wrapper,
  95                 args=(keep_alive_child_end,
  96                       result_child_end))
  97     p.start()
  98     last_test_temp_dir = None
  99     last_test_vpp_binary = None
 100     last_test = None
 101     result = None
 102     while result is None:
 103         readable = select.select([keep_alive_parent_end.fileno(),
 104                                   result_parent_end.fileno(),
 105                                   ],
 106                                  [], [], test_timeout)[0]
 107         if result_parent_end.fileno() in readable:
 108             result = result_parent_end.recv()
 109         elif keep_alive_parent_end.fileno() in readable:
 110             while keep_alive_parent_end.poll():
 111                 last_test, last_test_vpp_binary, last_test_temp_dir =\
 112                     keep_alive_parent_end.recv()
 113         else:
 114             global_logger.critical("Timeout while waiting for child test "
 115                                    "runner process (last test running was "
 116                                    "`%s' in `%s')!" %
 117                                    (last_test, last_test_temp_dir))
 118             if last_test_temp_dir and last_test_vpp_binary:
 119                 core_path = "%s/core" % last_test_temp_dir
 120                 if os.path.isfile(core_path):
 121                     global_logger.error("Core-file exists in test temporary "
 122                                         "directory: %s!" % core_path)
 123                     handle_core(last_test_vpp_binary, core_path)
 124             p.terminate()
 125             result = -1
 126     keep_alive_parent_end.close()
 127     result_parent_end.close()
 128     sys.exit(result)