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
15 def test_runner_wrapper(suite, keep_alive_pipe, result_pipe):
16 result = not VppTestRunner(
19 failfast=failfast).run(suite).wasSuccessful()
20 result_pipe.send(result)
22 keep_alive_pipe.close()
25 class add_to_suite_callback:
26 def __init__(self, suite):
29 def __call__(self, file_name, cls, method):
30 suite.addTest(cls(method))
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)
37 child = Process(target=test_runner_wrapper,
38 args=(suite, keep_alive_child_end, result_child_end))
40 last_test_temp_dir = None
41 last_test_vpp_binary = None
45 readable = select.select([keep_alive_parent_end.fileno(),
46 result_parent_end.fileno(),
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()
56 global_logger.critical("Timeout while waiting for child test "
57 "runner process (last test running was "
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,
70 keep_alive_parent_end.close()
71 result_parent_end.close()
75 if __name__ == '__main__':
78 verbose = int(os.getenv("V", 0))
82 default_test_timeout = 600 # 10 minutes
84 test_timeout = int(os.getenv("TIMEOUT", default_test_timeout))
86 test_timeout = default_test_timeout
89 debug = os.getenv("DEBUG")
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
102 suite = unittest.TestSuite()
103 cb = add_to_suite_callback(suite)
105 global_logger.info("Adding tests from directory tree %s" % d)
106 discover_tests(d, cb)
108 if debug is None or debug.lower() not in ["gdb", "gdbserver"]:
109 sys.exit(run_forked(suite))
111 # don't fork if debugging..
112 sys.exit(not VppTestRunner(verbosity=verbose,
113 failfast=failfast).run(suite).wasSuccessful())