mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-27 02:39:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

Issue #7312 (new feature): Add a -F flag to run the selected tests in

Browse source 
a loop until a test fails.  Can be combined with -j.  Patch by Antoine
Pitrou.
This commit is contained in:
R. David Murray 2009-11-14 15:18:22 +00:00
parent b6fb8dcbce
commit 7f7eea651f
2 changed files with 53 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

77
Lib/test/regrtest.py
View file

@ -43,6 +43,7 @@ Special runs
-N/--nocoverdir -- Put coverage files alongside modules -N/--nocoverdir -- Put coverage files alongside modules
-t/--threshold THRESHOLD -t/--threshold THRESHOLD
-- call gc.set_threshold(THRESHOLD) -- call gc.set_threshold(THRESHOLD)
-F/--forever -- run the selected tests in a loop, until an error happens
If non-option arguments are present, they are names for tests to run, If non-option arguments are present, they are names for tests to run,
unless -x is given, in which case they are names for tests not to run. unless -x is given, in which case they are names for tests not to run.
@ -150,6 +151,7 @@ option '-uall,-bsddb'.
import cStringIO import cStringIO
import getopt import getopt
import itertools
import json import json
import os import os
import random import random
@ -219,7 +221,7 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
exclude=False, single=False, randomize=False, fromfile=None, exclude=False, single=False, randomize=False, fromfile=None,
findleaks=False, use_resources=None, trace=False, coverdir='coverage', findleaks=False, use_resources=None, trace=False, coverdir='coverage',
runleaks=False, huntrleaks=False, verbose2=False, print_slow=False, runleaks=False, huntrleaks=False, verbose2=False, print_slow=False,
random_seed=None, use_mp=None, verbose3=False): random_seed=None, use_mp=None, verbose3=False, forever=False):
"""Execute a test suite. """Execute a test suite.
This also parses command-line options and modifies its behavior This also parses command-line options and modifies its behavior
@ -244,12 +246,12 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
test_support.record_original_stdout(sys.stdout) test_support.record_original_stdout(sys.stdout)
try: try:
opts, args = getopt.getopt(sys.argv[1:], 'hvgqxsSrf:lu:t:TD:NLR:wWM:j:', opts, args = getopt.getopt(sys.argv[1:], 'hvgqxsSrf:lu:t:TD:NLR:FwWM:j:',
['help', 'verbose', 'verbose2', 'verbose3', 'quiet', ['help', 'verbose', 'verbose2', 'verbose3', 'quiet',
'exclude', 'single', 'slow', 'random', 'fromfile', 'findleaks', 'exclude', 'single', 'slow', 'random', 'fromfile', 'findleaks',
'use=', 'threshold=', 'trace', 'coverdir=', 'nocoverdir', 'use=', 'threshold=', 'trace', 'coverdir=', 'nocoverdir',
'runleaks', 'huntrleaks=', 'memlimit=', 'randseed=', 'runleaks', 'huntrleaks=', 'memlimit=', 'randseed=',
'multiprocess=', 'slaveargs=']) 'multiprocess=', 'slaveargs=', 'forever'])
except getopt.error, msg: except getopt.error, msg:
usage(2, msg) usage(2, msg)
@ -329,6 +331,8 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
use_resources.remove(r) use_resources.remove(r)
elif r not in use_resources: elif r not in use_resources:
use_resources.append(r) use_resources.append(r)
elif o in ('-F', '--forever'):
forever = True
elif o in ('-j', '--multiprocess'): elif o in ('-j', '--multiprocess'):
use_mp = int(a) use_mp = int(a)
elif o == '--slaveargs': elif o == '--slaveargs':
@ -371,8 +375,8 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
filename = os.path.join(gettempdir(), 'pynexttest') filename = os.path.join(gettempdir(), 'pynexttest')
try: try:
fp = open(filename, 'r') fp = open(filename, 'r')
next = fp.read().strip() next_test = fp.read().strip()
tests = [next] tests = [next_test]
fp.close() fp.close()
except IOError: except IOError:
pass pass
@ -411,6 +415,7 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
import trace import trace
tracer = trace.Trace(ignoredirs=[sys.prefix, sys.exec_prefix], tracer = trace.Trace(ignoredirs=[sys.prefix, sys.exec_prefix],
trace=False, count=True) trace=False, count=True)
test_times = [] test_times = []
test_support.use_resources = use_resources test_support.use_resources = use_resources
save_modules = sys.modules.keys() save_modules = sys.modules.keys()
@ -431,6 +436,17 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
skipped.append(test) skipped.append(test)
resource_denieds.append(test) resource_denieds.append(test)
if forever:
def test_forever(tests=list(tests)):
while True:
for test in tests:
yield test
if bad:
return
tests = test_forever()
else:
tests = iter(tests)
if use_mp: if use_mp:
from threading import Thread from threading import Thread
from Queue import Queue, Empty from Queue import Queue, Empty
@ -439,19 +455,21 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
debug_output_pat = re.compile(r"\[\d+ refs\]$") debug_output_pat = re.compile(r"\[\d+ refs\]$")
pending = deque() pending = deque()
output = Queue() output = Queue()
for test in tests: def tests_and_args():
args_tuple = ( for test in tests:
(test, verbose, quiet, testdir), args_tuple = (
dict(huntrleaks=huntrleaks, use_resources=use_resources) (test, verbose, quiet, testdir),
) dict(huntrleaks=huntrleaks, use_resources=use_resources)
pending.append((test, args_tuple)) )
yield (test, args_tuple)
pending = tests_and_args()
def work(): def work():
# A worker thread. # A worker thread.
try: try:
while True: while True:
try: try:
test, args_tuple = pending.popleft() test, args_tuple = next(pending)
except IndexError: except StopIteration:
output.put((None, None, None, None)) output.put((None, None, None, None))
return return
# -E is needed by some tests, e.g. test_import # -E is needed by some tests, e.g. test_import
@ -464,6 +482,9 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
# comes from the shutdown of the interpreter in the subcommand. # comes from the shutdown of the interpreter in the subcommand.
stderr = debug_output_pat.sub("", stderr) stderr = debug_output_pat.sub("", stderr)
stdout, _, result = stdout.strip().rpartition("\n") stdout, _, result = stdout.strip().rpartition("\n")
if not result:
output.put((None, None, None, None))
return
result = json.loads(result) result = json.loads(result)
if not quiet: if not quiet:
stdout = test+'\n'+stdout stdout = test+'\n'+stdout
@ -475,20 +496,22 @@ def main(tests=None, testdir=None, verbose=0, quiet=False,
for worker in workers: for worker in workers:
worker.start() worker.start()
finished = 0 finished = 0
while finished < use_mp: try:
test, stdout, stderr, result = output.get() while finished < use_mp:
if test is None: test, stdout, stderr, result = output.get()
finished += 1 if test is None:
continue finished += 1
if stdout: continue
print stdout if stdout:
if stderr: print stdout
print >>sys.stderr, stderr if stderr:
if result[0] == INTERRUPTED: print >>sys.stderr, stderr
assert result[1] == 'KeyboardInterrupt' if result[0] == INTERRUPTED:
pending.clear() assert result[1] == 'KeyboardInterrupt'
raise KeyboardInterrupt # What else? raise KeyboardInterrupt # What else?
accumulate_result(test, result) accumulate_result(test, result)
except KeyboardInterrupt:
pending.close()
for worker in workers: for worker in workers:
worker.join() worker.join()
else: else:

3
Misc/NEWS
View file

@ -1567,6 +1567,9 @@ Extension Modules
Tests Tests
----- -----
- Issue #7312: Add a -F flag to run the selected tests in a loop until
a test fails. Can be combined with -j.
- Issue #7295: Do not use a hardcoded file name in test_tarfile. - Issue #7295: Do not use a hardcoded file name in test_tarfile.
- Issue #7270: Add some dedicated unit tests for multi-thread synchronization - Issue #7270: Add some dedicated unit tests for multi-thread synchronization