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(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 handle_core(vpp_binary, core_path):
59 d = os.getenv("DEBUG")
62 if d and d.lower() == "core":
63 spawn_gdb(vpp_binary, core_path, global_logger)
66 if __name__ == '__main__':
68 verbose = int(os.getenv("V", 0))
72 default_test_timeout = 600 # 10 minutes
74 test_timeout = int(os.getenv("TIMEOUT", default_test_timeout))
76 test_timeout = default_test_timeout
78 parser = argparse.ArgumentParser(description="VPP unit tests")
79 parser.add_argument("-f", "--failfast", action='count',
80 help="fast failure flag")
81 parser.add_argument("-d", "--dir", action='append', type=str,
82 help="directory containing test files "
83 "(may be specified multiple times)")
84 args = parser.parse_args()
85 failfast = True if args.failfast == 1 else False
87 suite = unittest.TestSuite()
89 global_logger.info("Adding tests from directory tree %s" % d)
90 add_from_dir(suite, d)
91 keep_alive_parent_end, keep_alive_child_end = Pipe(duplex=False)
92 result_parent_end, result_child_end = Pipe(duplex=False)
94 p = Process(target=test_runner_wrapper,
95 args=(keep_alive_child_end,
98 last_test_temp_dir = None
99 last_test_vpp_binary = None
102 while result is None:
103 readable = select.select([keep_alive_parent_end.fileno(),
104 result_parent_end.fileno(),
106 [], [], test_timeout)[0]
107 if result_parent_end.fileno() in readable:
108 result = result_parent_end.recv()
109 elif keep_alive_parent_end.fileno() in readable:
110 while keep_alive_parent_end.poll():
111 last_test, last_test_vpp_binary, last_test_temp_dir =\
112 keep_alive_parent_end.recv()
114 global_logger.critical("Timeout while waiting for child test "
115 "runner process (last test running was "
117 (last_test, last_test_temp_dir))
118 if last_test_temp_dir and last_test_vpp_binary:
119 core_path = "%s/core" % last_test_temp_dir
120 if os.path.isfile(core_path):
121 global_logger.error("Core-file exists in test temporary "
122 "directory: %s!" % core_path)
123 handle_core(last_test_vpp_binary, core_path)
126 keep_alive_parent_end.close()
127 result_parent_end.close()