Python code coverage for Lib/test/libregrtest/main.py

#	count	content
1	n/a	import datetime
2	n/a	import faulthandler
3	n/a	import locale
4	n/a	import os
5	n/a	import platform
6	n/a	import random
7	n/a	import re
8	n/a	import sys
9	n/a	import sysconfig
10	n/a	import tempfile
11	n/a	import textwrap
12	n/a	import time
13	n/a	from test.libregrtest.cmdline import _parse_args
14	n/a	from test.libregrtest.runtest import (
15	n/a	findtests, runtest,
16	n/a	STDTESTS, NOTTESTS, PASSED, FAILED, ENV_CHANGED, SKIPPED, RESOURCE_DENIED,
17	n/a	INTERRUPTED, CHILD_ERROR,
18	n/a	PROGRESS_MIN_TIME, format_test_result)
19	n/a	from test.libregrtest.setup import setup_tests
20	n/a	from test import support
21	n/a	try:
22	n/a	import gc
23	n/a	except ImportError:
24	n/a	gc = None
25	n/a
26	n/a
27	n/a	# When tests are run from the Python build directory, it is best practice
28	n/a	# to keep the test files in a subfolder. This eases the cleanup of leftover
29	n/a	# files using the "make distclean" command.
30	n/a	if sysconfig.is_python_build():
31	n/a	TEMPDIR = os.path.join(sysconfig.get_config_var('srcdir'), 'build')
32	n/a	else:
33	n/a	TEMPDIR = tempfile.gettempdir()
34	n/a	TEMPDIR = os.path.abspath(TEMPDIR)
35	n/a
36	n/a
37	n/a	def format_duration(seconds):
38	n/a	if seconds < 1.0:
39	n/a	return '%.0f ms' % (seconds * 1e3)
40	n/a	if seconds < 60.0:
41	n/a	return '%.0f sec' % seconds
42	n/a
43	n/a	minutes, seconds = divmod(seconds, 60.0)
44	n/a	return '%.0f min %.0f sec' % (minutes, seconds)
45	n/a
46	n/a
47	n/a	class Regrtest:
48	n/a	"""Execute a test suite.
49	n/a
50	n/a	This also parses command-line options and modifies its behavior
51	n/a	accordingly.
52	n/a
53	n/a	tests -- a list of strings containing test names (optional)
54	n/a	testdir -- the directory in which to look for tests (optional)
55	n/a
56	n/a	Users other than the Python test suite will certainly want to
57	n/a	specify testdir; if it's omitted, the directory containing the
58	n/a	Python test suite is searched for.
59	n/a
60	n/a	If the tests argument is omitted, the tests listed on the
61	n/a	command-line will be used. If that's empty, too, then all *.py
62	n/a	files beginning with test_ will be used.
63	n/a
64	n/a	The other default arguments (verbose, quiet, exclude,
65	n/a	single, randomize, findleaks, use_resources, trace, coverdir,
66	n/a	print_slow, and random_seed) allow programmers calling main()
67	n/a	directly to set the values that would normally be set by flags
68	n/a	on the command line.
69	n/a	"""
70	n/a	def __init__(self):
71	n/a	# Namespace of command line options
72	n/a	self.ns = None
73	n/a
74	n/a	# tests
75	n/a	self.tests = []
76	n/a	self.selected = []
77	n/a
78	n/a	# test results
79	n/a	self.good = []
80	n/a	self.bad = []
81	n/a	self.skipped = []
82	n/a	self.resource_denieds = []
83	n/a	self.environment_changed = []
84	n/a	self.interrupted = False
85	n/a
86	n/a	# used by --slow
87	n/a	self.test_times = []
88	n/a
89	n/a	# used by --coverage, trace.Trace instance
90	n/a	self.tracer = None
91	n/a
92	n/a	# used by --findleaks, store for gc.garbage
93	n/a	self.found_garbage = []
94	n/a
95	n/a	# used to display the progress bar "[ 3/100]"
96	n/a	self.start_time = time.monotonic()
97	n/a	self.test_count = ''
98	n/a	self.test_count_width = 1
99	n/a
100	n/a	# used by --single
101	n/a	self.next_single_test = None
102	n/a	self.next_single_filename = None
103	n/a
104	n/a	def accumulate_result(self, test, result):
105	n/a	ok, test_time = result
106	n/a	if ok not in (CHILD_ERROR, INTERRUPTED):
107	n/a	self.test_times.append((test_time, test))
108	n/a	if ok == PASSED:
109	n/a	self.good.append(test)
110	n/a	elif ok in (FAILED, CHILD_ERROR):
111	n/a	self.bad.append(test)
112	n/a	elif ok == ENV_CHANGED:
113	n/a	self.environment_changed.append(test)
114	n/a	elif ok == SKIPPED:
115	n/a	self.skipped.append(test)
116	n/a	elif ok == RESOURCE_DENIED:
117	n/a	self.skipped.append(test)
118	n/a	self.resource_denieds.append(test)
119	n/a
120	n/a	def display_progress(self, test_index, test):
121	n/a	if self.ns.quiet:
122	n/a	return
123	n/a	if self.bad and not self.ns.pgo:
124	n/a	fmt = "{time} [{test_index:{count_width}}{test_count}/{nbad}] {test_name}"
125	n/a	else:
126	n/a	fmt = "{time} [{test_index:{count_width}}{test_count}] {test_name}"
127	n/a	test_time = time.monotonic() - self.start_time
128	n/a	test_time = datetime.timedelta(seconds=int(test_time))
129	n/a	line = fmt.format(count_width=self.test_count_width,
130	n/a	test_index=test_index,
131	n/a	test_count=self.test_count,
132	n/a	nbad=len(self.bad),
133	n/a	test_name=test,
134	n/a	time=test_time)
135	n/a	print(line, flush=True)
136	n/a
137	n/a	def parse_args(self, kwargs):
138	n/a	ns = _parse_args(sys.argv[1:], **kwargs)
139	n/a
140	n/a	if ns.timeout and not hasattr(faulthandler, 'dump_traceback_later'):
141	n/a	print("Warning: The timeout option requires "
142	n/a	"faulthandler.dump_traceback_later", file=sys.stderr)
143	n/a	ns.timeout = None
144	n/a
145	n/a	if ns.threshold is not None and gc is None:
146	n/a	print('No GC available, ignore --threshold.', file=sys.stderr)
147	n/a	ns.threshold = None
148	n/a
149	n/a	if ns.findleaks:
150	n/a	if gc is not None:
151	n/a	# Uncomment the line below to report garbage that is not
152	n/a	# freeable by reference counting alone. By default only
153	n/a	# garbage that is not collectable by the GC is reported.
154	n/a	pass
155	n/a	#gc.set_debug(gc.DEBUG_SAVEALL)
156	n/a	else:
157	n/a	print('No GC available, disabling --findleaks',
158	n/a	file=sys.stderr)
159	n/a	ns.findleaks = False
160	n/a
161	n/a	# Strip .py extensions.
162	n/a	removepy(ns.args)
163	n/a
164	n/a	return ns
165	n/a
166	n/a	def find_tests(self, tests):
167	n/a	self.tests = tests
168	n/a
169	n/a	if self.ns.single:
170	n/a	self.next_single_filename = os.path.join(TEMPDIR, 'pynexttest')
171	n/a	try:
172	n/a	with open(self.next_single_filename, 'r') as fp:
173	n/a	next_test = fp.read().strip()
174	n/a	self.tests = [next_test]
175	n/a	except OSError:
176	n/a	pass
177	n/a
178	n/a	if self.ns.fromfile:
179	n/a	self.tests = []
180	n/a	# regex to match 'test_builtin' in line:
181	n/a	# '0:00:00 [ 4/400] test_builtin -- test_dict took 1 sec'
182	n/a	regex = re.compile(r'\btest_[a-zA-Z0-9_]+\b')
183	n/a	with open(os.path.join(support.SAVEDCWD, self.ns.fromfile)) as fp:
184	n/a	for line in fp:
185	n/a	line = line.split('#', 1)[0]
186	n/a	line = line.strip()
187	n/a	match = regex.search(line)
188	n/a	if match is not None:
189	n/a	self.tests.append(match.group())
190	n/a
191	n/a	removepy(self.tests)
192	n/a
193	n/a	stdtests = STDTESTS[:]
194	n/a	nottests = NOTTESTS.copy()
195	n/a	if self.ns.exclude:
196	n/a	for arg in self.ns.args:
197	n/a	if arg in stdtests:
198	n/a	stdtests.remove(arg)
199	n/a	nottests.add(arg)
200	n/a	self.ns.args = []
201	n/a
202	n/a	# if testdir is set, then we are not running the python tests suite, so
203	n/a	# don't add default tests to be executed or skipped (pass empty values)
204	n/a	if self.ns.testdir:
205	n/a	alltests = findtests(self.ns.testdir, list(), set())
206	n/a	else:
207	n/a	alltests = findtests(self.ns.testdir, stdtests, nottests)
208	n/a
209	n/a	if not self.ns.fromfile:
210	n/a	self.selected = self.tests or self.ns.args or alltests
211	n/a	else:
212	n/a	self.selected = self.tests
213	n/a	if self.ns.single:
214	n/a	self.selected = self.selected[:1]
215	n/a	try:
216	n/a	pos = alltests.index(self.selected[0])
217	n/a	self.next_single_test = alltests[pos + 1]
218	n/a	except IndexError:
219	n/a	pass
220	n/a
221	n/a	# Remove all the selected tests that precede start if it's set.
222	n/a	if self.ns.start:
223	n/a	try:
224	n/a	del self.selected[:self.selected.index(self.ns.start)]
225	n/a	except ValueError:
226	n/a	print("Couldn't find starting test (%s), using all tests"
227	n/a	% self.ns.start, file=sys.stderr)
228	n/a
229	n/a	if self.ns.randomize:
230	n/a	if self.ns.random_seed is None:
231	n/a	self.ns.random_seed = random.randrange(10000000)
232	n/a	random.seed(self.ns.random_seed)
233	n/a	random.shuffle(self.selected)
234	n/a
235	n/a	def list_tests(self):
236	n/a	for name in self.selected:
237	n/a	print(name)
238	n/a
239	n/a	def rerun_failed_tests(self):
240	n/a	self.ns.verbose = True
241	n/a	self.ns.failfast = False
242	n/a	self.ns.verbose3 = False
243	n/a	self.ns.match_tests = None
244	n/a
245	n/a	print("Re-running failed tests in verbose mode")
246	n/a	for test in self.bad[:]:
247	n/a	print("Re-running test %r in verbose mode" % test, flush=True)
248	n/a	try:
249	n/a	self.ns.verbose = True
250	n/a	ok = runtest(self.ns, test)
251	n/a	except KeyboardInterrupt:
252	n/a	self.interrupted = True
253	n/a	# print a newline separate from the ^C
254	n/a	print()
255	n/a	break
256	n/a	else:
257	n/a	if ok[0] in {PASSED, ENV_CHANGED, SKIPPED, RESOURCE_DENIED}:
258	n/a	self.bad.remove(test)
259	n/a	else:
260	n/a	if self.bad:
261	n/a	print(count(len(self.bad), 'test'), "failed again:")
262	n/a	printlist(self.bad)
263	n/a
264	n/a	def display_result(self):
265	n/a	if self.interrupted:
266	n/a	# print a newline after ^C
267	n/a	print()
268	n/a	print("Test suite interrupted by signal SIGINT.")
269	n/a	executed = set(self.good) \| set(self.bad) \| set(self.skipped)
270	n/a	omitted = set(self.selected) - executed
271	n/a	print(count(len(omitted), "test"), "omitted:")
272	n/a	printlist(omitted)
273	n/a
274	n/a	# If running the test suite for PGO then no one cares about
275	n/a	# results.
276	n/a	if self.ns.pgo:
277	n/a	return
278	n/a
279	n/a	if self.good and not self.ns.quiet:
280	n/a	if (not self.bad
281	n/a	and not self.skipped
282	n/a	and not self.interrupted
283	n/a	and len(self.good) > 1):
284	n/a	print("All", end=' ')
285	n/a	print(count(len(self.good), "test"), "OK.")
286	n/a
287	n/a	if self.ns.print_slow:
288	n/a	self.test_times.sort(reverse=True)
289	n/a	print()
290	n/a	print("10 slowest tests:")
291	n/a	for time, test in self.test_times[:10]:
292	n/a	print("- %s: %s" % (test, format_duration(time)))
293	n/a
294	n/a	if self.bad:
295	n/a	print()
296	n/a	print(count(len(self.bad), "test"), "failed:")
297	n/a	printlist(self.bad)
298	n/a
299	n/a	if self.environment_changed:
300	n/a	print()
301	n/a	print("{} altered the execution environment:".format(
302	n/a	count(len(self.environment_changed), "test")))
303	n/a	printlist(self.environment_changed)
304	n/a
305	n/a	if self.skipped and not self.ns.quiet:
306	n/a	print()
307	n/a	print(count(len(self.skipped), "test"), "skipped:")
308	n/a	printlist(self.skipped)
309	n/a
310	n/a	def run_tests_sequential(self):
311	n/a	if self.ns.trace:
312	n/a	import trace
313	n/a	self.tracer = trace.Trace(trace=False, count=True)
314	n/a
315	n/a	save_modules = sys.modules.keys()
316	n/a
317	n/a	print("Run tests sequentially")
318	n/a
319	n/a	previous_test = None
320	n/a	for test_index, test in enumerate(self.tests, 1):
321	n/a	start_time = time.monotonic()
322	n/a
323	n/a	text = test
324	n/a	if previous_test:
325	n/a	text = '%s -- %s' % (text, previous_test)
326	n/a	self.display_progress(test_index, text)
327	n/a
328	n/a	if self.tracer:
329	n/a	# If we're tracing code coverage, then we don't exit with status
330	n/a	# if on a false return value from main.
331	n/a	cmd = ('result = runtest(self.ns, test); '
332	n/a	'self.accumulate_result(test, result)')
333	n/a	ns = dict(locals())
334	n/a	self.tracer.runctx(cmd, globals=globals(), locals=ns)
335	n/a	result = ns['result']
336	n/a	else:
337	n/a	try:
338	n/a	result = runtest(self.ns, test)
339	n/a	except KeyboardInterrupt:
340	n/a	self.interrupted = True
341	n/a	self.accumulate_result(test, (INTERRUPTED, None))
342	n/a	break
343	n/a	else:
344	n/a	self.accumulate_result(test, result)
345	n/a
346	n/a	previous_test = format_test_result(test, result[0])
347	n/a	test_time = time.monotonic() - start_time
348	n/a	if test_time >= PROGRESS_MIN_TIME:
349	n/a	previous_test = "%s in %s" % (previous_test, format_duration(test_time))
350	n/a	elif result[0] == PASSED:
351	n/a	# be quiet: say nothing if the test passed shortly
352	n/a	previous_test = None
353	n/a
354	n/a	if self.ns.findleaks:
355	n/a	gc.collect()
356	n/a	if gc.garbage:
357	n/a	print("Warning: test created", len(gc.garbage), end=' ')
358	n/a	print("uncollectable object(s).")
359	n/a	# move the uncollectable objects somewhere so we don't see
360	n/a	# them again
361	n/a	self.found_garbage.extend(gc.garbage)
362	n/a	del gc.garbage[:]
363	n/a
364	n/a	# Unload the newly imported modules (best effort finalization)
365	n/a	for module in sys.modules.keys():
366	n/a	if module not in save_modules and module.startswith("test."):
367	n/a	support.unload(module)
368	n/a
369	n/a	if previous_test:
370	n/a	print(previous_test)
371	n/a
372	n/a	def _test_forever(self, tests):
373	n/a	while True:
374	n/a	for test in tests:
375	n/a	yield test
376	n/a	if self.bad:
377	n/a	return
378	n/a
379	n/a	def run_tests(self):
380	n/a	# For a partial run, we do not need to clutter the output.
381	n/a	if (self.ns.verbose
382	n/a	or self.ns.header
383	n/a	or not (self.ns.pgo or self.ns.quiet or self.ns.single
384	n/a	or self.tests or self.ns.args)):
385	n/a	# Print basic platform information
386	n/a	print("==", platform.python_implementation(), *sys.version.split())
387	n/a	print("== ", platform.platform(aliased=True),
388	n/a	"%s-endian" % sys.byteorder)
389	n/a	print("== ", "hash algorithm:", sys.hash_info.algorithm,
390	n/a	"64bit" if sys.maxsize > 2**32 else "32bit")
391	n/a	print("== cwd:", os.getcwd())
392	n/a	print("== encodings: locale=%s, FS=%s"
393	n/a	% (locale.getpreferredencoding(False),
394	n/a	sys.getfilesystemencoding()))
395	n/a	print("Testing with flags:", sys.flags)
396	n/a
397	n/a	if self.ns.randomize:
398	n/a	print("Using random seed", self.ns.random_seed)
399	n/a
400	n/a	if self.ns.forever:
401	n/a	self.tests = self._test_forever(list(self.selected))
402	n/a	self.test_count = ''
403	n/a	self.test_count_width = 3
404	n/a	else:
405	n/a	self.tests = iter(self.selected)
406	n/a	self.test_count = '/{}'.format(len(self.selected))
407	n/a	self.test_count_width = len(self.test_count) - 1
408	n/a
409	n/a	if self.ns.use_mp:
410	n/a	from test.libregrtest.runtest_mp import run_tests_multiprocess
411	n/a	run_tests_multiprocess(self)
412	n/a	else:
413	n/a	self.run_tests_sequential()
414	n/a
415	n/a	def finalize(self):
416	n/a	if self.next_single_filename:
417	n/a	if self.next_single_test:
418	n/a	with open(self.next_single_filename, 'w') as fp:
419	n/a	fp.write(self.next_single_test + '\n')
420	n/a	else:
421	n/a	os.unlink(self.next_single_filename)
422	n/a
423	n/a	if self.tracer:
424	n/a	r = self.tracer.results()
425	n/a	r.write_results(show_missing=True, summary=True,
426	n/a	coverdir=self.ns.coverdir)
427	n/a
428	n/a	print()
429	n/a	duration = time.monotonic() - self.start_time
430	n/a	print("Total duration: %s" % format_duration(duration))
431	n/a
432	n/a	if self.bad:
433	n/a	result = "FAILURE"
434	n/a	elif self.interrupted:
435	n/a	result = "INTERRUPTED"
436	n/a	else:
437	n/a	result = "SUCCESS"
438	n/a	print("Tests result: %s" % result)
439	n/a
440	n/a	if self.ns.runleaks:
441	n/a	os.system("leaks %d" % os.getpid())
442	n/a
443	n/a	def main(self, tests=None, **kwargs):
444	n/a	global TEMPDIR
445	n/a
446	n/a	if sysconfig.is_python_build():
447	n/a	try:
448	n/a	os.mkdir(TEMPDIR)
449	n/a	except FileExistsError:
450	n/a	pass
451	n/a
452	n/a	# Define a writable temp dir that will be used as cwd while running
453	n/a	# the tests. The name of the dir includes the pid to allow parallel
454	n/a	# testing (see the -j option).
455	n/a	test_cwd = 'test_python_{}'.format(os.getpid())
456	n/a	test_cwd = os.path.join(TEMPDIR, test_cwd)
457	n/a
458	n/a	# Run the tests in a context manager that temporarily changes the CWD to a
459	n/a	# temporary and writable directory. If it's not possible to create or
460	n/a	# change the CWD, the original CWD will be used. The original CWD is
461	n/a	# available from support.SAVEDCWD.
462	n/a	with support.temp_cwd(test_cwd, quiet=True):
463	n/a	self._main(tests, kwargs)
464	n/a
465	n/a	def _main(self, tests, kwargs):
466	n/a	self.ns = self.parse_args(kwargs)
467	n/a
468	n/a	if self.ns.slaveargs is not None:
469	n/a	from test.libregrtest.runtest_mp import run_tests_slave
470	n/a	run_tests_slave(self.ns.slaveargs)
471	n/a
472	n/a	if self.ns.wait:
473	n/a	input("Press any key to continue...")
474	n/a
475	n/a	support.PGO = self.ns.pgo
476	n/a
477	n/a	setup_tests(self.ns)
478	n/a
479	n/a	self.find_tests(tests)
480	n/a
481	n/a	if self.ns.list_tests:
482	n/a	self.list_tests()
483	n/a	sys.exit(0)
484	n/a
485	n/a	self.run_tests()
486	n/a	self.display_result()
487	n/a
488	n/a	if self.ns.verbose2 and self.bad:
489	n/a	self.rerun_failed_tests()
490	n/a
491	n/a	self.finalize()
492	n/a	sys.exit(len(self.bad) > 0 or self.interrupted)
493	n/a
494	n/a
495	n/a	def removepy(names):
496	n/a	if not names:
497	n/a	return
498	n/a	for idx, name in enumerate(names):
499	n/a	basename, ext = os.path.splitext(name)
500	n/a	if ext == '.py':
501	n/a	names[idx] = basename
502	n/a
503	n/a
504	n/a	def count(n, word):
505	n/a	if n == 1:
506	n/a	return "%d %s" % (n, word)
507	n/a	else:
508	n/a	return "%d %ss" % (n, word)
509	n/a
510	n/a
511	n/a	def printlist(x, width=70, indent=4):
512	n/a	"""Print the elements of iterable x to stdout.
513	n/a
514	n/a	Optional arg width (default 70) is the maximum line length.
515	n/a	Optional arg indent (default 4) is the number of blanks with which to
516	n/a	begin each line.
517	n/a	"""
518	n/a
519	n/a	blanks = ' ' * indent
520	n/a	# Print the sorted list: 'x' may be a '--random' list or a set()
521	n/a	print(textwrap.fill(' '.join(str(elt) for elt in sorted(x)), width,
522	n/a	initial_indent=blanks, subsequent_indent=blanks))
523	n/a
524	n/a
525	n/a	def main(tests=None, **kwargs):
526	n/a	"""Run the Python suite."""
527	n/a	Regrtest().main(tests=tests, **kwargs)