#!/usr/bin/env python## run-tests.py - Run a set of tests on Mercurial## Copyright 2006 Matt Mackall <mpm@selenic.com>## This software may be used and distributed according to the terms of the# GNU General Public License version 2 or any later version. # Modifying this script is tricky because it has many modes:#   - serial (default) vs parallel (-jN, N > 1)#   - no coverage (default) vs coverage (-c, -C, -s)#   - temp install (default) vs specific hg script (--with-hg, --local)#   - tests are a mix of shell scripts and Python scripts## If you change this script, it is recommended that you ensure you# haven't broken it by running it in various modes with a representative# sample of test scripts.  For example:##  1) serial, no coverage, temp install:#      ./run-tests.py test-s*#  2) serial, no coverage, local hg:#      ./run-tests.py --local test-s*#  3) serial, coverage, temp install:#      ./run-tests.py -c test-s*#  4) serial, coverage, local hg:#      ./run-tests.py -c --local test-s*      # unsupported#  5) parallel, no coverage, temp install:#      ./run-tests.py -j2 test-s*#  6) parallel, no coverage, local hg:#      ./run-tests.py -j2 --local test-s*#  7) parallel, coverage, temp install:#      ./run-tests.py -j2 -c test-s*          # currently broken#  8) parallel, coverage, local install:#      ./run-tests.py -j2 -c --local test-s*  # unsupported (and broken)#  9) parallel, custom tmp dir:#      ./run-tests.py -j2 --tmpdir /tmp/myhgtests#  10) parallel, pure, tests that call run-tests:#      ./run-tests.py --pure `grep -l run-tests.py *.t`## (You could use any subset of the tests: test-s* happens to match# enough that it's worth doing parallel runs, few enough that it# completes fairly quickly, includes both shell and Python scripts, and# includes some scripts that run daemon processes.) from __future__ import absolute_import, print_function import argparseimport collectionsimport difflibimport distutils.version as versionimport errnoimport jsonimport multiprocessingimport osimport randomimport reimport shutilimport signalimport socketimport subprocessimport sysimport sysconfigimport tempfileimport threadingimport timeimport unittestimport uuidimport xml.dom.minidom as minidom try:    import Queue as queueexcept ImportError:    import queue try:    import shlex     shellquote = shlex.quoteexcept (ImportError, AttributeError):    import pipes     shellquote = pipes.quote processlock = threading.Lock() pygmentspresent = False# ANSI color is unsupported prior to Windows 10if os.name != 'nt':    try:  # is pygments installed        import pygments        import pygments.lexers as lexers        import pygments.lexer as lexer        import pygments.formatters as formatters        import pygments.token as token        import pygments.style as style         pygmentspresent = True        difflexer = lexers.DiffLexer()        terminal256formatter = formatters.Terminal256Formatter()    except ImportError:        pass if pygmentspresent:     class TestRunnerStyle(style.Style):        default_style = ""        skipped = token.string_to_tokentype("Token.Generic.Skipped")        failed = token.string_to_tokentype("Token.Generic.Failed")        skippedname = token.string_to_tokentype("Token.Generic.SName")        failedname = token.string_to_tokentype("Token.Generic.FName")        styles = {            skipped: '#e5e5e5',            skippedname: '#00ffff',            failed: '#7f0000',            failedname: '#ff0000',        }     class TestRunnerLexer(lexer.RegexLexer):        testpattern = r'[\w-]+\.(t|py)(#[a-zA-Z0-9_\-\.]+)?'        tokens = {            'root': [                (r'^Skipped', token.Generic.Skipped, 'skipped'),                (r'^Failed ', token.Generic.Failed, 'failed'),                (r'^ERROR: ', token.Generic.Failed, 'failed'),            ],            'skipped': [                (testpattern, token.Generic.SName),                (r':.*', token.Generic.Skipped),            ],            'failed': [                (testpattern, token.Generic.FName),                (r'(:| ).*', token.Generic.Failed),            ],        }     runnerformatter = formatters.Terminal256Formatter(style=TestRunnerStyle)    runnerlexer = TestRunnerLexer() origenviron = os.environ.copy() if sys.version_info > (3, 5, 0):    PYTHON3 = True    xrange = range  # we use xrange in one place, and we'd rather not use range     def _bytespath(p):        if p is None:            return p        return p.encode('utf-8')     def _strpath(p):        if p is None:            return p        return p.decode('utf-8')     osenvironb = getattr(os, 'environb', None)    if osenvironb is None:        # Windows lacks os.environb, for instance.  A proxy over the real thing        # instead of a copy allows the environment to be updated via bytes on        # all platforms.        class environbytes(object):            def __init__(self, strenv):                self.__len__ = strenv.__len__                self.clear = strenv.clear                self._strenv = strenv             def __getitem__(self, k):                v = self._strenv.__getitem__(_strpath(k))                return _bytespath(v)             def __setitem__(self, k, v):                self._strenv.__setitem__(_strpath(k), _strpath(v))             def __delitem__(self, k):                self._strenv.__delitem__(_strpath(k))             def __contains__(self, k):                return self._strenv.__contains__(_strpath(k))             def __iter__(self):                return iter([_bytespath(k) for k in iter(self._strenv)])             def get(self, k, default=None):                v = self._strenv.get(_strpath(k), _strpath(default))                return _bytespath(v)             def pop(self, k, default=None):                v = self._strenv.pop(_strpath(k), _strpath(default))                return _bytespath(v)         osenvironb = environbytes(os.environ)     getcwdb = getattr(os, 'getcwdb')    if not getcwdb or os.name == 'nt':        getcwdb = lambda: _bytespath(os.getcwd()) elif sys.version_info >= (3, 0, 0):    print(        '%s is only supported on Python 3.5+ and 2.7, not %s'        % (sys.argv[0], '.'.join(str(v) for v in sys.version_info[:3]))    )    sys.exit(70)  # EX_SOFTWARE from `man 3 sysexit`else:    PYTHON3 = False     # In python 2.x, path operations are generally done using    # bytestrings by default, so we don't have to do any extra    # fiddling there. We define the wrapper functions anyway just to    # help keep code consistent between platforms.    def _bytespath(p):        return p     _strpath = _bytespath    osenvironb = os.environ    getcwdb = os.getcwd # For Windows supportwifexited = getattr(os, "WIFEXITED", lambda x: False) # Whether to use IPv6def checksocketfamily(name, port=20058):    """return true if we can listen on localhost using family=name     name should be either 'AF_INET', or 'AF_INET6'.    port being used is okay - EADDRINUSE is considered as successful.    """    family = getattr(socket, name, None)    if family is None:        return False    try:        s = socket.socket(family, socket.SOCK_STREAM)        s.bind(('localhost', port))        s.close()        return True    except socket.error as exc:        if exc.errno == errno.EADDRINUSE:            return True        elif exc.errno in (errno.EADDRNOTAVAIL, errno.EPROTONOSUPPORT):            return False        else:            raise    else:        return False  # useipv6 will be set by parseargsuseipv6 = None  def checkportisavailable(port):    """return true if a port seems free to bind on localhost"""    if useipv6:        family = socket.AF_INET6    else:        family = socket.AF_INET    try:        s = socket.socket(family, socket.SOCK_STREAM)        s.bind(('localhost', port))        s.close()        return True    except socket.error as exc:        if exc.errno not in (            errno.EADDRINUSE,            errno.EADDRNOTAVAIL,            errno.EPROTONOSUPPORT,        ):            raise    return False  closefds = os.name == 'posix'  def Popen4(cmd, wd, timeout, env=None):    processlock.acquire()    p = subprocess.Popen(        _strpath(cmd),        shell=True,        bufsize=-1,        cwd=_strpath(wd),        env=env,        close_fds=closefds,        stdin=subprocess.PIPE,        stdout=subprocess.PIPE,        stderr=subprocess.STDOUT,    )    processlock.release()     p.fromchild = p.stdout    p.tochild = p.stdin    p.childerr = p.stderr     p.timeout = False    if timeout:         def t():            start = time.time()            while time.time() - start < timeout and p.returncode is None:                time.sleep(0.1)            p.timeout = True            if p.returncode is None:                terminate(p)         threading.Thread(target=t).start()     return p  if sys.executable:    sysexecutable = sys.executableelif os.environ.get('PYTHONEXECUTABLE'):    sysexecutable = os.environ['PYTHONEXECUTABLE']elif os.environ.get('PYTHON'):    sysexecutable = os.environ['PYTHON']else:    raise AssertionError('Could not find Python interpreter') PYTHON = _bytespath(sysexecutable.replace('\\', '/'))IMPL_PATH = b'PYTHONPATH'if 'java' in sys.platform:    IMPL_PATH = b'JYTHONPATH' defaults = {    'jobs': ('HGTEST_JOBS', multiprocessing.cpu_count()),    'timeout': ('HGTEST_TIMEOUT', 180),    'slowtimeout': ('HGTEST_SLOWTIMEOUT', 1500),    'port': ('HGTEST_PORT', 20059),    'shell': ('HGTEST_SHELL', 'sh'),}  def canonpath(path):    return os.path.realpath(os.path.expanduser(path))  def parselistfiles(files, listtype, warn=True):    entries = dict()    for filename in files:        try:            path = os.path.expanduser(os.path.expandvars(filename))            f = open(path, "rb")        except IOError as err:            if err.errno != errno.ENOENT:                raise            if warn:                print("warning: no such %s file: %s" % (listtype, filename))            continue         for line in f.readlines():            line = line.split(b'#', 1)[0].strip()            if line:                entries[line] = filename         f.close()    return entries  def parsettestcases(path):    """read a .t test file, return a set of test case names     If path does not exist, return an empty set.    """    cases = []    try:        with open(path, 'rb') as f:            for l in f:                if l.startswith(b'#testcases '):                    cases.append(sorted(l[11:].split()))    except IOError as ex:        if ex.errno != errno.ENOENT:            raise    return cases  def getparser():    """Obtain the OptionParser used by the CLI."""    parser = argparse.ArgumentParser(usage='%(prog)s [options] [tests]')     selection = parser.add_argument_group('Test Selection')    selection.add_argument(        '--allow-slow-tests',        action='store_true',        help='allow extremely slow tests',    )    selection.add_argument(        "--blacklist",        action="append",        help="skip tests listed in the specified blacklist file",    )    selection.add_argument(        "--changed",        help="run tests that are changed in parent rev or working directory",    )    selection.add_argument(        "-k", "--keywords", help="run tests matching keywords"    )    selection.add_argument(        "-r", "--retest", action="store_true", help="retest failed tests"    )    selection.add_argument(        "--test-list",        action="append",        help="read tests to run from the specified file",    )    selection.add_argument(        "--whitelist",        action="append",        help="always run tests listed in the specified whitelist file",    )    selection.add_argument(        'tests', metavar='TESTS', nargs='*', help='Tests to run'    )     harness = parser.add_argument_group('Test Harness Behavior')    harness.add_argument(        '--bisect-repo',        metavar='bisect_repo',        help=(            "Path of a repo to bisect. Use together with " "--known-good-rev"        ),    )    harness.add_argument(        "-d",        "--debug",        action="store_true",        help="debug mode: write output of test scripts to console"        " rather than capturing and diffing it (disables timeout)",    )    harness.add_argument(        "-f",        "--first",        action="store_true",        help="exit on the first test failure",    )    harness.add_argument(        "-i",        "--interactive",        action="store_true",        help="prompt to accept changed output",    )    harness.add_argument(        "-j",        "--jobs",        type=int,        help="number of jobs to run in parallel"        " (default: $%s or %d)" % defaults['jobs'],    )    harness.add_argument(        "--keep-tmpdir",        action="store_true",        help="keep temporary directory after running tests",    )    harness.add_argument(        '--known-good-rev',        metavar="known_good_rev",        help=(            "Automatically bisect any failures using this "            "revision as a known-good revision."        ),    )    harness.add_argument(        "--list-tests",        action="store_true",        help="list tests instead of running them",    )    harness.add_argument(        "--loop", action="store_true", help="loop tests repeatedly"    )    harness.add_argument(        '--random', action="store_true", help='run tests in random order'    )    harness.add_argument(        '--order-by-runtime',        action="store_true",        help='run slowest tests first, according to .testtimes',    )    harness.add_argument(        "-p",        "--port",        type=int,        help="port on which servers should listen"        " (default: $%s or %d)" % defaults['port'],    )    harness.add_argument(        '--profile-runner',        action='store_true',        help='run statprof on run-tests',    )    harness.add_argument(        "-R", "--restart", action="store_true", help="restart at last error"    )    harness.add_argument(        "--runs-per-test",        type=int,        dest="runs_per_test",        help="run each test N times (default=1)",        default=1,    )    harness.add_argument(        "--shell", help="shell to use (default: $%s or %s)" % defaults['shell']

Lines 1–500 of 3745. Load more