test/run_tests.py

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