if logger:
logger.error(e)
+ def send_failure_through_pipe(self, test):
+ if hasattr(self, 'test_framework_failed_pipe'):
+ pipe = self.test_framework_failed_pipe
+ if pipe:
+ pipe.send(test.__class__)
+
def addFailure(self, test, err):
"""
Record a test failed result
else:
self.result_string = colorize("FAIL", RED) + ' [no temp dir]'
+ self.send_failure_through_pipe(test)
+
def addError(self, test, err):
"""
Record a test error result
else:
self.result_string = colorize("ERROR", RED) + ' [no temp dir]'
+ self.send_failure_through_pipe(test)
+
def getDescription(self, test):
"""
Get test description
self.stream.writeln("%s" % err)
+class Filter_by_test_option:
+ def __init__(self, filter_file_name, filter_class_name, filter_func_name):
+ self.filter_file_name = filter_file_name
+ self.filter_class_name = filter_class_name
+ self.filter_func_name = filter_func_name
+
+ def __call__(self, file_name, class_name, func_name):
+ if self.filter_file_name and file_name != self.filter_file_name:
+ return False
+ if self.filter_class_name and class_name != self.filter_class_name:
+ return False
+ if self.filter_func_name and func_name != self.filter_func_name:
+ return False
+ return True
+
+
class VppTestRunner(unittest.TextTestRunner):
"""
A basic test runner implementation which prints results to standard error.
"""Class maintaining the results of the tests"""
return VppTestResult
- def __init__(self, pipe=None, stream=sys.stderr, descriptions=True,
+ def __init__(self, keep_alive_pipe=None, failed_pipe=None,
+ stream=sys.stderr, descriptions=True,
verbosity=1, failfast=False, buffer=False, resultclass=None):
# ignore stream setting here, use hard-coded stdout to be in sync
# with prints from VppTestCase methods ...
verbosity, failfast, buffer,
resultclass)
reporter = KeepAliveReporter()
- reporter.pipe = pipe
+ reporter.pipe = keep_alive_pipe
+ # this is super-ugly, but very simple to implement and works as long
+ # as we run only one test at the same time
+ VppTestResult.test_framework_failed_pipe = failed_pipe
test_option = "TEST"
filter_file_name = 'test_%s' % f
return filter_file_name, filter_class_name, filter_func_name
- def filter_tests(self, tests, filter_file, filter_class, filter_func):
+ @staticmethod
+ def filter_tests(tests, filter_cb):
result = unittest.suite.TestSuite()
for t in tests:
if isinstance(t, unittest.suite.TestSuite):
# this is a bunch of tests, recursively filter...
- x = self.filter_tests(t, filter_file, filter_class,
- filter_func)
+ x = filter_tests(t, filter_cb)
if x.countTestCases() > 0:
result.addTest(x)
elif isinstance(t, unittest.TestCase):
# test_classifier.TestClassifier.test_acl_ip
# apply filtering only if it is so
if len(parts) == 3:
- if filter_file and filter_file != parts[0]:
- continue
- if filter_class and filter_class != parts[1]:
- continue
- if filter_func and filter_func != parts[2]:
+ if not filter_cb(parts[0], parts[1], parts[2]):
continue
result.addTest(t)
else:
filter_file, filter_class, filter_func = self.parse_test_option()
print("Active filters: file=%s, class=%s, function=%s" % (
filter_file, filter_class, filter_func))
- filtered = self.filter_tests(test, filter_file, filter_class,
- filter_func)
+ filter_cb = Filter_by_test_option(
+ filter_file, filter_class, filter_func)
+ filtered = self.filter_tests(test, filter_cb)
print("%s out of %s tests match specified filters" % (
filtered.countTestCases(), test.countTestCases()))
if not running_extended_tests():
from discover_tests import discover_tests
-def test_runner_wrapper(suite, keep_alive_pipe, result_pipe):
+def test_runner_wrapper(suite, keep_alive_pipe, result_pipe, failed_pipe):
result = not VppTestRunner(
- pipe=keep_alive_pipe,
+ keep_alive_pipe=keep_alive_pipe,
+ failed_pipe=failed_pipe,
verbosity=verbose,
failfast=failfast).run(suite).wasSuccessful()
result_pipe.send(result)
result_pipe.close()
keep_alive_pipe.close()
+ failed_pipe.close()
class add_to_suite_callback:
suite.addTest(cls(method))
+class Filter_by_class_list:
+ def __init__(self, class_list):
+ self.class_list = class_list
+
+ def __call__(self, file_name, class_name, func_name):
+ return class_name in self.class_list
+
+
+def suite_from_failed(suite, failed):
+ filter_cb = Filter_by_class_list(failed)
+ return VppTestRunner.filter_tests(suite, filter_cb)
+
+
def run_forked(suite):
keep_alive_parent_end, keep_alive_child_end = Pipe(duplex=False)
result_parent_end, result_child_end = Pipe(duplex=False)
+ failed_parent_end, failed_child_end = Pipe(duplex=False)
child = Process(target=test_runner_wrapper,
- args=(suite, keep_alive_child_end, result_child_end))
+ args=(suite, keep_alive_child_end, result_child_end,
+ failed_child_end))
child.start()
last_test_temp_dir = None
last_test_vpp_binary = None
last_test = None
result = None
+ failed = set()
while result is None:
readable = select.select([keep_alive_parent_end.fileno(),
result_parent_end.fileno(),
+ failed_parent_end.fileno(),
],
[], [], test_timeout)[0]
+ timeout = True
if result_parent_end.fileno() in readable:
result = result_parent_end.recv()
- elif keep_alive_parent_end.fileno() in readable:
+ timeout = False
+ if keep_alive_parent_end.fileno() in readable:
while keep_alive_parent_end.poll():
last_test, last_test_vpp_binary,\
last_test_temp_dir, vpp_pid = keep_alive_parent_end.recv()
- else:
+ timeout = False
+ if failed_parent_end.fileno() in readable:
+ while failed_parent_end.poll():
+ failed_test = failed_parent_end.recv()
+ failed.add(failed_test.__name__)
+ timeout = False
+ if timeout:
global_logger.critical("Timeout while waiting for child test "
"runner process (last test running was "
"`%s' in `%s')!" %
result = -1
keep_alive_parent_end.close()
result_parent_end.close()
- return result
+ failed_parent_end.close()
+ return result, failed
if __name__ == '__main__':
suite = unittest.TestSuite()
cb = add_to_suite_callback(suite)
for d in args.dir:
- global_logger.info("Adding tests from directory tree %s" % d)
+ print("Adding tests from directory tree %s" % d)
discover_tests(d, cb)
+ try:
+ retries = int(os.getenv("RETRIES"))
+ except:
+ retries = 0
+ if retries is None:
+ retries = 0
+ attempts = retries + 1
+ if attempts > 1:
+ print("Perform %s attempts to pass the suite..." % attempts)
if debug is None or debug.lower() not in ["gdb", "gdbserver"]:
- sys.exit(run_forked(suite))
+ while True:
+ result, failed = run_forked(suite)
+ attempts = attempts - 1
+ print("%s test(s) failed, %s attempt(s) left" %
+ (len(failed), attempts))
+ if len(failed) > 0 and attempts > 0:
+ suite = suite_from_failed(suite, failed)
+ continue
+ sys.exit(result)
# don't fork if debugging..
sys.exit(not VppTestRunner(verbosity=verbose,
Code Review / vpp.git / commitdiff