help='suppress error message boxes on Windows')
group.add_argument('-F', '--forever', action='store_true',
help='run the specified tests in a loop, until an '
- 'error happens')
+ 'error happens; imply --failfast')
group.add_argument('--list-tests', action='store_true',
help="only write the name of tests that will be run, "
"don't execute them")
with open(ns.match_filename) as fp:
for line in fp:
ns.match_tests.append(line.strip())
+ if ns.forever:
+ # --forever implies --failfast
+ ns.failfast = True
return ns
findtests, runtest, get_abs_module,
STDTESTS, NOTTESTS, PASSED, FAILED, ENV_CHANGED, SKIPPED, RESOURCE_DENIED,
INTERRUPTED, CHILD_ERROR, TEST_DID_NOT_RUN,
- PROGRESS_MIN_TIME, format_test_result)
+ PROGRESS_MIN_TIME, format_test_result, is_failed)
from test.libregrtest.setup import setup_tests
from test.libregrtest.utils import removepy, count, format_duration, printlist
from test import support
test_time = time.monotonic() - start_time
if test_time >= PROGRESS_MIN_TIME:
previous_test = "%s in %s" % (previous_test, format_duration(test_time))
- elif result[0] == PASSED:
+ elif result.result == PASSED:
# be quiet: say nothing if the test passed shortly
previous_test = None
if module not in save_modules and module.startswith("test."):
support.unload(module)
+ if self.ns.failfast and is_failed(result, self.ns):
+ break
+
if previous_test:
print(previous_test)
RESOURCE_DENIED = -3
INTERRUPTED = -4
CHILD_ERROR = -5 # error in a child process
-TEST_DID_NOT_RUN = -6 # error in a child process
+TEST_DID_NOT_RUN = -6
_FORMAT_TEST_RESULT = {
PASSED: '%s passed',
FOUND_GARBAGE = []
+def is_failed(result, ns):
+ ok = result.result
+ if ok in (PASSED, RESOURCE_DENIED, SKIPPED, TEST_DID_NOT_RUN):
+ return False
+ if ok == ENV_CHANGED:
+ return ns.fail_env_changed
+ return True
+
+
def format_test_result(result):
fmt = _FORMAT_TEST_RESULT.get(result.result, "%s")
return fmt % result.test_name
from test.libregrtest.runtest import (
runtest, INTERRUPTED, CHILD_ERROR, PROGRESS_MIN_TIME,
- format_test_result, TestResult)
+ format_test_result, TestResult, is_failed)
from test.libregrtest.setup import setup_tests
from test.libregrtest.utils import format_duration
PROGRESS_UPDATE = 30.0 # seconds
-def must_stop(result):
- return result.result in (INTERRUPTED, CHILD_ERROR)
+def must_stop(result, ns):
+ if result.result == INTERRUPTED:
+ return True
+ if ns.failfast and is_failed(result, ns):
+ return True
+ return False
def run_test_in_subprocess(testname, ns):
"""A thread-safe iterator over tests for multiprocess mode."""
- def __init__(self, tests):
+ def __init__(self, tests_iter):
self.lock = threading.Lock()
- self.tests = tests
+ self.tests_iter = tests_iter
def __iter__(self):
return self
def __next__(self):
with self.lock:
- return next(self.tests)
+ if self.tests_iter is None:
+ raise StopIteration
+ return next(self.tests_iter)
+
+ def stop(self):
+ with self.lock:
+ self.tests_iter = None
MultiprocessResult = collections.namedtuple('MultiprocessResult',
self._popen = None
def kill(self):
- if not self.is_alive():
+ popen = self._popen
+ if popen is None:
return
- if self._popen is not None:
- self._popen.kill()
+ print("Kill regrtest worker process %s" % popen.pid)
+ popen.kill()
def _runtest(self, test_name):
try:
self.start_time = time.monotonic()
self.current_test_name = test_name
- popen = run_test_in_subprocess(test_name, self.ns)
- self._popen = popen
+ self._popen = run_test_in_subprocess(test_name, self.ns)
+ popen = self._popen
with popen:
try:
stdout, stderr = popen.communicate()
except:
- popen.kill()
+ self.kill()
popen.wait()
raise
mp_result = self._runtest(test_name)
self.output.put((False, mp_result))
- if must_stop(mp_result.result):
+ if must_stop(mp_result.result, self.ns):
break
except BaseException:
self.output.put((True, traceback.format_exc()))
if mp_result.stderr and not self.ns.pgo:
print(mp_result.stderr, file=sys.stderr, flush=True)
- if must_stop(mp_result.result):
+ if must_stop(mp_result.result, self.ns):
return True
return False
if self.test_timeout is not None:
faulthandler.cancel_dump_traceback_later()
+ # a test failed (and --failfast is set) or all tests completed
+ self.pending.stop()
self.wait_workers()