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, failed_pipe):
17 result = not VppTestRunner(
18 keep_alive_pipe=keep_alive_pipe,
19 failed_pipe=failed_pipe,
21 failfast=failfast).run(suite).wasSuccessful()
22 result_pipe.send(result)
24 keep_alive_pipe.close()
28 class add_to_suite_callback:
29 def __init__(self, suite):
32 def __call__(self, file_name, cls, method):
33 suite.addTest(cls(method))
36 class Filter_by_class_list:
37 def __init__(self, class_list):
38 self.class_list = class_list
40 def __call__(self, file_name, class_name, func_name):
41 return class_name in self.class_list
44 def suite_from_failed(suite, failed):
45 filter_cb = Filter_by_class_list(failed)
46 return VppTestRunner.filter_tests(suite, filter_cb)
49 def run_forked(suite):
50 keep_alive_parent_end, keep_alive_child_end = Pipe(duplex=False)
51 result_parent_end, result_child_end = Pipe(duplex=False)
52 failed_parent_end, failed_child_end = Pipe(duplex=False)
54 child = Process(target=test_runner_wrapper,
55 args=(suite, keep_alive_child_end, result_child_end,
58 last_test_temp_dir = None
59 last_test_vpp_binary = None
64 readable = select.select([keep_alive_parent_end.fileno(),
65 result_parent_end.fileno(),
66 failed_parent_end.fileno(),
68 [], [], test_timeout)[0]
70 if result_parent_end.fileno() in readable:
71 result = result_parent_end.recv()
73 if keep_alive_parent_end.fileno() in readable:
74 while keep_alive_parent_end.poll():
75 last_test, last_test_vpp_binary,\
76 last_test_temp_dir, vpp_pid = keep_alive_parent_end.recv()
78 if failed_parent_end.fileno() in readable:
79 while failed_parent_end.poll():
80 failed_test = failed_parent_end.recv()
81 failed.add(failed_test.__name__)
84 global_logger.critical("Timeout while waiting for child test "
85 "runner process (last test running was "
87 (last_test, last_test_temp_dir))
88 failed_dir = os.getenv('VPP_TEST_FAILED_DIR')
89 lttd = last_test_temp_dir.split("/")[-1]
90 link_path = '%s%s-FAILED' % (failed_dir, lttd)
91 global_logger.error("Creating a link to the failed " +
92 "test: %s -> %s" % (link_path, lttd))
93 os.symlink(last_test_temp_dir, link_path)
94 api_post_mortem_path = "/tmp/api_post_mortem.%d" % vpp_pid
95 if os.path.isfile(api_post_mortem_path):
96 global_logger.error("Copying api_post_mortem.%d to %s" %
97 (vpp_pid, last_test_temp_dir))
98 shutil.copy2(api_post_mortem_path, last_test_temp_dir)
99 if last_test_temp_dir and last_test_vpp_binary:
100 core_path = "%s/core" % last_test_temp_dir
101 if os.path.isfile(core_path):
102 global_logger.error("Core-file exists in test temporary "
103 "directory: %s!" % core_path)
104 if d and d.lower() == "core":
105 spawn_gdb(last_test_vpp_binary, core_path,
109 keep_alive_parent_end.close()
110 result_parent_end.close()
111 failed_parent_end.close()
112 return result, failed
115 if __name__ == '__main__':
118 verbose = int(os.getenv("V", 0))
122 default_test_timeout = 600 # 10 minutes
124 test_timeout = int(os.getenv("TIMEOUT", default_test_timeout))
126 test_timeout = default_test_timeout
129 debug = os.getenv("DEBUG")
133 parser = argparse.ArgumentParser(description="VPP unit tests")
134 parser.add_argument("-f", "--failfast", action='count',
135 help="fast failure flag")
136 parser.add_argument("-d", "--dir", action='append', type=str,
137 help="directory containing test files "
138 "(may be specified multiple times)")
139 args = parser.parse_args()
140 failfast = True if args.failfast == 1 else False
142 suite = unittest.TestSuite()
143 cb = add_to_suite_callback(suite)
145 print("Adding tests from directory tree %s" % d)
146 discover_tests(d, cb)
149 retries = int(os.getenv("RETRIES"))
154 attempts = retries + 1
156 print("Perform %s attempts to pass the suite..." % attempts)
157 if debug is None or debug.lower() not in ["gdb", "gdbserver"]:
159 result, failed = run_forked(suite)
160 attempts = attempts - 1
161 print("%s test(s) failed, %s attempt(s) left" %
162 (len(failed), attempts))
163 if len(failed) > 0 and attempts > 0:
164 suite = suite_from_failed(suite, failed)
168 # don't fork if debugging..
169 sys.exit(not VppTestRunner(verbosity=verbose,
170 failfast=failfast).run(suite).wasSuccessful())