tc_id = testcase.id()
if tc_id not in self[PASS] and tc_id not in self[SKIP]:
rerun_ids.add(tc_id)
- if len(rerun_ids) > 0:
+ if rerun_ids:
return suite_from_failed(self.testcase_suite, rerun_ids)
def get_testcase_names(self, test_id):
- if re.match(r'.+\..+\..+', test_id):
+ # could be tearDownClass (test_ipsec_esp.TestIpsecEsp1)
+ setup_teardown_match = re.match(
+ r'((tearDownClass)|(setUpClass)) \((.+\..+)\)', test_id)
+ if setup_teardown_match:
+ test_name, _, _, testcase_name = setup_teardown_match.groups()
+ if len(testcase_name.split('.')) == 2:
+ for key in self.testcases_by_id.keys():
+ if key.startswith(testcase_name):
+ testcase_name = key
+ break
+ testcase_name = self._get_testcase_doc_name(testcase_name)
+ else:
test_name = self._get_test_description(test_id)
testcase_name = self._get_testcase_doc_name(test_id)
- else:
- # could be tearDownClass (test_ipsec_esp.TestIpsecEsp1)
- setup_teardown_match = re.match(
- r'((tearDownClass)|(setUpClass)) \((.+\..+)\)', test_id)
- if setup_teardown_match:
- test_name, _, _, testcase_name = setup_teardown_match.groups()
- if len(testcase_name.split('.')) == 2:
- for key in self.testcases_by_id.keys():
- if key.startswith(testcase_name):
- testcase_name = key
- break
- testcase_name = self._get_testcase_doc_name(testcase_name)
- else:
- test_name = test_id
- testcase_name = test_id
return testcase_name, test_name
def _get_test_description(self, test_id):
- return get_test_description(descriptions,
- self.testcases_by_id[test_id])
+ if test_id in self.testcases_by_id:
+ desc = get_test_description(descriptions,
+ self.testcases_by_id[test_id])
+ else:
+ desc = test_id
+ return desc
def _get_testcase_doc_name(self, test_id):
- return get_testcase_doc_name(self.testcases_by_id[test_id])
+ if test_id in self.testcases_by_id:
+ doc_name = get_testcase_doc_name(self.testcases_by_id[test_id])
+ else:
+ doc_name = test_id
+ return doc_name
def test_runner_wrapper(suite, keep_alive_pipe, stdouterr_queue,
def stdouterr_reader_wrapper(unread_testcases, finished_unread_testcases,
read_testcases):
read_testcase = None
- while read_testcases.is_set() or len(unread_testcases):
- if len(finished_unread_testcases):
+ while read_testcases.is_set() or unread_testcases:
+ if finished_unread_testcases:
read_testcase = finished_unread_testcases.pop()
unread_testcases.remove(read_testcase)
- elif len(unread_testcases):
+ elif unread_testcases:
read_testcase = unread_testcases.pop()
if read_testcase:
data = ''
"Core-file exists in test temporary directory: %s!" %
core_path)
check_core_path(logger, core_path)
- logger.debug("Running `file %s':" % core_path)
+ logger.debug("Running 'file %s':" % core_path)
try:
info = check_output(["file", core_path])
logger.debug(info)
except CalledProcessError as e:
- logger.error("Could not run `file' utility on core-file, "
- "rc=%s" % e.returncode)
+ logger.error("Subprocess returned with return code "
+ "while running `file' utility on core-file "
+ "returned: "
+ "rc=%s", e.returncode)
+ except OSError as e:
+ logger.error("Subprocess returned with OS error while "
+ "running 'file' utility "
+ "on core-file: "
+ "(%s) %s", e.errno, e.strerror)
+ except Exception as e:
+ logger.exception("Unexpected error running `file' utility "
+ "on core-file")
if vpp_pid:
# Copy api post mortem
if debug_core:
for failed_testcase in failed_testcases:
tcs_with_core = failed_testcase.testclasess_with_core
- if len(tcs_with_core) > 0:
+ if tcs_with_core:
for test, vpp_binary, tempdir in tcs_with_core.values():
check_and_handle_core(vpp_binary, tempdir, test)
manager = StreamQueueManager()
manager.start()
for i in range(concurrent_tests):
- if len(testcase_suites) > 0:
+ if testcase_suites:
wrapped_testcase_suite = TestCaseWrapper(testcase_suites.pop(0),
manager)
wrapped_testcase_suites.add(wrapped_testcase_suite)
stop_run = False
try:
- while len(wrapped_testcase_suites) > 0:
+ while wrapped_testcase_suites:
finished_testcase_suites = set()
for wrapped_testcase_suite in wrapped_testcase_suites:
while wrapped_testcase_suite.result_parent_end.poll():
finished_unread_testcases.add(finished_testcase)
finished_testcase.stdouterr_queue.put(None)
if stop_run:
- while len(testcase_suites) > 0:
+ while testcase_suites:
results.append(TestResult(testcase_suites.pop(0)))
- elif len(testcase_suites) > 0:
+ elif testcase_suites:
new_testcase = TestCaseWrapper(testcase_suites.pop(0),
manager)
wrapped_testcase_suites.add(new_testcase)
unread_testcases.add(new_testcase)
+ time.sleep(0.1)
except Exception:
for wrapped_testcase_suite in wrapped_testcase_suites:
wrapped_testcase_suite.child.terminate()
failed_testcase_ids = result[FAIL]
errored_testcase_ids = result[ERROR]
old_testcase_name = None
- if len(failed_testcase_ids) or len(errored_testcase_ids):
+ if failed_testcase_ids or errored_testcase_ids:
for failed_test_id in failed_testcase_ids:
new_testcase_name, test_name = \
result.get_testcase_names(failed_test_id)
old_testcase_name = new_testcase_name
print(' ERROR: {} [{}]'.format(
colorize(test_name, RED), failed_test_id))
- if len(self.testsuites_no_tests_run) > 0:
+ if self.testsuites_no_tests_run:
print('TESTCASES WHERE NO TESTS WERE SUCCESSFULLY EXECUTED:')
tc_classes = set()
for testsuite in self.testsuites_no_tests_run:
print("%s out of %s tests match specified filters" % (
tests_amount, tests_amount + cb.filtered.countTestCases()))
- if not running_extended_tests():
+ if not running_extended_tests:
print("Not running extended tests (some tests will be skipped)")
attempts = retries + 1
if attempts > 1:
print("Perform %s attempts to pass the suite..." % attempts)
- if run_interactive:
+ if run_interactive and suites:
# don't fork if requiring interactive terminal
+ full_suite = unittest.TestSuite()
+ map(full_suite.addTests, suites)
result = VppTestRunner(verbosity=verbose,
failfast=failfast,
- print_summary=True).run(suites[0])
+ print_summary=True).run(full_suite)
was_successful = result.wasSuccessful()
if not was_successful:
for test_case_info in result.failed_test_cases_info:
sys.exit(not was_successful)
else:
exit_code = 0
- while len(suites) > 0 and attempts > 0:
+ while suites and attempts > 0:
results = run_forked(suites)
exit_code, suites = parse_results(results)
attempts -= 1
Code Review / vpp.git / blobdiff