Issue #25220: Move most regrtest.py code to libregrtest

Lib/test/libregrtest/__init__.py

 from test.libregrtest.cmdline import _parse_args, RESOURCE_NAMES
+from test.libregrtest.main import main_in_temp_cwd

Lib/test/libregrtest/cmdline.py

 import argparse
 import faulthandler
 import os
-
 from test import support
 
+
 USAGE = """\
 python -m test [options] [test_name1 [test_name2 ...]]
 python path/to/Lib/test/regrtest.py [options] [test_name1 [test_name2 ...]]

Lib/test/libregrtest/refleak.py

+import os
+import re
+import sys
+import warnings
+from inspect import isabstract
+from test import support
+
+
+def dash_R(the_module, test, indirect_test, huntrleaks):
+    """Run a test multiple times, looking for reference leaks.
+
+    Returns:
+        False if the test didn't leak references; True if we detected refleaks.
+    """
+    # This code is hackish and inelegant, but it seems to do the job.
+    import copyreg
+    import collections.abc
+
+    if not hasattr(sys, 'gettotalrefcount'):
+        raise Exception("Tracking reference leaks requires a debug build "
+                        "of Python")
+
+    # Save current values for dash_R_cleanup() to restore.
+    fs = warnings.filters[:]
+    ps = copyreg.dispatch_table.copy()
+    pic = sys.path_importer_cache.copy()
+    try:
+        import zipimport
+    except ImportError:
+        zdc = None # Run unmodified on platforms without zipimport support
+    else:
+        zdc = zipimport._zip_directory_cache.copy()
+    abcs = {}
+    for abc in [getattr(collections.abc, a) for a in collections.abc.__all__]:
+        if not isabstract(abc):
+            continue
+        for obj in abc.__subclasses__() + [abc]:
+            abcs[obj] = obj._abc_registry.copy()
+
+    nwarmup, ntracked, fname = huntrleaks
+    fname = os.path.join(support.SAVEDCWD, fname)
+    repcount = nwarmup + ntracked
+    rc_deltas = [0] * repcount
+    alloc_deltas = [0] * repcount
+
+    print("beginning", repcount, "repetitions", file=sys.stderr)
+    print(("1234567890"*(repcount//10 + 1))[:repcount], file=sys.stderr)
+    sys.stderr.flush()
+    for i in range(repcount):
+        indirect_test()
+        alloc_after, rc_after = dash_R_cleanup(fs, ps, pic, zdc, abcs)
+        sys.stderr.write('.')
+        sys.stderr.flush()
+        if i >= nwarmup:
+            rc_deltas[i] = rc_after - rc_before
+            alloc_deltas[i] = alloc_after - alloc_before
+        alloc_before, rc_before = alloc_after, rc_after
+    print(file=sys.stderr)
+    # These checkers return False on success, True on failure
+    def check_rc_deltas(deltas):
+        return any(deltas)
+    def check_alloc_deltas(deltas):
+        # At least 1/3rd of 0s
+        if 3 * deltas.count(0) < len(deltas):
+            return True
+        # Nothing else than 1s, 0s and -1s
+        if not set(deltas) <= {1,0,-1}:
+            return True
+        return False
+    failed = False
+    for deltas, item_name, checker in [
+        (rc_deltas, 'references', check_rc_deltas),
+        (alloc_deltas, 'memory blocks', check_alloc_deltas)]:
+        if checker(deltas):
+            msg = '%s leaked %s %s, sum=%s' % (
+                test, deltas[nwarmup:], item_name, sum(deltas))
+            print(msg, file=sys.stderr)
+            sys.stderr.flush()
+            with open(fname, "a") as refrep:
+                print(msg, file=refrep)
+                refrep.flush()
+            failed = True
+    return failed
+
+
+def dash_R_cleanup(fs, ps, pic, zdc, abcs):
+    import gc, copyreg
+    import _strptime, linecache
+    import urllib.parse, urllib.request, mimetypes, doctest
+    import struct, filecmp, collections.abc
+    from distutils.dir_util import _path_created
+    from weakref import WeakSet
+
+    # Clear the warnings registry, so they can be displayed again
+    for mod in sys.modules.values():
+        if hasattr(mod, '__warningregistry__'):
+            del mod.__warningregistry__
+
+    # Restore some original values.
+    warnings.filters[:] = fs
+    copyreg.dispatch_table.clear()
+    copyreg.dispatch_table.update(ps)
+    sys.path_importer_cache.clear()
+    sys.path_importer_cache.update(pic)
+    try:
+        import zipimport
+    except ImportError:
+        pass # Run unmodified on platforms without zipimport support
+    else:
+        zipimport._zip_directory_cache.clear()
+        zipimport._zip_directory_cache.update(zdc)
+
+    # clear type cache
+    sys._clear_type_cache()
+
+    # Clear ABC registries, restoring previously saved ABC registries.
+    for abc in [getattr(collections.abc, a) for a in collections.abc.__all__]:
+        if not isabstract(abc):
+            continue
+        for obj in abc.__subclasses__() + [abc]:
+            obj._abc_registry = abcs.get(obj, WeakSet()).copy()
+            obj._abc_cache.clear()
+            obj._abc_negative_cache.clear()
+
+    # Flush standard output, so that buffered data is sent to the OS and
+    # associated Python objects are reclaimed.
+    for stream in (sys.stdout, sys.stderr, sys.__stdout__, sys.__stderr__):
+        if stream is not None:
+            stream.flush()
+
+    # Clear assorted module caches.
+    _path_created.clear()
+    re.purge()
+    _strptime._regex_cache.clear()
+    urllib.parse.clear_cache()
+    urllib.request.urlcleanup()
+    linecache.clearcache()
+    mimetypes._default_mime_types()
+    filecmp._cache.clear()
+    struct._clearcache()
+    doctest.master = None
+    try:
+        import ctypes
+    except ImportError:
+        # Don't worry about resetting the cache if ctypes is not supported
+        pass
+    else:
+        ctypes._reset_cache()
+
+    # Collect cyclic trash and read memory statistics immediately after.
+    func1 = sys.getallocatedblocks
+    func2 = sys.gettotalrefcount
+    gc.collect()
+    return func1(), func2()
+
+
+def warm_caches():
+    # char cache
+    s = bytes(range(256))
+    for i in range(256):
+        s[i:i+1]
+    # unicode cache
+    x = [chr(i) for i in range(256)]
+    # int cache
+    x = list(range(-5, 257))

Lib/test/libregrtest/runtest.py

+import faulthandler
+import importlib
+import io
+import json
+import os
+import sys
+import time
+import traceback
+import unittest
+from test import support
+from test.libregrtest.refleak import dash_R
+from test.libregrtest.save_env import saved_test_environment
+
+
+# Test result constants.
+PASSED = 1
+FAILED = 0
+ENV_CHANGED = -1
+SKIPPED = -2
+RESOURCE_DENIED = -3
+INTERRUPTED = -4
+CHILD_ERROR = -5   # error in a child process
+
+
+def run_test_in_subprocess(testname, ns):
+    """Run the given test in a subprocess with --slaveargs.
+
+    ns is the option Namespace parsed from command-line arguments. regrtest
+    is invoked in a subprocess with the --slaveargs argument; when the
+    subprocess exits, its return code, stdout and stderr are returned as a
+    3-tuple.
+    """
+    from subprocess import Popen, PIPE
+    base_cmd = ([sys.executable] + support.args_from_interpreter_flags() +
+                ['-X', 'faulthandler', '-m', 'test.regrtest'])
+
+    slaveargs = (
+            (testname, ns.verbose, ns.quiet),
+            dict(huntrleaks=ns.huntrleaks,
+                 use_resources=ns.use_resources,
+                 output_on_failure=ns.verbose3,
+                 timeout=ns.timeout, failfast=ns.failfast,
+                 match_tests=ns.match_tests))
+    # Running the child from the same working directory as regrtest's original
+    # invocation ensures that TEMPDIR for the child is the same when
+    # sysconfig.is_python_build() is true. See issue 15300.
+    popen = Popen(base_cmd + ['--slaveargs', json.dumps(slaveargs)],
+                  stdout=PIPE, stderr=PIPE,
+                  universal_newlines=True,
+                  close_fds=(os.name != 'nt'),
+                  cwd=support.SAVEDCWD)
+    stdout, stderr = popen.communicate()
+    retcode = popen.wait()
+    return retcode, stdout, stderr
+
+
+# small set of tests to determine if we have a basically functioning interpreter
+# (i.e. if any of these fail, then anything else is likely to follow)
+STDTESTS = [
+    'test_grammar',
+    'test_opcodes',
+    'test_dict',
+    'test_builtin',
+    'test_exceptions',
+    'test_types',
+    'test_unittest',
+    'test_doctest',
+    'test_doctest2',
+    'test_support'
+]
+
+# set of tests that we don't want to be executed when using regrtest
+NOTTESTS = set()
+
+
+def findtests(testdir=None, stdtests=STDTESTS, nottests=NOTTESTS):
+    """Return a list of all applicable test modules."""
+    testdir = findtestdir(testdir)
+    names = os.listdir(testdir)
+    tests = []
+    others = set(stdtests) | nottests
+    for name in names:
+        mod, ext = os.path.splitext(name)
+        if mod[:5] == "test_" and ext in (".py", "") and mod not in others:
+            tests.append(mod)
+    return stdtests + sorted(tests)
+
+
+def runtest(test, verbose, quiet,
+            huntrleaks=False, use_resources=None,
+            output_on_failure=False, failfast=False, match_tests=None,
+            timeout=None):
+    """Run a single test.
+
+    test -- the name of the test
+    verbose -- if true, print more messages
+    quiet -- if true, don't print 'skipped' messages (probably redundant)
+    huntrleaks -- run multiple times to test for leaks; requires a debug
+                  build; a triple corresponding to -R's three arguments
+    use_resources -- list of extra resources to use
+    output_on_failure -- if true, display test output on failure
+    timeout -- dump the traceback and exit if a test takes more than
+               timeout seconds
+    failfast, match_tests -- See regrtest command-line flags for these.
+
+    Returns the tuple result, test_time, where result is one of the constants:
+        INTERRUPTED      KeyboardInterrupt when run under -j
+        RESOURCE_DENIED  test skipped because resource denied
+        SKIPPED          test skipped for some other reason
+        ENV_CHANGED      test failed because it changed the execution environment
+        FAILED           test failed
+        PASSED           test passed
+    """
+
+    if use_resources is not None:
+        support.use_resources = use_resources
+    use_timeout = (timeout is not None)
+    if use_timeout:
+        faulthandler.dump_traceback_later(timeout, exit=True)
+    try:
+        support.match_tests = match_tests
+        if failfast:
+            support.failfast = True
+        if output_on_failure:
+            support.verbose = True
+
+            # Reuse the same instance to all calls to runtest(). Some
+            # tests keep a reference to sys.stdout or sys.stderr
+            # (eg. test_argparse).
+            if runtest.stringio is None:
+                stream = io.StringIO()
+                runtest.stringio = stream
+            else:
+                stream = runtest.stringio
+                stream.seek(0)
+                stream.truncate()
+
+            orig_stdout = sys.stdout
+            orig_stderr = sys.stderr
+            try:
+                sys.stdout = stream
+                sys.stderr = stream
+                result = runtest_inner(test, verbose, quiet, huntrleaks,
+                                       display_failure=False)
+                if result[0] == FAILED:
+                    output = stream.getvalue()
+                    orig_stderr.write(output)
+                    orig_stderr.flush()
+            finally:
+                sys.stdout = orig_stdout
+                sys.stderr = orig_stderr
+        else:
+            support.verbose = verbose  # Tell tests to be moderately quiet
+            result = runtest_inner(test, verbose, quiet, huntrleaks,
+                                   display_failure=not verbose)
+        return result
+    finally:
+        if use_timeout:
+            faulthandler.cancel_dump_traceback_later()
+        cleanup_test_droppings(test, verbose)
+runtest.stringio = None
+
+
+def runtest_inner(test, verbose, quiet,
+                  huntrleaks=False, display_failure=True):
+    support.unload(test)
+
+    test_time = 0.0
+    refleak = False  # True if the test leaked references.
+    try:
+        if test.startswith('test.'):
+            abstest = test
+        else:
+            # Always import it from the test package
+            abstest = 'test.' + test
+        with saved_test_environment(test, verbose, quiet) as environment:
+            start_time = time.time()
+            the_module = importlib.import_module(abstest)
+            # If the test has a test_main, that will run the appropriate
+            # tests.  If not, use normal unittest test loading.
+            test_runner = getattr(the_module, "test_main", None)
+            if test_runner is None:
+                def test_runner():
+                    loader = unittest.TestLoader()
+                    tests = loader.loadTestsFromModule(the_module)
+                    for error in loader.errors:
+                        print(error, file=sys.stderr)
+                    if loader.errors:
+                        raise Exception("errors while loading tests")
+                    support.run_unittest(tests)
+            test_runner()
+            if huntrleaks:
+                refleak = dash_R(the_module, test, test_runner, huntrleaks)
+            test_time = time.time() - start_time
+    except support.ResourceDenied as msg:
+        if not quiet:
+            print(test, "skipped --", msg)
+            sys.stdout.flush()
+        return RESOURCE_DENIED, test_time
+    except unittest.SkipTest as msg:
+        if not quiet:
+            print(test, "skipped --", msg)
+            sys.stdout.flush()
+        return SKIPPED, test_time
+    except KeyboardInterrupt:
+        raise
+    except support.TestFailed as msg:
+        if display_failure:
+            print("test", test, "failed --", msg, file=sys.stderr)
+        else:
+            print("test", test, "failed", file=sys.stderr)
+        sys.stderr.flush()
+        return FAILED, test_time
+    except:
+        msg = traceback.format_exc()
+        print("test", test, "crashed --", msg, file=sys.stderr)
+        sys.stderr.flush()
+        return FAILED, test_time
+    else:
+        if refleak:
+            return FAILED, test_time
+        if environment.changed:
+            return ENV_CHANGED, test_time
+        return PASSED, test_time
+
+
+def cleanup_test_droppings(testname, verbose):
+    import shutil
+    import stat
+    import gc
+
+    # First kill any dangling references to open files etc.
+    # This can also issue some ResourceWarnings which would otherwise get
+    # triggered during the following test run, and possibly produce failures.
+    gc.collect()
+
+    # Try to clean up junk commonly left behind.  While tests shouldn't leave
+    # any files or directories behind, when a test fails that can be tedious
+    # for it to arrange.  The consequences can be especially nasty on Windows,
+    # since if a test leaves a file open, it cannot be deleted by name (while
+    # there's nothing we can do about that here either, we can display the
+    # name of the offending test, which is a real help).
+    for name in (support.TESTFN,
+                 "db_home",
+                ):
+        if not os.path.exists(name):
+            continue
+
+        if os.path.isdir(name):
+            kind, nuker = "directory", shutil.rmtree
+        elif os.path.isfile(name):
+            kind, nuker = "file", os.unlink
+        else:
+            raise SystemError("os.path says %r exists but is neither "
+                              "directory nor file" % name)
+
+        if verbose:
+            print("%r left behind %s %r" % (testname, kind, name))
+        try:
+            # if we have chmod, fix possible permissions problems
+            # that might prevent cleanup
+            if (hasattr(os, 'chmod')):
+                os.chmod(name, stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO)
+            nuker(name)
+        except Exception as msg:
+            print(("%r left behind %s %r and it couldn't be "
+                "removed: %s" % (testname, kind, name, msg)), file=sys.stderr)
+
+
+def findtestdir(path=None):
+    return path or os.path.dirname(os.path.dirname(__file__)) or os.curdir