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
16 def test_runner_wrapper(suite, keep_alive_pipe, result_pipe):
17 result = not VppTestRunner(
20 failfast=failfast).run(suite).wasSuccessful()
21 result_pipe.send(result)
23 keep_alive_pipe.close()
26 class add_to_suite_callback:
27 def __init__(self, suite):
30 def __call__(self, file_name, cls, method):
31 suite.addTest(cls(method))
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)
38 child = Process(target=test_runner_wrapper,
39 args=(suite, keep_alive_child_end, result_child_end))
41 last_test_temp_dir = None
42 last_test_vpp_binary = None
46 readable = select.select([keep_alive_parent_end.fileno(),
47 result_parent_end.fileno(),
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()
57 global_logger.critical("Timeout while waiting for child test "
58 "runner process (last test running was "
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,
82 keep_alive_parent_end.close()
83 result_parent_end.close()
87 if __name__ == '__main__':
90 verbose = int(os.getenv("V", 0))
94 default_test_timeout = 600 # 10 minutes
96 test_timeout = int(os.getenv("TIMEOUT", default_test_timeout))
98 test_timeout = default_test_timeout
101 debug = os.getenv("DEBUG")
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
114 suite = unittest.TestSuite()
115 cb = add_to_suite_callback(suite)
117 global_logger.info("Adding tests from directory tree %s" % d)
118 discover_tests(d, cb)
120 if debug is None or debug.lower() not in ["gdb", "gdbserver"]:
121 sys.exit(run_forked(suite))
123 # don't fork if debugging..
124 sys.exit(not VppTestRunner(verbosity=verbose,
125 failfast=failfast).run(suite).wasSuccessful())