9 from multiprocessing import Process, Pipe
10 from framework import VppTestRunner
11 from debug import spawn_gdb
12 from log import global_logger
15 def add_from_dir(suite, directory):
17 for _f in os.listdir(directory):
18 f = "%s/%s" % (directory, _f)
20 add_from_dir(suite, f)
22 if not os.path.isfile(f):
25 sys.path.insert(0, directory)
27 if not _f.startswith("test_") or not _f.endswith(".py"):
29 name = "".join(f.split("/")[-1].split(".")[:-1])
30 if name in sys.modules:
31 raise Exception("Duplicate test module `%s' found!" % name)
32 module = importlib.import_module(name)
33 for name, cls in module.__dict__.items():
34 if not isinstance(cls, type):
36 if not issubclass(cls, unittest.TestCase):
38 if name == "VppTestCase":
40 for method in dir(cls):
41 if not callable(getattr(cls, method)):
43 if method.startswith("test_"):
44 suite.addTest(cls(method))
47 def test_runner_wrapper(suite, keep_alive_pipe, result_pipe):
48 result = not VppTestRunner(
51 failfast=failfast).run(suite).wasSuccessful()
52 result_pipe.send(result)
54 keep_alive_pipe.close()
57 def run_forked(suite):
58 keep_alive_parent_end, keep_alive_child_end = Pipe(duplex=False)
59 result_parent_end, result_child_end = Pipe(duplex=False)
61 child = Process(target=test_runner_wrapper,
62 args=(suite, keep_alive_child_end, result_child_end))
64 last_test_temp_dir = None
65 last_test_vpp_binary = None
69 readable = select.select([keep_alive_parent_end.fileno(),
70 result_parent_end.fileno(),
72 [], [], test_timeout)[0]
73 if result_parent_end.fileno() in readable:
74 result = result_parent_end.recv()
75 elif keep_alive_parent_end.fileno() in readable:
76 while keep_alive_parent_end.poll():
77 last_test, last_test_vpp_binary, last_test_temp_dir =\
78 keep_alive_parent_end.recv()
80 global_logger.critical("Timeout while waiting for child test "
81 "runner process (last test running was "
83 (last_test, last_test_temp_dir))
84 if last_test_temp_dir and last_test_vpp_binary:
85 core_path = "%s/core" % last_test_temp_dir
86 if os.path.isfile(core_path):
87 global_logger.error("Core-file exists in test temporary "
88 "directory: %s!" % core_path)
89 if d and d.lower() == "core":
90 spawn_gdb(last_test_vpp_binary, core_path,
94 keep_alive_parent_end.close()
95 result_parent_end.close()
99 if __name__ == '__main__':
102 verbose = int(os.getenv("V", 0))
106 default_test_timeout = 600 # 10 minutes
108 test_timeout = int(os.getenv("TIMEOUT", default_test_timeout))
110 test_timeout = default_test_timeout
113 debug = os.getenv("DEBUG")
117 parser = argparse.ArgumentParser(description="VPP unit tests")
118 parser.add_argument("-f", "--failfast", action='count',
119 help="fast failure flag")
120 parser.add_argument("-d", "--dir", action='append', type=str,
121 help="directory containing test files "
122 "(may be specified multiple times)")
123 args = parser.parse_args()
124 failfast = True if args.failfast == 1 else False
126 suite = unittest.TestSuite()
128 global_logger.info("Adding tests from directory tree %s" % d)
129 add_from_dir(suite, d)
131 if debug is None or debug.lower() not in ["gdb", "gdbserver"]:
132 sys.exit(run_forked(suite))
134 # don't fork if debugging..
135 sys.exit(not VppTestRunner(verbosity=verbose,
136 failfast=failfast).run(suite).wasSuccessful())