+class TestCaseWrapper(object):
+ def __init__(self, testcase_suite, manager):
+ self.keep_alive_parent_end, self.keep_alive_child_end = Pipe(
+ duplex=False)
+ self.result_parent_end, self.result_child_end = Pipe(duplex=False)
+ self.testcase_suite = testcase_suite
+ self.stdouterr_queue = manager.Queue()
+ self.logger = get_parallel_logger(self.stdouterr_queue)
+ self.child = Process(target=test_runner_wrapper,
+ args=(testcase_suite, self.keep_alive_child_end,
+ self.result_child_end, self.stdouterr_queue,
+ self.logger)
+ )
+ self.child.start()
+ self.pid = self.child.pid
+ self.last_test_temp_dir = None
+ self.last_test_vpp_binary = None
+ self.last_test = None
+ self.result = None
+ self.last_heard = time.time()
+ self.core_detected_at = None
+ self.failed_tests = []
+
+ def close_pipes(self):
+ self.keep_alive_child_end.close()
+ self.result_child_end.close()
+ self.keep_alive_parent_end.close()
+ self.result_parent_end.close()
+
+
+def stdouterr_reader_wrapper(unread_testcases, finished_unread_testcases,
+ read_testcases):
+ read_testcase = None
+ while read_testcases.is_set() or len(unread_testcases) > 0:
+ if not read_testcase:
+ if len(finished_unread_testcases) > 0:
+ read_testcase = finished_unread_testcases.pop()
+ unread_testcases.remove(read_testcase)
+ elif len(unread_testcases) > 0:
+ read_testcase = unread_testcases.pop()
+ if read_testcase:
+ data = ''
+ while data is not None:
+ sys.stdout.write(data)
+ data = read_testcase.stdouterr_queue.get()
+
+ read_testcase.stdouterr_queue.close()
+ finished_unread_testcases.discard(read_testcase)
+ read_testcase = None
+
+
+def run_forked(testcases):
+ wrapped_testcase_suites = set()
+
+ # suites are unhashable, need to use list
+ results = []
+ debug_core = os.getenv("DEBUG", "").lower() == "core"
+ unread_testcases = set()
+ finished_unread_testcases = set()
+ manager = StreamQueueManager()
+ manager.start()
+ for i in range(concurrent_tests):
+ if len(testcases) > 0:
+ wrapped_testcase_suite = TestCaseWrapper(testcases.pop(0), manager)
+ wrapped_testcase_suites.add(wrapped_testcase_suite)
+ unread_testcases.add(wrapped_testcase_suite)
+ # time.sleep(1)
+ else:
+ break
+
+ read_from_testcases = threading.Event()
+ read_from_testcases.set()
+ stdouterr_thread = threading.Thread(target=stdouterr_reader_wrapper,
+ args=(unread_testcases,
+ finished_unread_testcases,
+ read_from_testcases))
+ stdouterr_thread.start()
+
+ while len(wrapped_testcase_suites) > 0:
+ finished_testcase_suites = set()
+ for wrapped_testcase_suite in wrapped_testcase_suites:
+ readable = select.select(
+ [wrapped_testcase_suite.keep_alive_parent_end.fileno(),
+ wrapped_testcase_suite.result_parent_end.fileno()],
+ [], [], 1)[0]
+ if wrapped_testcase_suite.result_parent_end.fileno() in readable:
+ results.append(
+ (wrapped_testcase_suite.testcase_suite,
+ wrapped_testcase_suite.result_parent_end.recv()))
+ finished_testcase_suites.add(wrapped_testcase_suite)
+ continue
+
+ if wrapped_testcase_suite.keep_alive_parent_end.fileno() \
+ in readable:
+ while wrapped_testcase_suite.keep_alive_parent_end.poll():
+ wrapped_testcase_suite.last_test, \
+ wrapped_testcase_suite.last_test_vpp_binary, \
+ wrapped_testcase_suite.last_test_temp_dir, \
+ wrapped_testcase_suite.vpp_pid = \
+ wrapped_testcase_suite.keep_alive_parent_end.recv()
+ wrapped_testcase_suite.last_heard = time.time()
+
+ fail = False
+ if wrapped_testcase_suite.last_heard + test_timeout < time.time() \
+ and not os.path.isfile(
+ "%s/_core_handled" %
+ wrapped_testcase_suite.last_test_temp_dir):
+ fail = True
+ wrapped_testcase_suite.logger.critical(
+ "Timeout while waiting for child test "
+ "runner process (last test running was "
+ "`%s' in `%s')!" %
+ (wrapped_testcase_suite.last_test,
+ wrapped_testcase_suite.last_test_temp_dir))
+ elif not wrapped_testcase_suite.child.is_alive():
+ fail = True
+ wrapped_testcase_suite.logger.critical(
+ "Child python process unexpectedly died "
+ "(last test running was `%s' in `%s')!" %
+ (wrapped_testcase_suite.last_test,
+ wrapped_testcase_suite.last_test_temp_dir))
+ elif wrapped_testcase_suite.last_test_temp_dir and \
+ wrapped_testcase_suite.last_test_vpp_binary:
+ core_path = "%s/core" % \
+ wrapped_testcase_suite.last_test_temp_dir
+ if os.path.isfile(core_path):
+ if wrapped_testcase_suite.core_detected_at is None:
+ wrapped_testcase_suite.core_detected_at = time.time()
+ elif wrapped_testcase_suite.core_detected_at + \
+ core_timeout < time.time():
+ if not os.path.isfile(
+ "%s/_core_handled" %
+ wrapped_testcase_suite.
+ last_test_temp_dir):
+ wrapped_testcase_suite.logger.critical(
+ "Child python process unresponsive and core-"
+ "file exists in test temporary directory!")
+ fail = True
+
+ if fail:
+ failed_dir = os.getenv('VPP_TEST_FAILED_DIR')
+ lttd = os.path.basename(
+ wrapped_testcase_suite.last_test_temp_dir)
+ link_path = '%s%s-FAILED' % (failed_dir, lttd)
+ wrapped_testcase_suite.logger.error(
+ "Creating a link to the failed test: %s -> %s" %
+ (link_path, lttd))
+ if not os.path.exists(link_path):
+ os.symlink(wrapped_testcase_suite.last_test_temp_dir,
+ link_path)
+ api_post_mortem_path = "/tmp/api_post_mortem.%d" % \
+ wrapped_testcase_suite.vpp_pid
+ if os.path.isfile(api_post_mortem_path):
+ wrapped_testcase_suite.logger.error(
+ "Copying api_post_mortem.%d to %s" %
+ (wrapped_testcase_suite.vpp_pid,
+ wrapped_testcase_suite.last_test_temp_dir))
+ shutil.copy2(api_post_mortem_path,
+ wrapped_testcase_suite.last_test_temp_dir)
+ if wrapped_testcase_suite.last_test_temp_dir and \
+ wrapped_testcase_suite.last_test_vpp_binary:
+ core_path = "%s/core" % \
+ wrapped_testcase_suite.last_test_temp_dir
+ if os.path.isfile(core_path):
+ wrapped_testcase_suite.logger.error(
+ "Core-file exists in test temporary directory: %s!"
+ % core_path)
+ check_core_path(wrapped_testcase_suite.logger,
+ core_path)
+ wrapped_testcase_suite.logger.debug(
+ "Running `file %s':" % core_path)
+ try:
+ info = check_output(["file", core_path])
+ wrapped_testcase_suite.logger.debug(info)
+ except CalledProcessError as e:
+ wrapped_testcase_suite.logger.error(
+ "Could not run `file' utility on core-file, "
+ "rc=%s" % e.returncode)
+ pass
+ if debug_core:
+ spawn_gdb(
+ wrapped_testcase_suite.last_test_vpp_binary,
+ core_path, wrapped_testcase_suite.logger)
+ wrapped_testcase_suite.child.terminate()
+ try:
+ # terminating the child process tends to leave orphan
+ # VPP process around
+ os.kill(wrapped_testcase_suite.vpp_pid, signal.SIGTERM)
+ except OSError:
+ # already dead
+ pass
+ results.append((wrapped_testcase_suite.testcase_suite, None))
+ finished_testcase_suites.add(wrapped_testcase_suite)
+
+ for finished_testcase in finished_testcase_suites:
+ finished_testcase.child.join()
+ finished_testcase.close_pipes()
+ wrapped_testcase_suites.remove(finished_testcase)
+ finished_unread_testcases.add(finished_testcase)
+ finished_testcase.stdouterr_queue.put(None)
+ if len(testcases) > 0:
+ new_testcase = TestCaseWrapper(testcases.pop(0), manager)
+ wrapped_testcase_suites.add(new_testcase)
+ unread_testcases.add(new_testcase)
+
+ read_from_testcases.clear()
+ stdouterr_thread.join(test_timeout)
+ manager.shutdown()
+ return results
+
+
+class SplitToSuitesCallback:
+ def __init__(self, filter_callback):
+ self.suites = {}
+ self.suite_name = 'default'
+ self.filter_callback = filter_callback
+ self.filtered = unittest.TestSuite()