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