python (3.11.7)
1 import os
2 import random
3 import re
4 import shlex
5 import sys
6 import sysconfig
7 import time
8
9 from test import support
10 from test.support import os_helper, MS_WINDOWS
11
12 from .cmdline import _parse_args, Namespace
13 from .findtests import findtests, split_test_packages, list_cases
14 from .logger import Logger
15 from .pgo import setup_pgo_tests
16 from .result import State
17 from .results import TestResults, EXITCODE_INTERRUPTED
18 from .runtests import RunTests, HuntRefleak
19 from .setup import setup_process, setup_test_dir
20 from .single import run_single_test, PROGRESS_MIN_TIME
21 from .utils import (
22 StrPath, StrJSON, TestName, TestList, TestTuple, TestFilter,
23 strip_py_suffix, count, format_duration,
24 printlist, get_temp_dir, get_work_dir, exit_timeout,
25 display_header, cleanup_temp_dir, print_warning,
26 is_cross_compiled, get_host_runner, process_cpu_count,
27 EXIT_TIMEOUT)
28
29
30 class ESC[4;38;5;81mRegrtest:
31 """Execute a test suite.
32
33 This also parses command-line options and modifies its behavior
34 accordingly.
35
36 tests -- a list of strings containing test names (optional)
37 testdir -- the directory in which to look for tests (optional)
38
39 Users other than the Python test suite will certainly want to
40 specify testdir; if it's omitted, the directory containing the
41 Python test suite is searched for.
42
43 If the tests argument is omitted, the tests listed on the
44 command-line will be used. If that's empty, too, then all *.py
45 files beginning with test_ will be used.
46
47 The other default arguments (verbose, quiet, exclude,
48 single, randomize, use_resources, trace, coverdir,
49 print_slow, and random_seed) allow programmers calling main()
50 directly to set the values that would normally be set by flags
51 on the command line.
52 """
53 def __init__(self, ns: Namespace, _add_python_opts: bool = False):
54 # Log verbosity
55 self.verbose: int = int(ns.verbose)
56 self.quiet: bool = ns.quiet
57 self.pgo: bool = ns.pgo
58 self.pgo_extended: bool = ns.pgo_extended
59
60 # Test results
61 self.results: TestResults = TestResults()
62 self.first_state: str | None = None
63
64 # Logger
65 self.logger = Logger(self.results, self.quiet, self.pgo)
66
67 # Actions
68 self.want_header: bool = ns.header
69 self.want_list_tests: bool = ns.list_tests
70 self.want_list_cases: bool = ns.list_cases
71 self.want_wait: bool = ns.wait
72 self.want_cleanup: bool = ns.cleanup
73 self.want_rerun: bool = ns.rerun
74 self.want_run_leaks: bool = ns.runleaks
75
76 self.ci_mode: bool = (ns.fast_ci or ns.slow_ci)
77 self.want_add_python_opts: bool = (_add_python_opts
78 and ns._add_python_opts)
79
80 # Select tests
81 self.match_tests: TestFilter = ns.match_tests
82 self.exclude: bool = ns.exclude
83 self.fromfile: StrPath | None = ns.fromfile
84 self.starting_test: TestName | None = ns.start
85 self.cmdline_args: TestList = ns.args
86
87 # Workers
88 if ns.use_mp is None:
89 num_workers = 0 # run sequentially
90 elif ns.use_mp <= 0:
91 num_workers = -1 # use the number of CPUs
92 else:
93 num_workers = ns.use_mp
94 self.num_workers: int = num_workers
95 self.worker_json: StrJSON | None = ns.worker_json
96
97 # Options to run tests
98 self.fail_fast: bool = ns.failfast
99 self.fail_env_changed: bool = ns.fail_env_changed
100 self.fail_rerun: bool = ns.fail_rerun
101 self.forever: bool = ns.forever
102 self.output_on_failure: bool = ns.verbose3
103 self.timeout: float | None = ns.timeout
104 if ns.huntrleaks:
105 warmups, runs, filename = ns.huntrleaks
106 filename = os.path.abspath(filename)
107 self.hunt_refleak: HuntRefleak | None = HuntRefleak(warmups, runs, filename)
108 else:
109 self.hunt_refleak = None
110 self.test_dir: StrPath | None = ns.testdir
111 self.junit_filename: StrPath | None = ns.xmlpath
112 self.memory_limit: str | None = ns.memlimit
113 self.gc_threshold: int | None = ns.threshold
114 self.use_resources: tuple[str, ...] = tuple(ns.use_resources)
115 if ns.python:
116 self.python_cmd: tuple[str, ...] | None = tuple(ns.python)
117 else:
118 self.python_cmd = None
119 self.coverage: bool = ns.trace
120 self.coverage_dir: StrPath | None = ns.coverdir
121 self.tmp_dir: StrPath | None = ns.tempdir
122
123 # Randomize
124 self.randomize: bool = ns.randomize
125 if ('SOURCE_DATE_EPOCH' in os.environ
126 # don't use the variable if empty
127 and os.environ['SOURCE_DATE_EPOCH']
128 ):
129 self.randomize = False
130 # SOURCE_DATE_EPOCH should be an integer, but use a string to not
131 # fail if it's not integer. random.seed() accepts a string.
132 # https://reproducible-builds.org/docs/source-date-epoch/
133 self.random_seed: int | str = os.environ['SOURCE_DATE_EPOCH']
134 elif ns.random_seed is None:
135 self.random_seed = random.getrandbits(32)
136 else:
137 self.random_seed = ns.random_seed
138
139 # tests
140 self.first_runtests: RunTests | None = None
141
142 # used by --slowest
143 self.print_slowest: bool = ns.print_slow
144
145 # used to display the progress bar "[ 3/100]"
146 self.start_time = time.perf_counter()
147
148 # used by --single
149 self.single_test_run: bool = ns.single
150 self.next_single_test: TestName | None = None
151 self.next_single_filename: StrPath | None = None
152
153 def log(self, line=''):
154 self.logger.log(line)
155
156 def find_tests(self, tests: TestList | None = None) -> tuple[TestTuple, TestList | None]:
157 if self.single_test_run:
158 self.next_single_filename = os.path.join(self.tmp_dir, 'pynexttest')
159 try:
160 with open(self.next_single_filename, 'r') as fp:
161 next_test = fp.read().strip()
162 tests = [next_test]
163 except OSError:
164 pass
165
166 if self.fromfile:
167 tests = []
168 # regex to match 'test_builtin' in line:
169 # '0:00:00 [ 4/400] test_builtin -- test_dict took 1 sec'
170 regex = re.compile(r'\btest_[a-zA-Z0-9_]+\b')
171 with open(os.path.join(os_helper.SAVEDCWD, self.fromfile)) as fp:
172 for line in fp:
173 line = line.split('#', 1)[0]
174 line = line.strip()
175 match = regex.search(line)
176 if match is not None:
177 tests.append(match.group())
178
179 strip_py_suffix(tests)
180
181 if self.pgo:
182 # add default PGO tests if no tests are specified
183 setup_pgo_tests(self.cmdline_args, self.pgo_extended)
184
185 exclude_tests = set()
186 if self.exclude:
187 for arg in self.cmdline_args:
188 exclude_tests.add(arg)
189 self.cmdline_args = []
190
191 alltests = findtests(testdir=self.test_dir,
192 exclude=exclude_tests)
193
194 if not self.fromfile:
195 selected = tests or self.cmdline_args
196 if selected:
197 selected = split_test_packages(selected)
198 else:
199 selected = alltests
200 else:
201 selected = tests
202
203 if self.single_test_run:
204 selected = selected[:1]
205 try:
206 pos = alltests.index(selected[0])
207 self.next_single_test = alltests[pos + 1]
208 except IndexError:
209 pass
210
211 # Remove all the selected tests that precede start if it's set.
212 if self.starting_test:
213 try:
214 del selected[:selected.index(self.starting_test)]
215 except ValueError:
216 print(f"Cannot find starting test: {self.starting_test}")
217 sys.exit(1)
218
219 random.seed(self.random_seed)
220 if self.randomize:
221 random.shuffle(selected)
222
223 return (tuple(selected), tests)
224
225 @staticmethod
226 def list_tests(tests: TestTuple):
227 for name in tests:
228 print(name)
229
230 def _rerun_failed_tests(self, runtests: RunTests):
231 # Configure the runner to re-run tests
232 if self.num_workers == 0:
233 # Always run tests in fresh processes to have more deterministic
234 # initial state. Don't re-run tests in parallel but limit to a
235 # single worker process to have side effects (on the system load
236 # and timings) between tests.
237 self.num_workers = 1
238
239 tests, match_tests_dict = self.results.prepare_rerun()
240
241 # Re-run failed tests
242 self.log(f"Re-running {len(tests)} failed tests in verbose mode in subprocesses")
243 runtests = runtests.copy(
244 tests=tests,
245 rerun=True,
246 verbose=True,
247 forever=False,
248 fail_fast=False,
249 match_tests_dict=match_tests_dict,
250 output_on_failure=False)
251 self.logger.set_tests(runtests)
252 self._run_tests_mp(runtests, self.num_workers)
253 return runtests
254
255 def rerun_failed_tests(self, runtests: RunTests):
256 if self.python_cmd:
257 # Temp patch for https://github.com/python/cpython/issues/94052
258 self.log(
259 "Re-running failed tests is not supported with --python "
260 "host runner option."
261 )
262 return
263
264 self.first_state = self.get_state()
265
266 print()
267 rerun_runtests = self._rerun_failed_tests(runtests)
268
269 if self.results.bad:
270 print(count(len(self.results.bad), 'test'), "failed again:")
271 printlist(self.results.bad)
272
273 self.display_result(rerun_runtests)
274
275 def display_result(self, runtests):
276 # If running the test suite for PGO then no one cares about results.
277 if runtests.pgo:
278 return
279
280 state = self.get_state()
281 print()
282 print(f"== Tests result: {state} ==")
283
284 self.results.display_result(runtests.tests,
285 self.quiet, self.print_slowest)
286
287 def run_test(self, test_name: TestName, runtests: RunTests, tracer):
288 if tracer is not None:
289 # If we're tracing code coverage, then we don't exit with status
290 # if on a false return value from main.
291 cmd = ('result = run_single_test(test_name, runtests)')
292 namespace = dict(locals())
293 tracer.runctx(cmd, globals=globals(), locals=namespace)
294 result = namespace['result']
295 else:
296 result = run_single_test(test_name, runtests)
297
298 self.results.accumulate_result(result, runtests)
299
300 return result
301
302 def run_tests_sequentially(self, runtests):
303 if self.coverage:
304 import trace
305 tracer = trace.Trace(trace=False, count=True)
306 else:
307 tracer = None
308
309 save_modules = set(sys.modules)
310
311 jobs = runtests.get_jobs()
312 if jobs is not None:
313 tests = count(jobs, 'test')
314 else:
315 tests = 'tests'
316 msg = f"Run {tests} sequentially"
317 if runtests.timeout:
318 msg += " (timeout: %s)" % format_duration(runtests.timeout)
319 self.log(msg)
320
321 previous_test = None
322 tests_iter = runtests.iter_tests()
323 for test_index, test_name in enumerate(tests_iter, 1):
324 start_time = time.perf_counter()
325
326 text = test_name
327 if previous_test:
328 text = '%s -- %s' % (text, previous_test)
329 self.logger.display_progress(test_index, text)
330
331 result = self.run_test(test_name, runtests, tracer)
332
333 # Unload the newly imported test modules (best effort finalization)
334 new_modules = [module for module in sys.modules
335 if module not in save_modules and
336 module.startswith(("test.", "test_"))]
337 for module in new_modules:
338 sys.modules.pop(module, None)
339 # Remove the attribute of the parent module.
340 parent, _, name = module.rpartition('.')
341 try:
342 delattr(sys.modules[parent], name)
343 except (KeyError, AttributeError):
344 pass
345
346 if result.must_stop(self.fail_fast, self.fail_env_changed):
347 break
348
349 previous_test = str(result)
350 test_time = time.perf_counter() - start_time
351 if test_time >= PROGRESS_MIN_TIME:
352 previous_test = "%s in %s" % (previous_test, format_duration(test_time))
353 elif result.state == State.PASSED:
354 # be quiet: say nothing if the test passed shortly
355 previous_test = None
356
357 if previous_test:
358 print(previous_test)
359
360 return tracer
361
362 def get_state(self):
363 state = self.results.get_state(self.fail_env_changed)
364 if self.first_state:
365 state = f'{self.first_state} then {state}'
366 return state
367
368 def _run_tests_mp(self, runtests: RunTests, num_workers: int) -> None:
369 from .run_workers import RunWorkers
370 RunWorkers(num_workers, runtests, self.logger, self.results).run()
371
372 def finalize_tests(self, tracer):
373 if self.next_single_filename:
374 if self.next_single_test:
375 with open(self.next_single_filename, 'w') as fp:
376 fp.write(self.next_single_test + '\n')
377 else:
378 os.unlink(self.next_single_filename)
379
380 if tracer is not None:
381 results = tracer.results()
382 results.write_results(show_missing=True, summary=True,
383 coverdir=self.coverage_dir)
384
385 if self.want_run_leaks:
386 os.system("leaks %d" % os.getpid())
387
388 if self.junit_filename:
389 self.results.write_junit(self.junit_filename)
390
391 def display_summary(self):
392 duration = time.perf_counter() - self.logger.start_time
393 filtered = bool(self.match_tests)
394
395 # Total duration
396 print()
397 print("Total duration: %s" % format_duration(duration))
398
399 self.results.display_summary(self.first_runtests, filtered)
400
401 # Result
402 state = self.get_state()
403 print(f"Result: {state}")
404
405 def create_run_tests(self, tests: TestTuple):
406 return RunTests(
407 tests,
408 fail_fast=self.fail_fast,
409 fail_env_changed=self.fail_env_changed,
410 match_tests=self.match_tests,
411 match_tests_dict=None,
412 rerun=False,
413 forever=self.forever,
414 pgo=self.pgo,
415 pgo_extended=self.pgo_extended,
416 output_on_failure=self.output_on_failure,
417 timeout=self.timeout,
418 verbose=self.verbose,
419 quiet=self.quiet,
420 hunt_refleak=self.hunt_refleak,
421 test_dir=self.test_dir,
422 use_junit=(self.junit_filename is not None),
423 memory_limit=self.memory_limit,
424 gc_threshold=self.gc_threshold,
425 use_resources=self.use_resources,
426 python_cmd=self.python_cmd,
427 randomize=self.randomize,
428 random_seed=self.random_seed,
429 )
430
431 def _run_tests(self, selected: TestTuple, tests: TestList | None) -> int:
432 if self.hunt_refleak and self.hunt_refleak.warmups < 3:
433 msg = ("WARNING: Running tests with --huntrleaks/-R and "
434 "less than 3 warmup repetitions can give false positives!")
435 print(msg, file=sys.stdout, flush=True)
436
437 if self.num_workers < 0:
438 # Use all CPUs + 2 extra worker processes for tests
439 # that like to sleep
440 self.num_workers = (process_cpu_count() or 1) + 2
441
442 # For a partial run, we do not need to clutter the output.
443 if (self.want_header
444 or not(self.pgo or self.quiet or self.single_test_run
445 or tests or self.cmdline_args)):
446 display_header(self.use_resources, self.python_cmd)
447
448 print("Using random seed:", self.random_seed)
449
450 runtests = self.create_run_tests(selected)
451 self.first_runtests = runtests
452 self.logger.set_tests(runtests)
453
454 setup_process()
455
456 if self.hunt_refleak and not self.num_workers:
457 # gh-109739: WindowsLoadTracker thread interfers with refleak check
458 use_load_tracker = False
459 else:
460 # WindowsLoadTracker is only needed on Windows
461 use_load_tracker = MS_WINDOWS
462
463 if use_load_tracker:
464 self.logger.start_load_tracker()
465 try:
466 if self.num_workers:
467 self._run_tests_mp(runtests, self.num_workers)
468 tracer = None
469 else:
470 tracer = self.run_tests_sequentially(runtests)
471
472 self.display_result(runtests)
473
474 if self.want_rerun and self.results.need_rerun():
475 self.rerun_failed_tests(runtests)
476 finally:
477 if use_load_tracker:
478 self.logger.stop_load_tracker()
479
480 self.display_summary()
481 self.finalize_tests(tracer)
482
483 return self.results.get_exitcode(self.fail_env_changed,
484 self.fail_rerun)
485
486 def run_tests(self, selected: TestTuple, tests: TestList | None) -> int:
487 os.makedirs(self.tmp_dir, exist_ok=True)
488 work_dir = get_work_dir(self.tmp_dir)
489
490 # Put a timeout on Python exit
491 with exit_timeout():
492 # Run the tests in a context manager that temporarily changes the
493 # CWD to a temporary and writable directory. If it's not possible
494 # to create or change the CWD, the original CWD will be used.
495 # The original CWD is available from os_helper.SAVEDCWD.
496 with os_helper.temp_cwd(work_dir, quiet=True):
497 # When using multiprocessing, worker processes will use
498 # work_dir as their parent temporary directory. So when the
499 # main process exit, it removes also subdirectories of worker
500 # processes.
501 return self._run_tests(selected, tests)
502
503 def _add_cross_compile_opts(self, regrtest_opts):
504 # WASM/WASI buildbot builders pass multiple PYTHON environment
505 # variables such as PYTHONPATH and _PYTHON_HOSTRUNNER.
506 keep_environ = bool(self.python_cmd)
507 environ = None
508
509 # Are we using cross-compilation?
510 cross_compile = is_cross_compiled()
511
512 # Get HOSTRUNNER
513 hostrunner = get_host_runner()
514
515 if cross_compile:
516 # emulate -E, but keep PYTHONPATH + cross compile env vars,
517 # so test executable can load correct sysconfigdata file.
518 keep = {
519 '_PYTHON_PROJECT_BASE',
520 '_PYTHON_HOST_PLATFORM',
521 '_PYTHON_SYSCONFIGDATA_NAME',
522 'PYTHONPATH'
523 }
524 old_environ = os.environ
525 new_environ = {
526 name: value for name, value in os.environ.items()
527 if not name.startswith(('PYTHON', '_PYTHON')) or name in keep
528 }
529 # Only set environ if at least one variable was removed
530 if new_environ != old_environ:
531 environ = new_environ
532 keep_environ = True
533
534 if cross_compile and hostrunner:
535 if self.num_workers == 0:
536 # For now use only two cores for cross-compiled builds;
537 # hostrunner can be expensive.
538 regrtest_opts.extend(['-j', '2'])
539
540 # If HOSTRUNNER is set and -p/--python option is not given, then
541 # use hostrunner to execute python binary for tests.
542 if not self.python_cmd:
543 buildpython = sysconfig.get_config_var("BUILDPYTHON")
544 python_cmd = f"{hostrunner} {buildpython}"
545 regrtest_opts.extend(["--python", python_cmd])
546 keep_environ = True
547
548 return (environ, keep_environ)
549
550 def _add_ci_python_opts(self, python_opts, keep_environ):
551 # --fast-ci and --slow-ci add options to Python:
552 # "-u -W default -bb -E"
553
554 # Unbuffered stdout and stderr
555 if not sys.stdout.write_through:
556 python_opts.append('-u')
557
558 # Add warnings filter 'default'
559 if 'default' not in sys.warnoptions:
560 python_opts.extend(('-W', 'default'))
561
562 # Error on bytes/str comparison
563 if sys.flags.bytes_warning < 2:
564 python_opts.append('-bb')
565
566 if not keep_environ:
567 # Ignore PYTHON* environment variables
568 if not sys.flags.ignore_environment:
569 python_opts.append('-E')
570
571 def _execute_python(self, cmd, environ):
572 # Make sure that messages before execv() are logged
573 sys.stdout.flush()
574 sys.stderr.flush()
575
576 cmd_text = shlex.join(cmd)
577 try:
578 print(f"+ {cmd_text}", flush=True)
579
580 if hasattr(os, 'execv') and not MS_WINDOWS:
581 os.execv(cmd[0], cmd)
582 # On success, execv() do no return.
583 # On error, it raises an OSError.
584 else:
585 import subprocess
586 with subprocess.Popen(cmd, env=environ) as proc:
587 try:
588 proc.wait()
589 except KeyboardInterrupt:
590 # There is no need to call proc.terminate(): on CTRL+C,
591 # SIGTERM is also sent to the child process.
592 try:
593 proc.wait(timeout=EXIT_TIMEOUT)
594 except subprocess.TimeoutExpired:
595 proc.kill()
596 proc.wait()
597 sys.exit(EXITCODE_INTERRUPTED)
598
599 sys.exit(proc.returncode)
600 except Exception as exc:
601 print_warning(f"Failed to change Python options: {exc!r}\n"
602 f"Command: {cmd_text}")
603 # continue executing main()
604
605 def _add_python_opts(self):
606 python_opts = []
607 regrtest_opts = []
608
609 environ, keep_environ = self._add_cross_compile_opts(regrtest_opts)
610 if self.ci_mode:
611 self._add_ci_python_opts(python_opts, keep_environ)
612
613 if (not python_opts) and (not regrtest_opts) and (environ is None):
614 # Nothing changed: nothing to do
615 return
616
617 # Create new command line
618 cmd = list(sys.orig_argv)
619 if python_opts:
620 cmd[1:1] = python_opts
621 if regrtest_opts:
622 cmd.extend(regrtest_opts)
623 cmd.append("--dont-add-python-opts")
624
625 self._execute_python(cmd, environ)
626
627 def _init(self):
628 # Set sys.stdout encoder error handler to backslashreplace,
629 # similar to sys.stderr error handler, to avoid UnicodeEncodeError
630 # when printing a traceback or any other non-encodable character.
631 sys.stdout.reconfigure(errors="backslashreplace")
632
633 if self.junit_filename and not os.path.isabs(self.junit_filename):
634 self.junit_filename = os.path.abspath(self.junit_filename)
635
636 strip_py_suffix(self.cmdline_args)
637
638 self.tmp_dir = get_temp_dir(self.tmp_dir)
639
640 def main(self, tests: TestList | None = None):
641 if self.want_add_python_opts:
642 self._add_python_opts()
643
644 self._init()
645
646 if self.want_cleanup:
647 cleanup_temp_dir(self.tmp_dir)
648 sys.exit(0)
649
650 if self.want_wait:
651 input("Press any key to continue...")
652
653 setup_test_dir(self.test_dir)
654 selected, tests = self.find_tests(tests)
655
656 exitcode = 0
657 if self.want_list_tests:
658 self.list_tests(selected)
659 elif self.want_list_cases:
660 list_cases(selected,
661 match_tests=self.match_tests,
662 test_dir=self.test_dir)
663 else:
664 exitcode = self.run_tests(selected, tests)
665
666 sys.exit(exitcode)
667
668
669 def main(tests=None, _add_python_opts=False, **kwargs):
670 """Run the Python suite."""
671 ns = _parse_args(sys.argv[1:], **kwargs)
672 Regrtest(ns, _add_python_opts=_add_python_opts).main(tests=tests)
Browse
Usage
Versions