text: import datetime
  import faulthandler
  import locale
  import os
  import platform
  import random
  import re
  import sys
  import sysconfig
  import tempfile
  import time
  import unittest
  from test.libregrtest.cmdline import _parse_args
  from test.libregrtest.runtest import (
      findtests, runtest, get_abs_module,
      STDTESTS, NOTTESTS, PASSED, FAILED, ENV_CHANGED, SKIPPED, RESOURCE_DENIED,
      INTERRUPTED, CHILD_ERROR,
      PROGRESS_MIN_TIME, format_test_result)
  from test.libregrtest.setup import setup_tests
  from test.libregrtest.utils import removepy, count, format_duration, printlist
  from test import support
  try:
      import gc
  except ImportError:
      gc = None
  
  
  # When tests are run from the Python build directory, it is best practice
  # to keep the test files in a subfolder.  This eases the cleanup of leftover
  # files using the "make distclean" command.
  if sysconfig.is_python_build():
      TEMPDIR = sysconfig.get_config_var('abs_builddir')
      if TEMPDIR is None:
          # bpo-30284: On Windows, only srcdir is available. Using abs_builddir
          # mostly matters on UNIX when building Python out of the source tree,
          # especially when the source tree is read only.
          TEMPDIR = sysconfig.get_config_var('srcdir')
      TEMPDIR = os.path.join(TEMPDIR, 'build')
  else:
      TEMPDIR = tempfile.gettempdir()
  TEMPDIR = os.path.abspath(TEMPDIR)
  
  
  class Regrtest:
      """Execute a test suite.
  
      This also parses command-line options and modifies its behavior
      accordingly.
  
      tests -- a list of strings containing test names (optional)
      testdir -- the directory in which to look for tests (optional)
  
      Users other than the Python test suite will certainly want to
      specify testdir; if it's omitted, the directory containing the
      Python test suite is searched for.
  
      If the tests argument is omitted, the tests listed on the
      command-line will be used.  If that's empty, too, then all *.py
      files beginning with test_ will be used.
  
      The other default arguments (verbose, quiet, exclude,
      single, randomize, findleaks, use_resources, trace, coverdir,
      print_slow, and random_seed) allow programmers calling main()
      directly to set the values that would normally be set by flags
      on the command line.
      """
      def __init__(self):
          # Namespace of command line options
          self.ns = None
  
          # tests
          self.tests = []
          self.selected = []
  
          # test results
          self.good = []
          self.bad = []
          self.skipped = []
          self.resource_denieds = []
          self.environment_changed = []
          self.rerun = []
          self.first_result = None
          self.interrupted = False
  
          # used by --slow
          self.test_times = []
  
          # used by --coverage, trace.Trace instance
          self.tracer = None
  
          # used by --findleaks, store for gc.garbage
          self.found_garbage = []
  
          # used to display the progress bar "[ 3/100]"
          self.start_time = time.monotonic()
          self.test_count = ''
          self.test_count_width = 1
  
          # used by --single
          self.next_single_test = None
          self.next_single_filename = None
  
      def accumulate_result(self, test, result):
          ok, test_time = result
          if ok not in (CHILD_ERROR, INTERRUPTED):
              self.test_times.append((test_time, test))
          if ok == PASSED:
              self.good.append(test)
          elif ok in (FAILED, CHILD_ERROR):
              self.bad.append(test)
          elif ok == ENV_CHANGED:
              self.environment_changed.append(test)
          elif ok == SKIPPED:
              self.skipped.append(test)
          elif ok == RESOURCE_DENIED:
              self.skipped.append(test)
              self.resource_denieds.append(test)
          elif ok != INTERRUPTED:
              raise ValueError("invalid test result: %r" % ok)
  
      def display_progress(self, test_index, test):
          if self.ns.quiet:
              return
  
          # "[ 51/405/1] test_tcl passed"
          line = f"{test_index:{self.test_count_width}}{self.test_count}"
          fails = len(self.bad) + len(self.environment_changed)
          if fails and not self.ns.pgo:
              line = f"{line}/{fails}"
          line = f"[{line}] {test}"
  
          # add the system load prefix: "load avg: 1.80 "
          if hasattr(os, 'getloadavg'):
              load_avg_1min = os.getloadavg()[0]
              line = f"load avg: {load_avg_1min:.2f} {line}"
  
          # add the timestamp prefix:  "0:01:05 "
          test_time = time.monotonic() - self.start_time
          test_time = datetime.timedelta(seconds=int(test_time))
          line = f"{test_time} {line}"
          print(line, flush=True)
  
      def parse_args(self, kwargs):
          ns = _parse_args(sys.argv[1:], **kwargs)
  
          if ns.timeout and not hasattr(faulthandler, 'dump_traceback_later'):
              print("Warning: The timeout option requires "
                    "faulthandler.dump_traceback_later", file=sys.stderr)
              ns.timeout = None
  
          if ns.threshold is not None and gc is None:
              print('No GC available, ignore --threshold.', file=sys.stderr)
              ns.threshold = None
  
          if ns.findleaks:
              if gc is not None:
                  # Uncomment the line below to report garbage that is not
                  # freeable by reference counting alone.  By default only
                  # garbage that is not collectable by the GC is reported.
                  pass
                  #gc.set_debug(gc.DEBUG_SAVEALL)
              else:
                  print('No GC available, disabling --findleaks',
                        file=sys.stderr)
                  ns.findleaks = False
  
          # Strip .py extensions.
          removepy(ns.args)
  
          return ns
  
      def find_tests(self, tests):
          self.tests = tests
  
          if self.ns.single:
              self.next_single_filename = os.path.join(TEMPDIR, 'pynexttest')
              try:
                  with open(self.next_single_filename, 'r') as fp:
                      next_test = fp.read().strip()
                      self.tests = [next_test]
              except OSError:
                  pass
  
          if self.ns.fromfile:
              self.tests = []
              # regex to match 'test_builtin' in line:
              # '0:00:00 [  4/400] test_builtin -- test_dict took 1 sec'
              regex = re.compile(r'\btest_[a-zA-Z0-9_]+\b')
              with open(os.path.join(support.SAVEDCWD, self.ns.fromfile)) as fp:
                  for line in fp:
                      line = line.split('#', 1)[0]
                      line = line.strip()
                      match = regex.search(line)
                      if match is not None:
                          self.tests.append(match.group())
  
          removepy(self.tests)
  
          stdtests = STDTESTS[:]
          nottests = NOTTESTS.copy()
          if self.ns.exclude:
              for arg in self.ns.args:
                  if arg in stdtests:
                      stdtests.remove(arg)
                  nottests.add(arg)
              self.ns.args = []
  
          # if testdir is set, then we are not running the python tests suite, so
          # don't add default tests to be executed or skipped (pass empty values)
          if self.ns.testdir:
              alltests = findtests(self.ns.testdir, list(), set())
          else:
              alltests = findtests(self.ns.testdir, stdtests, nottests)
  
          if not self.ns.fromfile:
              self.selected = self.tests or self.ns.args or alltests
          else:
              self.selected = self.tests
          if self.ns.single:
              self.selected = self.selected[:1]
              try:
                  pos = alltests.index(self.selected[0])
                  self.next_single_test = alltests[pos + 1]
              except IndexError:
                  pass
  
          # Remove all the selected tests that precede start if it's set.
          if self.ns.start:
              try:
                  del self.selected[:self.selected.index(self.ns.start)]
              except ValueError:
                  print("Couldn't find starting test (%s), using all tests"
                        % self.ns.start, file=sys.stderr)
  
          if self.ns.randomize:
              if self.ns.random_seed is None:
                  self.ns.random_seed = random.randrange(10000000)
              random.seed(self.ns.random_seed)
              random.shuffle(self.selected)
  
      def list_tests(self):
          for name in self.selected:
              print(name)
  
      def _list_cases(self, suite):
          for test in suite:
              if isinstance(test, unittest.loader._FailedTest):
                  continue
              if isinstance(test, unittest.TestSuite):
                  self._list_cases(test)
              elif isinstance(test, unittest.TestCase):
                  if support.match_test(test):
                      print(test.id())
  
      def list_cases(self):
          support.verbose = False
          support.set_match_tests(self.ns.match_tests)
  
          for test in self.selected:
              abstest = get_abs_module(self.ns, test)
              try:
                  suite = unittest.defaultTestLoader.loadTestsFromName(abstest)
                  self._list_cases(suite)
              except unittest.SkipTest:
                  self.skipped.append(test)
  
          if self.skipped:
              print(file=sys.stderr)
              print(count(len(self.skipped), "test"), "skipped:", file=sys.stderr)
              printlist(self.skipped, file=sys.stderr)
  
      def rerun_failed_tests(self):
          self.ns.verbose = True
          self.ns.failfast = False
          self.ns.verbose3 = False
  
          self.first_result = self.get_tests_result()
  
          print()
          print("Re-running failed tests in verbose mode")
          self.rerun = self.bad[:]
          for test in self.rerun:
              print("Re-running test %r in verbose mode" % test, flush=True)
              try:
                  self.ns.verbose = True
                  ok = runtest(self.ns, test)
              except KeyboardInterrupt:
                  self.interrupted = True
                  # print a newline separate from the ^C
                  print()
                  break
              else:
                  if ok[0] in {PASSED, ENV_CHANGED, SKIPPED, RESOURCE_DENIED}:
                      self.bad.remove(test)
          else:
              if self.bad:
                  print(count(len(self.bad), 'test'), "failed again:")
                  printlist(self.bad)
  
          self.display_result()
  
      def display_result(self):
          # If running the test suite for PGO then no one cares about results.
          if self.ns.pgo:
              return
  
          print()
          print("== Tests result: %s ==" % self.get_tests_result())
  
          if self.interrupted:
              print()
              # print a newline after ^C
              print("Test suite interrupted by signal SIGINT.")
              executed = set(self.good) | set(self.bad) | set(self.skipped)
              omitted = set(self.selected) - executed
              print(count(len(omitted), "test"), "omitted:")
              printlist(omitted)
  
          if self.good and not self.ns.quiet:
              print()
              if (not self.bad
                  and not self.skipped
                  and not self.interrupted
                  and len(self.good) > 1):
                  print("All", end=' ')
              print(count(len(self.good), "test"), "OK.")
  
          if self.ns.print_slow:
              self.test_times.sort(reverse=True)
              print()
              print("10 slowest tests:")
              for time, test in self.test_times[:10]:
                  print("- %s: %s" % (test, format_duration(time)))
  
          if self.bad:
              print()
              print(count(len(self.bad), "test"), "failed:")
              printlist(self.bad)
  
          if self.environment_changed:
              print()
              print("{} altered the execution environment:".format(
                       count(len(self.environment_changed), "test")))
              printlist(self.environment_changed)
  
          if self.skipped and not self.ns.quiet:
              print()
              print(count(len(self.skipped), "test"), "skipped:")
              printlist(self.skipped)
  
          if self.rerun:
              print()
              print("%s:" % count(len(self.rerun), "re-run test"))
              printlist(self.rerun)
  
      def run_tests_sequential(self):
          if self.ns.trace:
              import trace
              self.tracer = trace.Trace(trace=False, count=True)
  
          save_modules = sys.modules.keys()
  
          print("Run tests sequentially")
  
          previous_test = None
          for test_index, test in enumerate(self.tests, 1):
              start_time = time.monotonic()
  
              text = test
              if previous_test:
                  text = '%s -- %s' % (text, previous_test)
              self.display_progress(test_index, text)
  
              if self.tracer:
                  # If we're tracing code coverage, then we don't exit with status
                  # if on a false return value from main.
                  cmd = ('result = runtest(self.ns, test); '
                         'self.accumulate_result(test, result)')
                  ns = dict(locals())
                  self.tracer.runctx(cmd, globals=globals(), locals=ns)
                  result = ns['result']
              else:
                  try:
                      result = runtest(self.ns, test)
                  except KeyboardInterrupt:
                      self.interrupted = True
                      self.accumulate_result(test, (INTERRUPTED, None))
                      break
                  else:
                      self.accumulate_result(test, result)
  
              previous_test = format_test_result(test, result[0])
              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:
                  # be quiet: say nothing if the test passed shortly
                  previous_test = None
  
              if self.ns.findleaks:
                  gc.collect()
                  if gc.garbage:
                      print("Warning: test created", len(gc.garbage), end=' ')
                      print("uncollectable object(s).")
                      # move the uncollectable objects somewhere so we don't see
                      # them again
                      self.found_garbage.extend(gc.garbage)
                      del gc.garbage[:]
  
              # Unload the newly imported modules (best effort finalization)
              for module in sys.modules.keys():
                  if module not in save_modules and module.startswith("test."):
                      support.unload(module)
  
          if previous_test:
              print(previous_test)
  
      def _test_forever(self, tests):
          while True:
              for test in tests:
                  yield test
                  if self.bad:
                      return
                  if self.ns.fail_env_changed and self.environment_changed:
                      return
  
      def display_header(self):
          # Print basic platform information
          print("==", platform.python_implementation(), *sys.version.split())
          print("==", platform.platform(aliased=True),
                        "%s-endian" % sys.byteorder)
          print("== cwd:", os.getcwd())
          cpu_count = os.cpu_count()
          if cpu_count:
              print("== CPU count:", cpu_count)
          print("== encodings: locale=%s, FS=%s"
                % (locale.getpreferredencoding(False),
                   sys.getfilesystemencoding()))
  
      def get_tests_result(self):
          result = []
          if self.bad:
              result.append("FAILURE")
          elif self.ns.fail_env_changed and self.environment_changed:
              result.append("ENV CHANGED")
  
          if self.interrupted:
              result.append("INTERRUPTED")
  
          if not result:
              result.append("SUCCESS")
  
          result = ', '.join(result)
          if self.first_result:
              result = '%s then %s' % (self.first_result, result)
          return result
  
      def run_tests(self):
          # For a partial run, we do not need to clutter the output.
          if (self.ns.header
              or not(self.ns.pgo or self.ns.quiet or self.ns.single
                     or self.tests or self.ns.args)):
              self.display_header()
  
          if self.ns.huntrleaks:
              warmup, repetitions, _ = self.ns.huntrleaks
              if warmup < 3:
                  msg = ("WARNING: Running tests with --huntrleaks/-R and less than "
                          "3 warmup repetitions can give false positives!")
                  print(msg, file=sys.stdout, flush=True)
  
          if self.ns.randomize:
              print("Using random seed", self.ns.random_seed)
  
          if self.ns.forever:
              self.tests = self._test_forever(list(self.selected))
              self.test_count = ''
              self.test_count_width = 3
          else:
              self.tests = iter(self.selected)
              self.test_count = '/{}'.format(len(self.selected))
              self.test_count_width = len(self.test_count) - 1
  
          if self.ns.use_mp:
              from test.libregrtest.runtest_mp import run_tests_multiprocess
              run_tests_multiprocess(self)
          else:
              self.run_tests_sequential()
  
      def finalize(self):
          if self.next_single_filename:
              if self.next_single_test:
                  with open(self.next_single_filename, 'w') as fp:
                      fp.write(self.next_single_test + '\n')
              else:
                  os.unlink(self.next_single_filename)
  
          if self.tracer:
              r = self.tracer.results()
              r.write_results(show_missing=True, summary=True,
                              coverdir=self.ns.coverdir)
  
          print()
          duration = time.monotonic() - self.start_time
          print("Total duration: %s" % format_duration(duration))
          print("Tests result: %s" % self.get_tests_result())
  
          if self.ns.runleaks:
              os.system("leaks %d" % os.getpid())
  
      def main(self, tests=None, **kwargs):
          global TEMPDIR
  
          if sysconfig.is_python_build():
              try:
                  os.mkdir(TEMPDIR)
              except FileExistsError:
                  pass
  
          # Define a writable temp dir that will be used as cwd while running
          # the tests. The name of the dir includes the pid to allow parallel
          # testing (see the -j option).
          test_cwd = 'test_python_{}'.format(os.getpid())
          test_cwd = os.path.join(TEMPDIR, test_cwd)
  
          # Run the tests in a context manager that temporarily changes the CWD to a
          # temporary and writable directory.  If it's not possible to create or
          # change the CWD, the original CWD will be used.  The original CWD is
          # available from support.SAVEDCWD.
          with support.temp_cwd(test_cwd, quiet=True):
              self._main(tests, kwargs)
  
      def _main(self, tests, kwargs):
          self.ns = self.parse_args(kwargs)
  
          if self.ns.slaveargs is not None:
              from test.libregrtest.runtest_mp import run_tests_slave
              run_tests_slave(self.ns.slaveargs)
  
          if self.ns.wait:
              input("Press any key to continue...")
  
          support.PGO = self.ns.pgo
  
          setup_tests(self.ns)
  
          self.find_tests(tests)
  
          if self.ns.list_tests:
              self.list_tests()
              sys.exit(0)
  
          if self.ns.list_cases:
              self.list_cases()
              sys.exit(0)
  
          self.run_tests()
          self.display_result()
  
          if self.ns.verbose2 and self.bad:
              self.rerun_failed_tests()
  
          self.finalize()
          if self.bad:
              sys.exit(2)
          if self.interrupted:
              sys.exit(130)
          if self.ns.fail_env_changed and self.environment_changed:
              sys.exit(3)
          sys.exit(0)
  
  
  def main(tests=None, **kwargs):
      """Run the Python suite."""
      Regrtest().main(tests=tests, **kwargs)
  