comparison tests/run-tests.py @ 43076:2372284d9457

formatting: blacken the codebase

This is using my patch to black
(https://github.com/psf/black/pull/826) so we don't un-wrap collection
literals.

Done with:

  hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**"' | xargs black -S

# skip-blame mass-reformatting only

# no-check-commit reformats foo_bar functions

Differential Revision: https://phab.mercurial-scm.org/D6971

43075:57875cf423c9

except ImportError:
    import queue

try:
    import shlex
    shellquote = shlex.quote
except (ImportError, AttributeError):
    import pipes
    shellquote = pipes.quote

processlock = threading.Lock()

pygmentspresent = False
# ANSI color is unsupported prior to Windows 10
if 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 = {

                (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')

        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

        else:
            raise
    else:
        return False

# useipv6 will be set by parseargs
useipv6 = None

def checkportisavailable(port):
    """return true if a port seems free to bind on localhost"""
    if useipv6:
        family = socket.AF_INET6

        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(.1)
            p.timeout = True
            if p.returncode is None:
                terminate(p)
        threading.Thread(target=t).start()

    return p

if sys.executable:
    sysexecutable = sys.executable
elif os.environ.get('PYTHONEXECUTABLE'):
    sysexecutable = os.environ['PYTHONEXECUTABLE']

    '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:

                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.
    """

    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'])
    harness.add_argument('--showchannels', action='store_true',
                        help='show scheduling channels')
    harness.add_argument("--slowtimeout", type=int,
        help="kill errant slow tests after SLOWTIMEOUT seconds"
             " (default: $%s or %d)" % defaults['slowtimeout'])
    harness.add_argument("-t", "--timeout", type=int,
        help="kill errant tests after TIMEOUT seconds"
             " (default: $%s or %d)" % defaults['timeout'])
    harness.add_argument("--tmpdir",
        help="run tests in the given temporary directory"
             " (implies --keep-tmpdir)")
    harness.add_argument("-v", "--verbose", action="store_true",
        help="output verbose messages")

    hgconf = parser.add_argument_group('Mercurial Configuration')
    hgconf.add_argument("--chg", action="store_true",
        help="install and use chg wrapper in place of hg")
    hgconf.add_argument("--compiler",
        help="compiler to build with")
    hgconf.add_argument('--extra-config-opt', action="append", default=[],
        help='set the given config opt in the test hgrc')
    hgconf.add_argument("-l", "--local", action="store_true",
        help="shortcut for --with-hg=<testdir>/../hg, "
             "and --with-chg=<testdir>/../contrib/chg/chg if --chg is set")
    hgconf.add_argument("--ipv6", action="store_true",
        help="prefer IPv6 to IPv4 for network related tests")
    hgconf.add_argument("--pure", action="store_true",
        help="use pure Python code instead of C extensions")
    hgconf.add_argument("-3", "--py3-warnings", action="store_true",
        help="enable Py3k warnings on Python 2.7+")
    hgconf.add_argument("--with-chg", metavar="CHG",
        help="use specified chg wrapper in place of hg")
    hgconf.add_argument("--with-hg",
        metavar="HG",
        help="test using specified hg script rather than a "
             "temporary installation")

    reporting = parser.add_argument_group('Results Reporting')
    reporting.add_argument("-C", "--annotate", action="store_true",
        help="output files annotated with coverage")
    reporting.add_argument("--color", choices=["always", "auto", "never"],
        default=os.environ.get('HGRUNTESTSCOLOR', 'auto'),
        help="colorisation: always|auto|never (default: auto)")
    reporting.add_argument("-c", "--cover", action="store_true",
        help="print a test coverage report")
    reporting.add_argument('--exceptions', action='store_true',
        help='log all exceptions and generate an exception report')
    reporting.add_argument("-H", "--htmlcov", action="store_true",
        help="create an HTML report of the coverage of the files")
    reporting.add_argument("--json", action="store_true",
        help="store test result data in 'report.json' file")
    reporting.add_argument("--outputdir",
        help="directory to write error logs to (default=test directory)")
    reporting.add_argument("-n", "--nodiff", action="store_true",
        help="skip showing test changes")
    reporting.add_argument("-S", "--noskips", action="store_true",
        help="don't report skip tests verbosely")
    reporting.add_argument("--time", action="store_true",
        help="time how long each test takes")
    reporting.add_argument("--view",
        help="external diff viewer")
    reporting.add_argument("--xunit",
        help="record xunit results at specified path")

    for option, (envvar, default) in defaults.items():
        defaults[option] = type(default)(os.environ.get(envvar, default))
    parser.set_defaults(**defaults)

    return parser

def parseargs(args, parser):
    """Parse arguments with our OptionParser and validate results."""
    options = parser.parse_args(args)

        if options.chg:
            pathandattrs.append((b'contrib/chg/chg', 'with_chg'))
        for relpath, attr in pathandattrs:
            binpath = os.path.join(reporootdir, relpath)
            if os.name != 'nt' and not os.access(binpath, os.X_OK):
                parser.error('--local specified, but %r not found or '
                             'not executable' % binpath)
            setattr(options, attr, _strpath(binpath))

    if options.with_hg:
        options.with_hg = canonpath(_bytespath(options.with_hg))
        if not (os.path.isfile(options.with_hg) and
                os.access(options.with_hg, os.X_OK)):
            parser.error('--with-hg must specify an executable hg script')
        if os.path.basename(options.with_hg) not in [b'hg', b'hg.exe']:
            sys.stderr.write('warning: --with-hg should specify an hg script\n')
            sys.stderr.flush()

    if (options.chg or options.with_chg) and os.name == 'nt':
        parser.error('chg does not work on %s' % os.name)
    if options.with_chg:
        options.chg = False  # no installation to temporary location
        options.with_chg = canonpath(_bytespath(options.with_chg))
        if not (os.path.isfile(options.with_chg) and
                os.access(options.with_chg, os.X_OK)):
            parser.error('--with-chg must specify a chg executable')
    if options.chg and options.with_hg:
        # chg shares installation location with hg
        parser.error('--chg does not work when --with-hg is specified '
                     '(use --with-chg instead)')

    if options.color == 'always' and not pygmentspresent:
        sys.stderr.write('warning: --color=always ignored because '
                         'pygments is not installed\n')

    if options.bisect_repo and not options.known_good_rev:
        parser.error("--bisect-repo cannot be used without --known-good-rev")

    global useipv6
    if options.ipv6:
        useipv6 = checksocketfamily('AF_INET6')
    else:
        # only use IPv6 if IPv4 is unavailable and IPv6 is available
        useipv6 = ((not checksocketfamily('AF_INET'))
                   and checksocketfamily('AF_INET6'))

    options.anycoverage = options.cover or options.annotate or options.htmlcov
    if options.anycoverage:
        try:
            import coverage
            covver = version.StrictVersion(coverage.__version__).version
            if covver < (3, 3):
                parser.error('coverage options require coverage 3.3 or later')
        except ImportError:
            parser.error('coverage options now require the coverage package')

    if options.anycoverage and options.local:
        # this needs some path mangling somewhere, I guess
        parser.error("sorry, coverage options do not work when --local "
                     "is specified")

    if options.anycoverage and options.with_hg:
        parser.error("sorry, coverage options do not work when --with-hg "
                     "is specified")

    global verbose
    if options.verbose:
        verbose = ''

        parser.error('--jobs must be positive')
    if options.interactive and options.debug:
        parser.error("-i/--interactive and -d/--debug are incompatible")
    if options.debug:
        if options.timeout != defaults['timeout']:
            sys.stderr.write(
                'warning: --timeout option ignored with --debug\n')
        if options.slowtimeout != defaults['slowtimeout']:
            sys.stderr.write(
                'warning: --slowtimeout option ignored with --debug\n')
        options.timeout = 0
        options.slowtimeout = 0
    if options.py3_warnings:
        if PYTHON3:
            parser.error(
                '--py3-warnings can only be used on Python 2.7')

    if options.blacklist:
        options.blacklist = parselistfiles(options.blacklist, 'blacklist')
    if options.whitelist:
        options.whitelisted = parselistfiles(options.whitelist, 'whitelist')

    if options.showchannels:
        options.nodiff = True

    return options

def rename(src, dst):
    """Like os.rename(), trade atomicity and opened files friendliness
    for existing destination support.
    """
    shutil.copy(src, dst)
    os.remove(src)

def makecleanable(path):
    """Try to fix directory permission recursively so that the entire tree
    can be deleted"""
    for dirpath, dirnames, _filenames in os.walk(path, topdown=True):

            try:
                os.chmod(p, os.stat(p).st_mode & 0o777 | 0o700)  # chmod u+rwx
            except OSError:
                pass

_unified_diff = difflib.unified_diff
if PYTHON3:
    import functools
    _unified_diff = functools.partial(difflib.diff_bytes, difflib.unified_diff)

def getdiff(expected, output, ref, err):
    servefail = False
    lines = []
    for line in _unified_diff(expected, output, ref, err):

            line = line.replace(b'\\', b'/')
            if line.endswith(b' \n'):
                line = line[:-2] + b'\n'
        lines.append(line)
        if not servefail and line.startswith(
                             b'+  abort: child process failed to start'):
            servefail = True

    return servefail, lines

verbose = False
def vlog(*msg):
    """Log only when in verbose mode."""
    if verbose is False:
        return

    return log(*msg)

# Bytes that break XML even in a CDATA block: control characters 0-31
# sans \t, \n and \r
CDATA_EVIL = re.compile(br"[\000-\010\013\014\016-\037]")

# Match feature conditionalized output lines in the form, capturing the feature
# list in group 2, and the preceeding line output in group 1:
#
#   output..output (feature !)\n
optline = re.compile(br'(.*) \((.+?) !\)\n$')

def cdatasafe(data):
    """Make a string safe to include in a CDATA block.

    Certain control characters are illegal in a CDATA block, and
    there's no way to include a ]]> in a CDATA either. This function
    replaces illegal bytes with ? and adds a space between the ]] so
    that it won't break the CDATA block.
    """
    return CDATA_EVIL.sub(b'?', data).replace(b']]>', b'] ]>')

def log(*msg):
    """Log something to stdout.

    Arguments are strings to print.

        for m in msg:
            print(m, end=' ')
        print()
        sys.stdout.flush()

def highlightdiff(line, color):
    if not color:
        return line
    assert pygmentspresent
    return pygments.highlight(line.decode('latin1'), difflexer,
                              terminal256formatter).encode('latin1')

def highlightmsg(msg, color):
    if not color:
        return msg
    assert pygmentspresent
    return pygments.highlight(msg, runnerlexer, runnerformatter)

def terminate(proc):
    """Terminate subprocess"""
    vlog('# Terminating process %d' % proc.pid)
    try:
        proc.terminate()
    except OSError:
        pass

def killdaemons(pidfile):
    import killdaemons as killmod
    return killmod.killdaemons(pidfile, tryhard=False, remove=True,
                               logfn=vlog)

class Test(unittest.TestCase):
    """Encapsulates a single, runnable test.

    While this class conforms to the unittest.TestCase API, it differs in that

    """

    # Status code reserved for skipped tests (used by hghave).
    SKIPPED_STATUS = 80

    def __init__(self, path, outputdir, tmpdir, keeptmpdir=False,
                 debug=False,
                 first=False,
                 timeout=None,
                 startport=None, extraconfigopts=None,
                 py3warnings=False, shell=None, hgcommand=None,
                 slowtimeout=None, usechg=False,
                 useipv6=False):
        """Create a test from parameters.

        path is the full path to the file defining the test.

        tmpdir is the main temporary directory to use for this test.

    def readrefout(self):
        """read reference output"""
        # If we're not in --debug mode and reference output file exists,
        # check test output against it.
        if self._debug:
            return None # to match "out is None"
        elif os.path.exists(self.refpath):
            with open(self.refpath, 'rb') as f:
                return f.read().splitlines(True)
        else:
            return []

                # file.
                if e.errno != errno.ENOENT:
                    raise

        if self._usechg:
            self._chgsockdir = os.path.join(self._threadtmp,
                                            b'%s.chgsock' % name)
            os.mkdir(self._chgsockdir)

    def run(self, result):
        """Run this test and report results against a TestResult instance."""
        # This function is extremely similar to unittest.TestCase.run(). Once

            return 'returned error code %d' % ret

        self._skipped = False

        if ret == self.SKIPPED_STATUS:
            if out is None: # Debug mode, nothing to parse.
                missing = ['unknown']
                failed = None
            else:
                missing, failed = TTest.parsehghaveoutput(out)

            self.fail('timed out')
        elif ret is False:
            self.fail('no result code from test')
        elif out != self._refout:
            # Diff generation may rely on written .err file.
            if ((ret != 0 or out != self._refout) and not self._skipped
                and not self._debug):
                with open(self.errpath, 'wb') as f:
                    for line in out:
                        f.write(line)

            # The result object handles diff calculation for us.

        for entry in self._daemonpids:
            killdaemons(entry)
        self._daemonpids = []

        if self._keeptmpdir:
            log('\nKeeping testtmp dir: %s\nKeeping threadtmp dir: %s' %
                (self._testtmp.decode('utf-8'),
                 self._threadtmp.decode('utf-8')))
        else:
            try:
                shutil.rmtree(self._testtmp)
            except OSError:
                # unreadable directory may be left in $TESTTMP; fix permission

        if self._usechg:
            # chgservers will stop automatically after they find the socket
            # files are deleted
            shutil.rmtree(self._chgsockdir, True)

        if ((self._ret != 0 or self._out != self._refout) and not self._skipped
            and not self._debug and self._out):
            with open(self.errpath, 'wb') as f:
                for line in self._out:
                    f.write(line)

        vlog("# Ret was:", self._ret, '(%s)' % self.name)

            self._portmap(0),
            self._portmap(1),
            self._portmap(2),
            (br'([^0-9])%s' % re.escape(self._localip()), br'\1$LOCALIP'),
            (br'\bHG_TXNID=TXN:[a-f0-9]{40}\b', br'HG_TXNID=TXN:$ID$'),
            ]
        r.append((self._escapepath(self._testtmp), b'$TESTTMP'))

        replacementfile = os.path.join(self._testdir, b'common-pattern.py')

        if os.path.exists(replacementfile):

                    r.append(value)
        return r

    def _escapepath(self, p):
        if os.name == 'nt':
            return (
                (b''.join(c.isalpha() and b'[%s%s]' % (c.lower(), c.upper()) or
                    c in b'/\\' and br'[/\\]' or c.isdigit() and c or b'\\' + c
                    for c in [p[i:i + 1] for i in range(len(p))]))
            )
        else:
            return re.escape(p)

    def _localip(self):

                    continue
                envf.write('unset %s\n' % (name,))

    def _getenv(self):
        """Obtain environment variables to use during test execution."""
        def defineport(i):
            offset = '' if i == 0 else '%s' % i
            env["HGPORT%s" % offset] = '%s' % (self._startport + i)
        env = os.environ.copy()
        env['PYTHONUSERBASE'] = sysconfig.get_config_var('userbase') or ''
        env['HGEMITWARNINGS'] = '1'
        env['TESTTMP'] = _strpath(self._testtmp)
        env['TESTNAME'] = self.name

        # This number should match portneeded in _getport
        for port in xrange(3):
            # This list should be parallel to _portmap in _getreplacements
            defineport(port)
        env["HGRCPATH"] = _strpath(os.path.join(self._threadtmp, b'.hgrc'))
        env["DAEMON_PIDS"] = _strpath(os.path.join(self._threadtmp,
                                                   b'daemon.pids'))
        env["HGEDITOR"] = ('"' + sysexecutable + '"'
                           + ' -c "import sys; sys.exit(0)"')
        env["HGUSER"]   = "test"
        env["HGENCODING"] = "ascii"
        env["HGENCODINGMODE"] = "strict"
        env["HGHOSTNAME"] = "test-hostname"
        env['HGIPV6'] = str(int(self._useipv6))
        # See contrib/catapipe.py for how to use this functionality.
        if 'HGTESTCATAPULTSERVERPIPE' not in env:
            # If we don't have HGTESTCATAPULTSERVERPIPE explicitly set, pull the
            # non-test one in as a default, otherwise set to devnull
            env['HGTESTCATAPULTSERVERPIPE'] = env.get(
                'HGCATAPULTSERVERPIPE', os.devnull)

        extraextensions = []
        for opt in self._extraconfigopts:
            section, key = opt.encode('utf-8').split(b'.', 1)
            if section != 'extensions':

            hgrc.write(b'[defaults]\n')
            hgrc.write(b'[devel]\n')
            hgrc.write(b'all-warnings = true\n')
            hgrc.write(b'default-date = 0 0\n')
            hgrc.write(b'[largefiles]\n')
            hgrc.write(b'usercache = %s\n' %
                       (os.path.join(self._testtmp, b'.cache/largefiles')))
            hgrc.write(b'[lfs]\n')
            hgrc.write(b'usercache = %s\n' %
                       (os.path.join(self._testtmp, b'.cache/lfs')))
            hgrc.write(b'[web]\n')
            hgrc.write(b'address = localhost\n')
            hgrc.write(b'ipv6 = %s\n' % str(self._useipv6).encode('ascii'))
            hgrc.write(b'server-header = testing stub value\n')

            for opt in self._extraconfigopts:
                section, key = opt.encode('utf-8').split(b'.', 1)
                assert b'=' in key, ('extra config opt %s must '
                                     'have an = for assignment' % opt)
                hgrc.write(b'[%s]\n%s\n' % (section, key))

    def fail(self, msg):
        # unittest differentiates between errored and failed.
        # Failed is denoted by AssertionError (by default at least).

        stderr).

        Return a tuple (exitcode, output). output is None in debug mode.
        """
        if self._debug:
            proc = subprocess.Popen(_strpath(cmd), shell=True,
                                    cwd=_strpath(self._testtmp),
                                    env=env)
            ret = proc.wait()
            return (ret, None)

        proc = Popen4(cmd, self._testtmp, self._timeout, env)
        def cleanup():
            terminate(proc)
            ret = proc.wait()
            if ret == 0:
                ret = signal.SIGTERM << 8

        if normalizenewlines:
            output = output.replace(b'\r\n', b'\n')

        return ret, output.splitlines(True)

class PythonTest(Test):
    """A Python-based test."""

    @property
    def refpath(self):

        py3switch = self._py3warnings and b' -3' or b''
        # Quote the python(3) executable for Windows
        cmd = b'"%s"%s "%s"' % (PYTHON, py3switch, self.path)
        vlog("# Running", cmd)
        normalizenewlines = os.name == 'nt'
        result = self._runcommand(cmd, env,
                                  normalizenewlines=normalizenewlines)
        if self._aborted:
            raise KeyboardInterrupt()

        return result

# Some glob patterns apply only in some circumstances, so the script
# might want to remove (glob) annotations that otherwise should be
# retained.
checkcodeglobpats = [

WARN_YES = 2
WARN_NO = 3

MARK_OPTIONAL = b" (?)\n"

def isoptional(line):
    return line.endswith(MARK_OPTIONAL)

class TTest(Test):
    """A "t test" is a test backed by a .t file."""

    SKIPPED_PREFIX = b'skipped: '

            return self._have.get(allreqs)

        # TODO do something smarter when all other uses of hghave are gone.
        runtestdir = os.path.abspath(os.path.dirname(_bytespath(__file__)))
        tdir = runtestdir.replace(b'\\', b'/')
        proc = Popen4(b'%s -c "%s/hghave %s"' %
                      (self._shell, tdir, allreqs),
                      self._testtmp, 0, self._getenv())
        stdout, stderr = proc.communicate()
        ret = proc.wait()
        if wifexited(ret):
            ret = os.WEXITSTATUS(ret)
        if ret == 2:

    def _parsetest(self, lines):
        # We generate a shell script which outputs unique markers to line
        # up script results with our source. These markers include input
        # line number and the last return code.
        salt = b"SALT%d" % time.time()
        def addsalt(line, inpython):
            if inpython:
                script.append(b'%s %d 0\n' % (salt, line))
            else:
                script.append(b'echo %s %d $?\n' % (salt, line))
        activetrace = []
        session = str(uuid.uuid4())
        if PYTHON3:
            session = session.encode('ascii')
        hgcatapult = (os.getenv('HGTESTCATAPULTSERVERPIPE') or
                      os.getenv('HGCATAPULTSERVERPIPE'))
        def toggletrace(cmd=None):
            if not hgcatapult or hgcatapult == os.devnull:
                return

            if activetrace:
                script.append(
                    b'echo END %s %s >> "$HGTESTCATAPULTSERVERPIPE"\n' % (
                        session, activetrace[0]))
            if cmd is None:
                return

            if isinstance(cmd, str):
                quoted = shellquote(cmd.strip())
            else:
                quoted = shellquote(cmd.strip().decode('utf8')).encode('utf8')
            quoted = quoted.replace(b'\\', b'\\\\')
            script.append(
                b'echo START %s %s >> "$HGTESTCATAPULTSERVERPIPE"\n' % (
                    session, quoted))
            activetrace[0:] = [quoted]

        script = []

        # After we run the shell script, we re-unify the script output

                    after.setdefault(pos, []).append('  !!! missing #if\n')
                skipping = None
                after.setdefault(pos, []).append(l)
            elif skipping:
                after.setdefault(pos, []).append(l)
            elif l.startswith(b'  >>> '): # python inlines
                after.setdefault(pos, []).append(l)
                prepos = pos
                pos = n
                if not inpython:
                    # We've just entered a Python block. Add the header.
                    inpython = True
                    addsalt(prepos, False) # Make sure we report the exit code.
                    script.append(b'"%s" -m heredoctest <<EOF\n' % PYTHON)
                addsalt(n, True)
                script.append(l[2:])
            elif l.startswith(b'  ... '): # python inlines
                after.setdefault(prepos, []).append(l)
                script.append(l[2:])
            elif l.startswith(b'  $ '): # commands
                if inpython:
                    script.append(b'EOF\n')
                    inpython = False
                after.setdefault(pos, []).append(l)
                prepos = pos

                cmd = rawcmd.split()
                toggletrace(rawcmd)
                if len(cmd) == 2 and cmd[0] == b'cd':
                    l = b'  $ cd %s || exit 1\n' % cmd[1]
                script.append(rawcmd)
            elif l.startswith(b'  > '): # continuations
                after.setdefault(prepos, []).append(l)
                script.append(l[4:])
            elif l.startswith(b'  '): # results
                # Queue up a list of expected results.
                expected.setdefault(pos, []).append(l[2:])
            else:
                if inpython:
                    script.append(b'EOF\n')

            toggletrace()
        return salt, script, after, expected

    def _processoutput(self, exitcode, output, salt, after, expected):
        # Merge the script output back into a unified test.
        warnonly = WARN_UNDEFINED # 1: not yet; 2: yes; 3: for sure not
        if exitcode != 0:
            warnonly = WARN_NO

        pos = -1
        postout = []
        for out_rawline in output:
            out_line, cmd_line = out_rawline, None
            if salt in out_rawline:
                out_line, cmd_line = out_rawline.split(salt, 1)

            pos, postout, warnonly = self._process_out_line(out_line,
                                                            pos,
                                                            postout,
                                                            expected,
                                                            warnonly)
            pos, postout = self._process_cmd_line(cmd_line, pos, postout,
                                                  after)

        if pos in after:
            postout += after.pop(pos)

        if warnonly == WARN_YES:
            exitcode = False # Set exitcode to warned.

        return exitcode, postout

    def _process_out_line(self, out_line, pos, postout, expected, warnonly):
        while out_line:

                if el:
                    r, exact = self.linematch(el, out_line)
                if isinstance(r, str):
                    if r == '-glob':
                        out_line = ''.join(el.rsplit(' (glob)', 1))
                        r = '' # Warn only this line.
                    elif r == "retry":
                        postout.append(b'  ' + el)
                    else:
                        log('\ninfo, unknown linematch result: %r\n' % r)
                        r = False

                    if isoptional(el):
                        optional.append(i)
                    else:
                        m = optline.match(el)
                        if m:
                            conditions = [
                                c for c in m.group(2).split(b' ')]

                            if not self._iftest(conditions):
                                optional.append(i)
                    if exact:
                        # Don't allow line to be matches against a later

                for i in reversed(optional):
                    del els[i]
                postout.append(b'  ' + el)
            else:
                if self.NEEDESCAPE(out_line):
                    out_line = TTest._stringescape(b'%s (esc)\n' %
                                               out_line.rstrip(b'\n'))
                postout.append(b'  ' + out_line) # Let diff deal with it.
                if r != '': # If line failed.
                    warnonly = WARN_NO
                elif warnonly == WARN_UNDEFINED:
                    warnonly = WARN_YES
            break
        else:

            return True
        el = el.replace(b'$LOCALIP', b'*')
        i, n = 0, len(el)
        res = b''
        while i < n:
            c = el[i:i + 1]
            i += 1
            if c == b'\\' and i < n and el[i:i + 1] in b'*?\\/':
                res += el[i - 1:i + 1]
                i += 1
            elif c == b'*':
                res += b'.*'
            elif c == b'?':
                res += b'.'

            else:
                res += re.escape(c)
        return TTest.rematch(res, l)

    def linematch(self, el, l):
        if el == l: # perfect match (fast)
            return True, True
        retry = False
        if isoptional(el):
            retry = "retry"
            el = el[:-len(MARK_OPTIONAL)] + b"\n"
        else:
            m = optline.match(el)
            if m:
                conditions = [c for c in m.group(2).split(b' ')]

        missing = []
        failed = []
        for line in lines:
            if line.startswith(TTest.SKIPPED_PREFIX):
                line = line.splitlines()[0]
                missing.append(line[len(TTest.SKIPPED_PREFIX):].decode('utf-8'))
            elif line.startswith(TTest.FAILED_PREFIX):
                line = line.splitlines()[0]
                failed.append(line[len(TTest.FAILED_PREFIX):].decode('utf-8'))

        return missing, failed

    @staticmethod
    def _escapef(m):

    @staticmethod
    def _stringescape(s):
        return TTest.ESCAPESUB(TTest._escapef, s)

iolock = threading.RLock()
firstlock = threading.RLock()
firsterror = False

class TestResult(unittest._TextTestResult):
    """Holds results when executing via unittest."""
    # Don't worry too much about accessing the non-public _TextTestResult.
    # It is relatively common in Python testing tools.
    def __init__(self, options, *args, **kwargs):
        super(TestResult, self).__init__(*args, **kwargs)

        if options.color == 'auto':
            self.color = pygmentspresent and self.stream.isatty()
        elif options.color == 'never':
            self.color = False
        else: # 'always', for testing purposes
            self.color = pygmentspresent

    def onStart(self, test):
        """ Can be overriden by custom TestResult
        """

        with iolock:
            if self._options.nodiff:
                pass
            elif self._options.view:
                v = self._options.view
                subprocess.call(r'"%s" "%s" "%s"' %
                                (v, _strpath(test.refpath),
                                 _strpath(test.errpath)), shell=True)
            else:
                servefail, lines = getdiff(expected, got,
                                           test.refpath, test.errpath)
                self.stream.write('\n')
                for line in lines:
                    line = highlightdiff(line, self.color)
                    if PYTHON3:
                        self.stream.flush()

                        self.stream.write(line)
                        self.stream.flush()

                if servefail:
                    raise test.failureException(
                        'server failed to start (HGPORT=%s)' % test._startport)

            # handle interactive prompt without releasing iolock
            if self._options.interactive:
                if test.readrefout() != expected:
                    self.stream.write(
                        'Reference output has changed (run again to prompt '
                        'changes)')
                else:
                    self.stream.write('Accept this change? [n] ')
                    self.stream.flush()
                    answer = sys.stdin.readline().strip()
                    if answer.lower() in ('y', 'yes'):

        # os.times module computes the user time and system time spent by
        # child's processes along with real elapsed time taken by a process.
        # This module has one limitation. It can only work for Linux user
        # and not for Windows.
        test.started = os.times()
        if self._firststarttime is None: # thread racy but irrelevant
            self._firststarttime = test.started[4]

    def stopTest(self, test, interrupted=False):
        super(TestResult, self).stopTest(test)

        test.stopped = os.times()

        starttime = test.started
        endtime = test.stopped
        origin = self._firststarttime
        self.times.append((test.name,
                           endtime[2] - starttime[2], # user space CPU time
                           endtime[3] - starttime[3], # sys  space CPU time
                           endtime[4] - starttime[4], # real time
                           starttime[4] - origin, # start date in run context
                           endtime[4] - origin, # end date in run context
                           ))

        if interrupted:
            with iolock:
                self.stream.writeln('INTERRUPTED: %s (after %d seconds)' % (
                    test.name, self.times[-1][3]))

def getTestResult():
    """
    Returns the relevant test result
    """

        testresultmodule = __import__(os.environ["CUSTOM_TEST_RESULT"])
        return testresultmodule.TestResult
    else:
        return TestResult

class TestSuite(unittest.TestSuite):
    """Custom unittest TestSuite that knows how to execute Mercurial tests."""

    def __init__(self, testdir, jobs=1, whitelist=None, blacklist=None,
                 retest=False, keywords=None, loop=False, runs_per_test=1,
                 loadtest=None, showchannels=False,
                 *args, **kwargs):
        """Create a new instance that can run tests with a configuration.

        testdir specifies the directory where tests are executed from. This
        is typically the ``tests`` directory from Mercurial's source
        repository.

        # here instead of inside Test because it makes the running logic for
        # Test simpler.
        tests = []
        num_tests = [0]
        for test in self._tests:
            def get():
                num_tests[0] += 1
                if getattr(test, 'should_reload', False):
                    return self._loadtest(test, num_tests[0])
                return test
            if not os.path.exists(test.path):
                result.addSkip(test, "Doesn't exist")
                continue

            if not (self._whitelist and test.bname in self._whitelist):

            try:
                test(result)
                done.put(None)
            except KeyboardInterrupt:
                pass
            except: # re-raises
                done.put(('!', test, 'run-test raised an error, see traceback'))
                raise
            finally:
                try:
                    channels[channel] = ''

                with iolock:
                    sys.stdout.write(d + '  ')
                    sys.stdout.flush()
                for x in xrange(10):
                    if channels:
                        time.sleep(.1)
                count += 1

        stoppedearly = False

        if self._showchannels:

                if tests and not running == self._jobs:
                    test = tests.pop(0)
                    if self._loop:
                        if getattr(test, 'should_reload', False):
                            num_tests[0] += 1
                            tests.append(
                                self._loadtest(test, num_tests[0]))
                        else:
                            tests.append(test)
                    if self._jobs == 1:
                        job(test, result)
                    else:
                        t = threading.Thread(target=job, name=test.name,
                                             args=(test, result))
                        t.start()
                    running += 1

            # If we stop early we still need to wait on started tests to
            # finish. Otherwise, there is a race between the test completing

        channels = []

        return result

# Save the most recent 5 wall-clock runtimes of each test to a
# human-readable text file named .testtimes. Tests are sorted
# alphabetically, while times for each test are listed from oldest to
# newest.

def loadtimes(outputdir):
    times = []
    try:
        with open(os.path.join(outputdir, b'.testtimes')) as fp:
            for line in fp:
                m = re.match('(.*?) ([0-9. ]+)', line)
                times.append((m.group(1),
                              [float(t) for t in m.group(2).split()]))
    except IOError as err:
        if err.errno != errno.ENOENT:
            raise
    return times

def savetimes(outputdir, result):
    saved = dict(loadtimes(outputdir))
    maxruns = 5
    skipped = set([str(t[0]) for t in result.skipped])

        if test not in skipped:
            ts = saved.setdefault(test, [])
            ts.append(real)
            ts[:] = ts[-maxruns:]

    fd, tmpname = tempfile.mkstemp(prefix=b'.testtimes',
                                   dir=outputdir, text=True)
    with os.fdopen(fd, 'w') as fp:
        for name, ts in sorted(saved.items()):
            fp.write('%s %s\n' % (name, ' '.join(['%.3f' % (t,) for t in ts])))
    timepath = os.path.join(outputdir, b'.testtimes')
    try:

    try:
        os.rename(tmpname, timepath)
    except OSError:
        pass

class TextTestRunner(unittest.TextTestRunner):
    """Custom unittest test runner that uses appropriate settings."""

    def __init__(self, runner, *args, **kwargs):
        super(TextTestRunner, self).__init__(*args, **kwargs)

        self._runner = runner

        self._result = getTestResult()(self._runner.options, self.stream,
                                       self.descriptions, self.verbosity)

    def listtests(self, test):
        test = sorted(test, key=lambda t: t.name)

        self._result.onStart(test)

        with iolock:
            self.stream.writeln('')

            if not self._runner.options.noskips:
                for test, msg in sorted(self._result.skipped,
                                        key=lambda s: s[0].name):
                    formatted = 'Skipped %s: %s\n' % (test.name, msg)
                    msg = highlightmsg(formatted, self._result.color)
                    self.stream.write(msg)
            for test, msg in sorted(self._result.failures,
                                    key=lambda f: f[0].name):
                formatted = 'Failed %s: %s\n' % (test.name, msg)
                self.stream.write(highlightmsg(formatted, self._result.color))
            for test, msg in sorted(self._result.errors,
                                    key=lambda e: e[0].name):
                self.stream.writeln('Errored %s: %s' % (test.name, msg))

            if self._runner.options.xunit:
                with open(self._runner.options.xunit, "wb") as xuf:
                    self._writexunit(self._result, xuf)

            if failed and self._runner.options.known_good_rev:
                self._bisecttests(t for t, m in self._result.failures)
            self.stream.writeln(
                '# Ran %d tests, %d skipped, %d failed.'
                % (self._result.testsRun, skipped + ignored, failed))
            if failed:
                self.stream.writeln('python hash seed: %s' %
                    os.environ['PYTHONHASHSEED'])
            if self._runner.options.time:
                self.printtimes(self._result.times)

            if self._runner.options.exceptions:
                exceptions = aggregateexceptions(
                    os.path.join(self._runner._outputdir, b'exceptions'))

                self.stream.writeln('Exceptions Report:')
                self.stream.writeln('%d total from %d frames' %
                                    (exceptions['total'],
                                     len(exceptions['exceptioncounts'])))
                combined = exceptions['combined']
                for key in sorted(combined, key=combined.get, reverse=True):
                    frame, line, exc = key
                    totalcount, testcount, leastcount, leasttest = combined[key]

                    self.stream.writeln('%d (%d tests)\t%s: %s (%s - %d total)'
                                        % (totalcount,
                                           testcount,
                                           frame, exc,
                                           leasttest, leastcount))

            self.stream.flush()

        return self._result

    def _bisecttests(self, tests):
        bisectcmd = ['hg', 'bisect']
        bisectrepo = self._runner.options.bisect_repo
        if bisectrepo:
            bisectcmd.extend(['-R', os.path.abspath(bisectrepo)])
        def pread(args):
            env = os.environ.copy()
            env['HGPLAIN'] = '1'
            p = subprocess.Popen(args, stderr=subprocess.STDOUT,
                                 stdout=subprocess.PIPE, env=env)
            data = p.stdout.read()
            p.wait()
            return data
        for test in tests:
            pread(bisectcmd + ['--reset']),
            pread(bisectcmd + ['--bad', '.'])
            pread(bisectcmd + ['--good', self._runner.options.known_good_rev])
            # TODO: we probably need to forward more options
            # that alter hg's behavior inside the tests.
            opts = ''
            withhg = self._runner.options.with_hg
            if withhg:
                opts += ' --with-hg=%s ' % shellquote(_strpath(withhg))
            rtc = '%s %s %s %s' % (sysexecutable, sys.argv[0], opts,
                                   test)
            data = pread(bisectcmd + ['--command', rtc])
            m = re.search(
                (br'\nThe first (?P<goodbad>bad|good) revision '
                 br'is:\nchangeset: +\d+:(?P<node>[a-f0-9]+)\n.*\n'
                 br'summary: +(?P<summary>[^\n]+)\n'),
                data, (re.MULTILINE | re.DOTALL))
            if m is None:
                self.stream.writeln(
                    'Failed to identify failure point for %s' % test)
                continue
            dat = m.groupdict()
            verb = 'broken' if dat['goodbad'] == b'bad' else 'fixed'
            self.stream.writeln(
                '%s %s by %s (%s)' % (
                    test, verb, dat['node'].decode('ascii'),
                    dat['summary'].decode('utf8', 'ignore')))

    def printtimes(self, times):
        # iolock held by run
        self.stream.writeln('# Producing time report')
        times.sort(key=lambda t: (t[3]))
        cols = '%7.3f %7.3f %7.3f %7.3f %7.3f   %s'
        self.stream.writeln('%-7s %-7s %-7s %-7s %-7s   %s' %
                            ('start', 'end', 'cuser', 'csys', 'real', 'Test'))
        for tdata in times:
            test = tdata[0]
            cuser, csys, real, start, end = tdata[1:6]
            self.stream.writeln(cols % (start, end, cuser, csys, real, test))

    def _writexunit(result, outf):
        # See http://llg.cubic.org/docs/junit/ for a reference.
        timesd = dict((t[0], t[3]) for t in result.times)
        doc = minidom.Document()
        s = doc.createElement('testsuite')
        s.setAttribute('errors', "0") # TODO
        s.setAttribute('failures', str(len(result.failures)))
        s.setAttribute('name', 'run-tests')
        s.setAttribute('skipped', str(len(result.skipped) +
                                      len(result.ignored)))
        s.setAttribute('tests', str(result.testsRun))
        doc.appendChild(s)
        for tc in result.successes:
            t = doc.createElement('testcase')
            t.setAttribute('name', tc.name)

        for tdata in result.times:
            test = tdata[0]
            timesd[test] = tdata[1:]

        outcome = {}
        groups = [('success', ((tc, None)
                   for tc in result.successes)),
                  ('failure', result.failures),
                  ('skip', result.skipped)]
        for res, testcases in groups:
            for tc, __ in testcases:
                if tc.name in timesd:
                    diff = result.faildata.get(tc.name, b'')
                    try:
                        diff = diff.decode('unicode_escape')
                    except UnicodeDecodeError as e:
                        diff = '%r decoding diff, sorry' % e
                    tres = {'result': res,
                            'time': ('%0.3f' % timesd[tc.name][2]),
                            'cuser': ('%0.3f' % timesd[tc.name][0]),
                            'csys': ('%0.3f' % timesd[tc.name][1]),
                            'start': ('%0.3f' % timesd[tc.name][3]),
                            'end': ('%0.3f' % timesd[tc.name][4]),
                            'diff': diff,
                            }
                else:
                    # blacklisted test
                    tres = {'result': res}

                outcome[tc.name] = tres
        jsonout = json.dumps(outcome, sort_keys=True, indent=4,
                             separators=(',', ': '))
        outf.writelines(("testreport =", jsonout))

def sorttests(testdescs, previoustimes, shuffle=False):
    """Do an in-place sort of tests."""
    if shuffle:
        random.shuffle(testdescs)
        return

    if previoustimes:
        def sortkey(f):
            f = f['path']
            if f in previoustimes:
                # Use most recent time as estimate
                return -previoustimes[f][-1]
            else:
                # Default to a rather arbitrary value of 1 second for new tests
                return -1.0
    else:
        # keywords for slow tests
        slow = {b'svn': 10,
                b'cvs': 10,
                b'hghave': 10,
                b'largefiles-update': 10,
                b'run-tests': 10,
                b'corruption': 10,
                b'race': 10,
                b'i18n': 10,
                b'check': 100,
                b'gendoc': 100,
                b'contrib-perf': 200,
                b'merge-combination': 100,
                }
        perf = {}

        def sortkey(f):
            # run largest tests first, as they tend to take the longest
            f = f['path']

                perf[f] = val / 1000.0
                return perf[f]

    testdescs.sort(key=sortkey)

class TestRunner(object):
    """Holds context for executing tests.

    Tests rely on a lot of state. This object holds it for them.
    """

            self._checktools()
            testdescs = self.findtests(tests)
            if options.profile_runner:
                import statprof
                statprof.start()
            result = self._run(testdescs)
            if options.profile_runner:
                statprof.stop()
                statprof.display()

                # without this, we get the default temp dir location, but
                # in all lowercase, which causes troubles with paths (issue3490)
                d = osenvironb.get(b'TMP', None)
            tmpdir = tempfile.mkdtemp(b'', b'hgtests.', d)

        self._hgtmp = osenvironb[b'HGTMP'] = (
            os.path.realpath(tmpdir))

        if self.options.with_hg:
            self._installdir = None
            whg = self.options.with_hg
            self._bindir = os.path.dirname(os.path.realpath(whg))

                os.unlink(os.path.join(exceptionsdir, f))

            osenvironb[b'HGEXCEPTIONSDIR'] = exceptionsdir
            logexceptions = os.path.join(self._testdir, b'logexceptions.py')
            self.options.extra_config_opt.append(
                'extensions.logexceptions=%s' % logexceptions.decode('utf-8'))

        vlog("# Using TESTDIR", self._testdir)
        vlog("# Using RUNTESTDIR", osenvironb[b'RUNTESTDIR'])
        vlog("# Using HGTMP", self._hgtmp)
        vlog("# Using PATH", os.environ["PATH"])

        vlog("# Writing to directory", self._outputdir)

        try:
            return self._runtests(testdescs) or 0
        finally:
            time.sleep(.1)
            self._cleanup()

    def findtests(self, args):
        """Finds possible test files from arguments.

        If you wish to inject custom tests into the test harness, this would
        be a good function to monkeypatch or override in a derived class.
        """
        if not args:
            if self.options.changed:
                proc = Popen4(b'hg st --rev "%s" -man0 .' %
                              _bytespath(self.options.changed), None, 0)
                stdout, stderr = proc.communicate()
                args = stdout.strip(b'\0').split(b'\0')
            else:
                args = os.listdir(b'.')

            else:
                expanded_args.append(arg)
        args = expanded_args

        testcasepattern = re.compile(
            br'([\w-]+\.t|py)(?:#([a-zA-Z0-9_\-\.#]+))')
        tests = []
        for t in args:
            case = []

            if not (os.path.basename(t).startswith(b'test-')
                    and (t.endswith(b'.py') or t.endswith(b'.t'))):

                m = testcasepattern.match(os.path.basename(t))
                if m is not None:
                    t_basename, casestr = m.groups()
                    t = os.path.join(os.path.dirname(t), t_basename)

            if t.endswith(b'.t'):
                # .t file may contain multiple test cases
                casedimensions = parsettestcases(t)
                if casedimensions:
                    cases = []
                    def addcases(case, casedimensions):
                        if not casedimensions:
                            cases.append(case)
                        else:
                            for c in casedimensions[0]:
                                addcases(case + [c], casedimensions[1:])
                    addcases([], casedimensions)
                    if case and case in cases:
                        cases = [case]
                    elif case:
                        # Ignore invalid cases

            failed = False
            kws = self.options.keywords
            if kws is not None and PYTHON3:
                kws = kws.encode('utf-8')

            suite = TestSuite(self._testdir,
                              jobs=jobs,
                              whitelist=self.options.whitelisted,
                              blacklist=self.options.blacklist,
                              retest=self.options.retest,
                              keywords=kws,
                              loop=self.options.loop,
                              runs_per_test=self.options.runs_per_test,
                              showchannels=self.options.showchannels,
                              tests=tests, loadtest=_reloadtest)
            verbosity = 1
            if self.options.list_tests:
                verbosity = 0
            elif self.options.verbose:
                verbosity = 2

                    self._usecorrectpython()
                if self.options.chg:
                    assert self._installdir
                    self._installchg()

                log('running %d tests using %d parallel processes' % (
                    num_tests, jobs))

                result = runner.run(suite)

            if result.failures or result.errors:
                failed = True

        if failed:
            return 1

    def _getport(self, count):
        port = self._ports.get(count) # do we have a cached entry?
        if port is None:
            portneeded = 3
            # above 100 tries we just give up and let test reports failure
            for tries in xrange(100):
                allfree = True

        tmpdir = os.path.join(self._hgtmp, b'child%d' % count)

        # extra keyword parameters. 'case' is used by .t tests
        kwds = dict((k, testdesc[k]) for k in ['case'] if k in testdesc)

        t = testcls(refpath, self._outputdir, tmpdir,
                    keeptmpdir=self.options.keep_tmpdir,
                    debug=self.options.debug,
                    first=self.options.first,
                    timeout=self.options.timeout,
                    startport=self._getport(count),
                    extraconfigopts=self.options.extra_config_opt,
                    py3warnings=self.options.py3_warnings,
                    shell=self.options.shell,
                    hgcommand=self._hgcommand,
                    usechg=bool(self.options.with_chg or self.options.chg),
                    useipv6=useipv6, **kwds)
        t.should_reload = True
        return t

    def _cleanup(self):
        """Clean up state from this test invocation."""

        pyexename = sys.platform == 'win32' and b'python.exe' or b'python'

        # os.symlink() is a thing with py3 on Windows, but it requires
        # Administrator rights.
        if getattr(os, 'symlink', None) and os.name != 'nt':
            vlog("# Making python executable in test path a symlink to '%s'" %
                 sysexecutable)
            mypython = os.path.join(self._tmpbindir, pyexename)
            try:
                if os.readlink(mypython) == sysexecutable:
                    return
                os.unlink(mypython)

                    # child processes may race, which is harmless
                    if err.errno != errno.EEXIST:
                        raise
        else:
            exedir, exename = os.path.split(sysexecutable)
            vlog("# Modifying search path to find %s as %s in '%s'" %
                 (exename, pyexename, exedir))
            path = os.environ['PATH'].split(os.pathsep)
            while exedir in path:
                path.remove(exedir)
            os.environ['PATH'] = os.pathsep.join([exedir] + path)
            if not self._findprogram(pyexename):

            # The --home="" trick works only on OS where os.sep == '/'
            # because of a distutils convert_path() fast-path. Avoid it at
            # least on Windows for now, deal with .pydistutils.cfg bugs
            # when they happen.
            nohome = b''
        cmd = (b'"%(exe)s" setup.py %(pure)s clean --all'
               b' build %(compiler)s --build-base="%(base)s"'
               b' install --force --prefix="%(prefix)s"'
               b' --install-lib="%(libdir)s"'
               b' --install-scripts="%(bindir)s" %(nohome)s >%(logfile)s 2>&1'
               % {b'exe': exe, b'pure': pure,
                  b'compiler': compiler,
                  b'base': os.path.join(self._hgtmp, b"build"),
                  b'prefix': self._installdir, b'libdir': self._pythondir,
                  b'bindir': self._bindir,
                  b'nohome': nohome, b'logfile': installerrs})

        # setuptools requires install directories to exist.
        def makedirs(p):
            try:
                os.makedirs(p)
            except OSError as e:
                if e.errno != errno.EEXIST:
                    raise
        makedirs(self._pythondir)
        makedirs(self._bindir)

        vlog("# Running", cmd)
        if subprocess.call(_strpath(cmd), shell=True) == 0:

            # hg.bat expects to be put in bin/scripts while run-tests.py
            # installation layout put it in bin/ directly. Fix it
            with open(hgbat, 'rb') as f:
                data = f.read()
            if br'"%~dp0..\python" "%~dp0hg" %*' in data:
                data = data.replace(br'"%~dp0..\python" "%~dp0hg" %*',
                                    b'"%~dp0python" "%~dp0hg" %*')
                with open(hgbat, 'wb') as f:
                    f.write(data)
            else:
                print('WARNING: cannot fix hg.bat reference to python.exe')

            os.environ['COVERAGE_DIR'] = covdir

    def _checkhglib(self, verb):
        """Ensure that the 'mercurial' package imported by python is
        the one we expect it to be.  If not, print a warning to stderr."""
        if ((self._bindir == self._pythondir) and
            (self._bindir != self._tmpbindir)):
            # The pythondir has been inferred from --with-hg flag.
            # We cannot expect anything sensible here.
            return
        expecthg = os.path.join(self._pythondir, b'mercurial')
        actualhg = self._gethgpath()
        if os.path.abspath(actualhg) != os.path.abspath(expecthg):
            sys.stderr.write('warning: %s with unexpected mercurial lib: %s\n'
                             '         (expected %s)\n'
                             % (verb, actualhg, expecthg))
    def _gethgpath(self):
        """Return the path to the mercurial package that is actually found by
        the current Python interpreter."""
        if self._hgpath is not None:
            return self._hgpath

    def _installchg(self):
        """Install chg into the test environment"""
        vlog('# Performing temporary installation of CHG')
        assert os.path.dirname(self._bindir) == self._installdir
        assert self._hgroot, 'must be called after _installhg()'
        cmd = (b'"%(make)s" clean install PREFIX="%(prefix)s"'
               % {b'make': b'make',  # TODO: switch by option or environment?
                  b'prefix': self._installdir})
        cwd = os.path.join(self._hgroot, b'contrib', b'chg')
        vlog("# Running", cmd)
        proc = subprocess.Popen(cmd, shell=True, cwd=cwd,
                                stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                stderr=subprocess.STDOUT)
        out, _err = proc.communicate()
        if proc.returncode != 0:
            if PYTHON3:
                sys.stdout.buffer.write(out)
            else:

            sys.exit(1)

    def _outputcoverage(self):
        """Produce code coverage output."""
        import coverage
        coverage = coverage.coverage

        vlog('# Producing coverage report')
        # chdir is the easiest way to get short, relative paths in the
        # output.

                p += b'.exe'
            found = self._findprogram(p)
            if found:
                vlog("# Found prerequisite", p, "at", found)
            else:
                print("WARNING: Did not find prerequisite tool: %s " %
                      p.decode("utf-8"))

def aggregateexceptions(path):
    exceptioncounts = collections.Counter()
    testsbyfailure = collections.defaultdict(set)
    failuresbytest = collections.defaultdict(set)

    # Create a combined counter so we can sort by total occurrences and
    # impacted tests.
    combined = {}
    for key in exceptioncounts:
        combined[key] = (exceptioncounts[key],
                         len(testsbyfailure[key]),
                         leastfailing[key][0],
                         leastfailing[key][1])

    return {
        'exceptioncounts': exceptioncounts,
        'total': sum(exceptioncounts.values()),
        'combined': combined,
        'leastfailing': leastfailing,
        'byfailure': testsbyfailure,
        'bytest': failuresbytest,
    }

if __name__ == '__main__':
    runner = TestRunner()

    try:
        import msvcrt
        msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)
        msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)
        msvcrt.setmode(sys.stderr.fileno(), os.O_BINARY)
    except ImportError:
        pass

43076:2372284d9457

except ImportError:
    import queue

try:
    import shlex

    shellquote = shlex.quote
except (ImportError, AttributeError):
    import pipes

    shellquote = pipes.quote

processlock = threading.Lock()

pygmentspresent = False
# ANSI color is unsupported prior to Windows 10
if 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 = {

                (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')

        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

        else:
            raise
    else:
        return False


# useipv6 will be set by parseargs
useipv6 = None


def checkportisavailable(port):
    """return true if a port seems free to bind on localhost"""
    if useipv6:
        family = socket.AF_INET6

        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.executable
elif os.environ.get('PYTHONEXECUTABLE'):
    sysexecutable = os.environ['PYTHONEXECUTABLE']

    '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:

                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.
    """

    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']
    )
    harness.add_argument(
        '--showchannels', action='store_true', help='show scheduling channels'
    )
    harness.add_argument(
        "--slowtimeout",
        type=int,
        help="kill errant slow tests after SLOWTIMEOUT seconds"
        " (default: $%s or %d)" % defaults['slowtimeout'],
    )
    harness.add_argument(
        "-t",
        "--timeout",
        type=int,
        help="kill errant tests after TIMEOUT seconds"
        " (default: $%s or %d)" % defaults['timeout'],
    )
    harness.add_argument(
        "--tmpdir",
        help="run tests in the given temporary directory"
        " (implies --keep-tmpdir)",
    )
    harness.add_argument(
        "-v", "--verbose", action="store_true", help="output verbose messages"
    )

    hgconf = parser.add_argument_group('Mercurial Configuration')
    hgconf.add_argument(
        "--chg",
        action="store_true",
        help="install and use chg wrapper in place of hg",
    )
    hgconf.add_argument("--compiler", help="compiler to build with")
    hgconf.add_argument(
        '--extra-config-opt',
        action="append",
        default=[],
        help='set the given config opt in the test hgrc',
    )
    hgconf.add_argument(
        "-l",
        "--local",
        action="store_true",
        help="shortcut for --with-hg=<testdir>/../hg, "
        "and --with-chg=<testdir>/../contrib/chg/chg if --chg is set",
    )
    hgconf.add_argument(
        "--ipv6",
        action="store_true",
        help="prefer IPv6 to IPv4 for network related tests",
    )
    hgconf.add_argument(
        "--pure",
        action="store_true",
        help="use pure Python code instead of C extensions",
    )
    hgconf.add_argument(
        "-3",
        "--py3-warnings",
        action="store_true",
        help="enable Py3k warnings on Python 2.7+",
    )
    hgconf.add_argument(
        "--with-chg",
        metavar="CHG",
        help="use specified chg wrapper in place of hg",
    )
    hgconf.add_argument(
        "--with-hg",
        metavar="HG",
        help="test using specified hg script rather than a "
        "temporary installation",
    )

    reporting = parser.add_argument_group('Results Reporting')
    reporting.add_argument(
        "-C",
        "--annotate",
        action="store_true",
        help="output files annotated with coverage",
    )
    reporting.add_argument(
        "--color",
        choices=["always", "auto", "never"],
        default=os.environ.get('HGRUNTESTSCOLOR', 'auto'),
        help="colorisation: always|auto|never (default: auto)",
    )
    reporting.add_argument(
        "-c",
        "--cover",
        action="store_true",
        help="print a test coverage report",
    )
    reporting.add_argument(
        '--exceptions',
        action='store_true',
        help='log all exceptions and generate an exception report',
    )
    reporting.add_argument(
        "-H",
        "--htmlcov",
        action="store_true",
        help="create an HTML report of the coverage of the files",
    )
    reporting.add_argument(
        "--json",
        action="store_true",
        help="store test result data in 'report.json' file",
    )
    reporting.add_argument(
        "--outputdir",
        help="directory to write error logs to (default=test directory)",
    )
    reporting.add_argument(
        "-n", "--nodiff", action="store_true", help="skip showing test changes"
    )
    reporting.add_argument(
        "-S",
        "--noskips",
        action="store_true",
        help="don't report skip tests verbosely",
    )
    reporting.add_argument(
        "--time", action="store_true", help="time how long each test takes"
    )
    reporting.add_argument("--view", help="external diff viewer")
    reporting.add_argument(
        "--xunit", help="record xunit results at specified path"
    )

    for option, (envvar, default) in defaults.items():
        defaults[option] = type(default)(os.environ.get(envvar, default))
    parser.set_defaults(**defaults)

    return parser


def parseargs(args, parser):
    """Parse arguments with our OptionParser and validate results."""
    options = parser.parse_args(args)

        if options.chg:
            pathandattrs.append((b'contrib/chg/chg', 'with_chg'))
        for relpath, attr in pathandattrs:
            binpath = os.path.join(reporootdir, relpath)
            if os.name != 'nt' and not os.access(binpath, os.X_OK):
                parser.error(
                    '--local specified, but %r not found or '
                    'not executable' % binpath
                )
            setattr(options, attr, _strpath(binpath))

    if options.with_hg:
        options.with_hg = canonpath(_bytespath(options.with_hg))
        if not (
            os.path.isfile(options.with_hg)
            and os.access(options.with_hg, os.X_OK)
        ):
            parser.error('--with-hg must specify an executable hg script')
        if os.path.basename(options.with_hg) not in [b'hg', b'hg.exe']:
            sys.stderr.write('warning: --with-hg should specify an hg script\n')
            sys.stderr.flush()

    if (options.chg or options.with_chg) and os.name == 'nt':
        parser.error('chg does not work on %s' % os.name)
    if options.with_chg:
        options.chg = False  # no installation to temporary location
        options.with_chg = canonpath(_bytespath(options.with_chg))
        if not (
            os.path.isfile(options.with_chg)
            and os.access(options.with_chg, os.X_OK)
        ):
            parser.error('--with-chg must specify a chg executable')
    if options.chg and options.with_hg:
        # chg shares installation location with hg
        parser.error(
            '--chg does not work when --with-hg is specified '
            '(use --with-chg instead)'
        )

    if options.color == 'always' and not pygmentspresent:
        sys.stderr.write(
            'warning: --color=always ignored because '
            'pygments is not installed\n'
        )

    if options.bisect_repo and not options.known_good_rev:
        parser.error("--bisect-repo cannot be used without --known-good-rev")

    global useipv6
    if options.ipv6:
        useipv6 = checksocketfamily('AF_INET6')
    else:
        # only use IPv6 if IPv4 is unavailable and IPv6 is available
        useipv6 = (not checksocketfamily('AF_INET')) and checksocketfamily(
            'AF_INET6'
        )

    options.anycoverage = options.cover or options.annotate or options.htmlcov
    if options.anycoverage:
        try:
            import coverage

            covver = version.StrictVersion(coverage.__version__).version
            if covver < (3, 3):
                parser.error('coverage options require coverage 3.3 or later')
        except ImportError:
            parser.error('coverage options now require the coverage package')

    if options.anycoverage and options.local:
        # this needs some path mangling somewhere, I guess
        parser.error(
            "sorry, coverage options do not work when --local " "is specified"
        )

    if options.anycoverage and options.with_hg:
        parser.error(
            "sorry, coverage options do not work when --with-hg " "is specified"
        )

    global verbose
    if options.verbose:
        verbose = ''

        parser.error('--jobs must be positive')
    if options.interactive and options.debug:
        parser.error("-i/--interactive and -d/--debug are incompatible")
    if options.debug:
        if options.timeout != defaults['timeout']:
            sys.stderr.write('warning: --timeout option ignored with --debug\n')
        if options.slowtimeout != defaults['slowtimeout']:
            sys.stderr.write(
                'warning: --slowtimeout option ignored with --debug\n'
            )
        options.timeout = 0
        options.slowtimeout = 0
    if options.py3_warnings:
        if PYTHON3:
            parser.error('--py3-warnings can only be used on Python 2.7')

    if options.blacklist:
        options.blacklist = parselistfiles(options.blacklist, 'blacklist')
    if options.whitelist:
        options.whitelisted = parselistfiles(options.whitelist, 'whitelist')

    if options.showchannels:
        options.nodiff = True

    return options


def rename(src, dst):
    """Like os.rename(), trade atomicity and opened files friendliness
    for existing destination support.
    """
    shutil.copy(src, dst)
    os.remove(src)


def makecleanable(path):
    """Try to fix directory permission recursively so that the entire tree
    can be deleted"""
    for dirpath, dirnames, _filenames in os.walk(path, topdown=True):

            try:
                os.chmod(p, os.stat(p).st_mode & 0o777 | 0o700)  # chmod u+rwx
            except OSError:
                pass


_unified_diff = difflib.unified_diff
if PYTHON3:
    import functools

    _unified_diff = functools.partial(difflib.diff_bytes, difflib.unified_diff)


def getdiff(expected, output, ref, err):
    servefail = False
    lines = []
    for line in _unified_diff(expected, output, ref, err):

            line = line.replace(b'\\', b'/')
            if line.endswith(b' \n'):
                line = line[:-2] + b'\n'
        lines.append(line)
        if not servefail and line.startswith(
            b'+  abort: child process failed to start'
        ):
            servefail = True

    return servefail, lines


verbose = False


def vlog(*msg):
    """Log only when in verbose mode."""
    if verbose is False:
        return

    return log(*msg)


# Bytes that break XML even in a CDATA block: control characters 0-31
# sans \t, \n and \r
CDATA_EVIL = re.compile(br"[\000-\010\013\014\016-\037]")

# Match feature conditionalized output lines in the form, capturing the feature
# list in group 2, and the preceeding line output in group 1:
#
#   output..output (feature !)\n
optline = re.compile(br'(.*) \((.+?) !\)\n$')


def cdatasafe(data):
    """Make a string safe to include in a CDATA block.

    Certain control characters are illegal in a CDATA block, and
    there's no way to include a ]]> in a CDATA either. This function
    replaces illegal bytes with ? and adds a space between the ]] so
    that it won't break the CDATA block.
    """
    return CDATA_EVIL.sub(b'?', data).replace(b']]>', b'] ]>')


def log(*msg):
    """Log something to stdout.

    Arguments are strings to print.

        for m in msg:
            print(m, end=' ')
        print()
        sys.stdout.flush()


def highlightdiff(line, color):
    if not color:
        return line
    assert pygmentspresent
    return pygments.highlight(
        line.decode('latin1'), difflexer, terminal256formatter
    ).encode('latin1')


def highlightmsg(msg, color):
    if not color:
        return msg
    assert pygmentspresent
    return pygments.highlight(msg, runnerlexer, runnerformatter)


def terminate(proc):
    """Terminate subprocess"""
    vlog('# Terminating process %d' % proc.pid)
    try:
        proc.terminate()
    except OSError:
        pass


def killdaemons(pidfile):
    import killdaemons as killmod

    return killmod.killdaemons(pidfile, tryhard=False, remove=True, logfn=vlog)


class Test(unittest.TestCase):
    """Encapsulates a single, runnable test.

    While this class conforms to the unittest.TestCase API, it differs in that

    """

    # Status code reserved for skipped tests (used by hghave).
    SKIPPED_STATUS = 80

    def __init__(
        self,
        path,
        outputdir,
        tmpdir,
        keeptmpdir=False,
        debug=False,
        first=False,
        timeout=None,
        startport=None,
        extraconfigopts=None,
        py3warnings=False,
        shell=None,
        hgcommand=None,
        slowtimeout=None,
        usechg=False,
        useipv6=False,
    ):
        """Create a test from parameters.

        path is the full path to the file defining the test.

        tmpdir is the main temporary directory to use for this test.

    def readrefout(self):
        """read reference output"""
        # If we're not in --debug mode and reference output file exists,
        # check test output against it.
        if self._debug:
            return None  # to match "out is None"
        elif os.path.exists(self.refpath):
            with open(self.refpath, 'rb') as f:
                return f.read().splitlines(True)
        else:
            return []

                # file.
                if e.errno != errno.ENOENT:
                    raise

        if self._usechg:
            self._chgsockdir = os.path.join(
                self._threadtmp, b'%s.chgsock' % name
            )
            os.mkdir(self._chgsockdir)

    def run(self, result):
        """Run this test and report results against a TestResult instance."""
        # This function is extremely similar to unittest.TestCase.run(). Once

            return 'returned error code %d' % ret

        self._skipped = False

        if ret == self.SKIPPED_STATUS:
            if out is None:  # Debug mode, nothing to parse.
                missing = ['unknown']
                failed = None
            else:
                missing, failed = TTest.parsehghaveoutput(out)

            self.fail('timed out')
        elif ret is False:
            self.fail('no result code from test')
        elif out != self._refout:
            # Diff generation may rely on written .err file.
            if (
                (ret != 0 or out != self._refout)
                and not self._skipped
                and not self._debug
            ):
                with open(self.errpath, 'wb') as f:
                    for line in out:
                        f.write(line)

            # The result object handles diff calculation for us.

        for entry in self._daemonpids:
            killdaemons(entry)
        self._daemonpids = []

        if self._keeptmpdir:
            log(
                '\nKeeping testtmp dir: %s\nKeeping threadtmp dir: %s'
                % (
                    self._testtmp.decode('utf-8'),
                    self._threadtmp.decode('utf-8'),
                )
            )
        else:
            try:
                shutil.rmtree(self._testtmp)
            except OSError:
                # unreadable directory may be left in $TESTTMP; fix permission

        if self._usechg:
            # chgservers will stop automatically after they find the socket
            # files are deleted
            shutil.rmtree(self._chgsockdir, True)

        if (
            (self._ret != 0 or self._out != self._refout)
            and not self._skipped
            and not self._debug
            and self._out
        ):
            with open(self.errpath, 'wb') as f:
                for line in self._out:
                    f.write(line)

        vlog("# Ret was:", self._ret, '(%s)' % self.name)

            self._portmap(0),
            self._portmap(1),
            self._portmap(2),
            (br'([^0-9])%s' % re.escape(self._localip()), br'\1$LOCALIP'),
            (br'\bHG_TXNID=TXN:[a-f0-9]{40}\b', br'HG_TXNID=TXN:$ID$'),
        ]
        r.append((self._escapepath(self._testtmp), b'$TESTTMP'))

        replacementfile = os.path.join(self._testdir, b'common-pattern.py')

        if os.path.exists(replacementfile):

                    r.append(value)
        return r

    def _escapepath(self, p):
        if os.name == 'nt':
            return b''.join(
                c.isalpha()
                and b'[%s%s]' % (c.lower(), c.upper())
                or c in b'/\\'
                and br'[/\\]'
                or c.isdigit()
                and c
                or b'\\' + c
                for c in [p[i : i + 1] for i in range(len(p))]
            )
        else:
            return re.escape(p)

    def _localip(self):

                    continue
                envf.write('unset %s\n' % (name,))

    def _getenv(self):
        """Obtain environment variables to use during test execution."""

        def defineport(i):
            offset = '' if i == 0 else '%s' % i
            env["HGPORT%s" % offset] = '%s' % (self._startport + i)

        env = os.environ.copy()
        env['PYTHONUSERBASE'] = sysconfig.get_config_var('userbase') or ''
        env['HGEMITWARNINGS'] = '1'
        env['TESTTMP'] = _strpath(self._testtmp)
        env['TESTNAME'] = self.name

        # This number should match portneeded in _getport
        for port in xrange(3):
            # This list should be parallel to _portmap in _getreplacements
            defineport(port)
        env["HGRCPATH"] = _strpath(os.path.join(self._threadtmp, b'.hgrc'))
        env["DAEMON_PIDS"] = _strpath(
            os.path.join(self._threadtmp, b'daemon.pids')
        )
        env["HGEDITOR"] = (
            '"' + sysexecutable + '"' + ' -c "import sys; sys.exit(0)"'
        )
        env["HGUSER"] = "test"
        env["HGENCODING"] = "ascii"
        env["HGENCODINGMODE"] = "strict"
        env["HGHOSTNAME"] = "test-hostname"
        env['HGIPV6'] = str(int(self._useipv6))
        # See contrib/catapipe.py for how to use this functionality.
        if 'HGTESTCATAPULTSERVERPIPE' not in env:
            # If we don't have HGTESTCATAPULTSERVERPIPE explicitly set, pull the
            # non-test one in as a default, otherwise set to devnull
            env['HGTESTCATAPULTSERVERPIPE'] = env.get(
                'HGCATAPULTSERVERPIPE', os.devnull
            )

        extraextensions = []
        for opt in self._extraconfigopts:
            section, key = opt.encode('utf-8').split(b'.', 1)
            if section != 'extensions':

            hgrc.write(b'[defaults]\n')
            hgrc.write(b'[devel]\n')
            hgrc.write(b'all-warnings = true\n')
            hgrc.write(b'default-date = 0 0\n')
            hgrc.write(b'[largefiles]\n')
            hgrc.write(
                b'usercache = %s\n'
                % (os.path.join(self._testtmp, b'.cache/largefiles'))
            )
            hgrc.write(b'[lfs]\n')
            hgrc.write(
                b'usercache = %s\n'
                % (os.path.join(self._testtmp, b'.cache/lfs'))
            )
            hgrc.write(b'[web]\n')
            hgrc.write(b'address = localhost\n')
            hgrc.write(b'ipv6 = %s\n' % str(self._useipv6).encode('ascii'))
            hgrc.write(b'server-header = testing stub value\n')

            for opt in self._extraconfigopts:
                section, key = opt.encode('utf-8').split(b'.', 1)
                assert b'=' in key, (
                    'extra config opt %s must ' 'have an = for assignment' % opt
                )
                hgrc.write(b'[%s]\n%s\n' % (section, key))

    def fail(self, msg):
        # unittest differentiates between errored and failed.
        # Failed is denoted by AssertionError (by default at least).

        stderr).

        Return a tuple (exitcode, output). output is None in debug mode.
        """
        if self._debug:
            proc = subprocess.Popen(
                _strpath(cmd), shell=True, cwd=_strpath(self._testtmp), env=env
            )
            ret = proc.wait()
            return (ret, None)

        proc = Popen4(cmd, self._testtmp, self._timeout, env)

        def cleanup():
            terminate(proc)
            ret = proc.wait()
            if ret == 0:
                ret = signal.SIGTERM << 8

        if normalizenewlines:
            output = output.replace(b'\r\n', b'\n')

        return ret, output.splitlines(True)


class PythonTest(Test):
    """A Python-based test."""

    @property
    def refpath(self):

        py3switch = self._py3warnings and b' -3' or b''
        # Quote the python(3) executable for Windows
        cmd = b'"%s"%s "%s"' % (PYTHON, py3switch, self.path)
        vlog("# Running", cmd)
        normalizenewlines = os.name == 'nt'
        result = self._runcommand(cmd, env, normalizenewlines=normalizenewlines)
        if self._aborted:
            raise KeyboardInterrupt()

        return result


# Some glob patterns apply only in some circumstances, so the script
# might want to remove (glob) annotations that otherwise should be
# retained.
checkcodeglobpats = [

WARN_YES = 2
WARN_NO = 3

MARK_OPTIONAL = b" (?)\n"


def isoptional(line):
    return line.endswith(MARK_OPTIONAL)


class TTest(Test):
    """A "t test" is a test backed by a .t file."""

    SKIPPED_PREFIX = b'skipped: '

            return self._have.get(allreqs)

        # TODO do something smarter when all other uses of hghave are gone.
        runtestdir = os.path.abspath(os.path.dirname(_bytespath(__file__)))
        tdir = runtestdir.replace(b'\\', b'/')
        proc = Popen4(
            b'%s -c "%s/hghave %s"' % (self._shell, tdir, allreqs),
            self._testtmp,
            0,
            self._getenv(),
        )
        stdout, stderr = proc.communicate()
        ret = proc.wait()
        if wifexited(ret):
            ret = os.WEXITSTATUS(ret)
        if ret == 2:

    def _parsetest(self, lines):
        # We generate a shell script which outputs unique markers to line
        # up script results with our source. These markers include input
        # line number and the last return code.
        salt = b"SALT%d" % time.time()

        def addsalt(line, inpython):
            if inpython:
                script.append(b'%s %d 0\n' % (salt, line))
            else:
                script.append(b'echo %s %d $?\n' % (salt, line))

        activetrace = []
        session = str(uuid.uuid4())
        if PYTHON3:
            session = session.encode('ascii')
        hgcatapult = os.getenv('HGTESTCATAPULTSERVERPIPE') or os.getenv(
            'HGCATAPULTSERVERPIPE'
        )

        def toggletrace(cmd=None):
            if not hgcatapult or hgcatapult == os.devnull:
                return

            if activetrace:
                script.append(
                    b'echo END %s %s >> "$HGTESTCATAPULTSERVERPIPE"\n'
                    % (session, activetrace[0])
                )
            if cmd is None:
                return

            if isinstance(cmd, str):
                quoted = shellquote(cmd.strip())
            else:
                quoted = shellquote(cmd.strip().decode('utf8')).encode('utf8')
            quoted = quoted.replace(b'\\', b'\\\\')
            script.append(
                b'echo START %s %s >> "$HGTESTCATAPULTSERVERPIPE"\n'
                % (session, quoted)
            )
            activetrace[0:] = [quoted]

        script = []

        # After we run the shell script, we re-unify the script output

                    after.setdefault(pos, []).append('  !!! missing #if\n')
                skipping = None
                after.setdefault(pos, []).append(l)
            elif skipping:
                after.setdefault(pos, []).append(l)
            elif l.startswith(b'  >>> '):  # python inlines
                after.setdefault(pos, []).append(l)
                prepos = pos
                pos = n
                if not inpython:
                    # We've just entered a Python block. Add the header.
                    inpython = True
                    addsalt(prepos, False)  # Make sure we report the exit code.
                    script.append(b'"%s" -m heredoctest <<EOF\n' % PYTHON)
                addsalt(n, True)
                script.append(l[2:])
            elif l.startswith(b'  ... '):  # python inlines
                after.setdefault(prepos, []).append(l)
                script.append(l[2:])
            elif l.startswith(b'  $ '):  # commands
                if inpython:
                    script.append(b'EOF\n')
                    inpython = False
                after.setdefault(pos, []).append(l)
                prepos = pos

                cmd = rawcmd.split()
                toggletrace(rawcmd)
                if len(cmd) == 2 and cmd[0] == b'cd':
                    l = b'  $ cd %s || exit 1\n' % cmd[1]
                script.append(rawcmd)
            elif l.startswith(b'  > '):  # continuations
                after.setdefault(prepos, []).append(l)
                script.append(l[4:])
            elif l.startswith(b'  '):  # results
                # Queue up a list of expected results.
                expected.setdefault(pos, []).append(l[2:])
            else:
                if inpython:
                    script.append(b'EOF\n')

            toggletrace()
        return salt, script, after, expected

    def _processoutput(self, exitcode, output, salt, after, expected):
        # Merge the script output back into a unified test.
        warnonly = WARN_UNDEFINED  # 1: not yet; 2: yes; 3: for sure not
        if exitcode != 0:
            warnonly = WARN_NO

        pos = -1
        postout = []
        for out_rawline in output:
            out_line, cmd_line = out_rawline, None
            if salt in out_rawline:
                out_line, cmd_line = out_rawline.split(salt, 1)

            pos, postout, warnonly = self._process_out_line(
                out_line, pos, postout, expected, warnonly
            )
            pos, postout = self._process_cmd_line(cmd_line, pos, postout, after)

        if pos in after:
            postout += after.pop(pos)

        if warnonly == WARN_YES:
            exitcode = False  # Set exitcode to warned.

        return exitcode, postout

    def _process_out_line(self, out_line, pos, postout, expected, warnonly):
        while out_line:

                if el:
                    r, exact = self.linematch(el, out_line)
                if isinstance(r, str):
                    if r == '-glob':
                        out_line = ''.join(el.rsplit(' (glob)', 1))
                        r = ''  # Warn only this line.
                    elif r == "retry":
                        postout.append(b'  ' + el)
                    else:
                        log('\ninfo, unknown linematch result: %r\n' % r)
                        r = False

                    if isoptional(el):
                        optional.append(i)
                    else:
                        m = optline.match(el)
                        if m:
                            conditions = [c for c in m.group(2).split(b' ')]

                            if not self._iftest(conditions):
                                optional.append(i)
                    if exact:
                        # Don't allow line to be matches against a later

                for i in reversed(optional):
                    del els[i]
                postout.append(b'  ' + el)
            else:
                if self.NEEDESCAPE(out_line):
                    out_line = TTest._stringescape(
                        b'%s (esc)\n' % out_line.rstrip(b'\n')
                    )
                postout.append(b'  ' + out_line)  # Let diff deal with it.
                if r != '':  # If line failed.
                    warnonly = WARN_NO
                elif warnonly == WARN_UNDEFINED:
                    warnonly = WARN_YES
            break
        else:

            return True
        el = el.replace(b'$LOCALIP', b'*')
        i, n = 0, len(el)
        res = b''
        while i < n:
            c = el[i : i + 1]
            i += 1
            if c == b'\\' and i < n and el[i : i + 1] in b'*?\\/':
                res += el[i - 1 : i + 1]
                i += 1
            elif c == b'*':
                res += b'.*'
            elif c == b'?':
                res += b'.'

            else:
                res += re.escape(c)
        return TTest.rematch(res, l)

    def linematch(self, el, l):
        if el == l:  # perfect match (fast)
            return True, True
        retry = False
        if isoptional(el):
            retry = "retry"
            el = el[: -len(MARK_OPTIONAL)] + b"\n"
        else:
            m = optline.match(el)
            if m:
                conditions = [c for c in m.group(2).split(b' ')]

        missing = []
        failed = []
        for line in lines:
            if line.startswith(TTest.SKIPPED_PREFIX):
                line = line.splitlines()[0]
                missing.append(
                    line[len(TTest.SKIPPED_PREFIX) :].decode('utf-8')
                )
            elif line.startswith(TTest.FAILED_PREFIX):
                line = line.splitlines()[0]
                failed.append(line[len(TTest.FAILED_PREFIX) :].decode('utf-8'))

        return missing, failed

    @staticmethod
    def _escapef(m):

    @staticmethod
    def _stringescape(s):
        return TTest.ESCAPESUB(TTest._escapef, s)


iolock = threading.RLock()
firstlock = threading.RLock()
firsterror = False


class TestResult(unittest._TextTestResult):
    """Holds results when executing via unittest."""

    # Don't worry too much about accessing the non-public _TextTestResult.
    # It is relatively common in Python testing tools.
    def __init__(self, options, *args, **kwargs):
        super(TestResult, self).__init__(*args, **kwargs)

        if options.color == 'auto':
            self.color = pygmentspresent and self.stream.isatty()
        elif options.color == 'never':
            self.color = False
        else:  # 'always', for testing purposes
            self.color = pygmentspresent

    def onStart(self, test):
        """ Can be overriden by custom TestResult
        """

        with iolock:
            if self._options.nodiff:
                pass
            elif self._options.view:
                v = self._options.view
                subprocess.call(
                    r'"%s" "%s" "%s"'
                    % (v, _strpath(test.refpath), _strpath(test.errpath)),
                    shell=True,
                )
            else:
                servefail, lines = getdiff(
                    expected, got, test.refpath, test.errpath
                )
                self.stream.write('\n')
                for line in lines:
                    line = highlightdiff(line, self.color)
                    if PYTHON3:
                        self.stream.flush()

                        self.stream.write(line)
                        self.stream.flush()

                if servefail:
                    raise test.failureException(
                        'server failed to start (HGPORT=%s)' % test._startport
                    )

            # handle interactive prompt without releasing iolock
            if self._options.interactive:
                if test.readrefout() != expected:
                    self.stream.write(
                        'Reference output has changed (run again to prompt '
                        'changes)'
                    )
                else:
                    self.stream.write('Accept this change? [n] ')
                    self.stream.flush()
                    answer = sys.stdin.readline().strip()
                    if answer.lower() in ('y', 'yes'):

        # os.times module computes the user time and system time spent by
        # child's processes along with real elapsed time taken by a process.
        # This module has one limitation. It can only work for Linux user
        # and not for Windows.
        test.started = os.times()
        if self._firststarttime is None:  # thread racy but irrelevant
            self._firststarttime = test.started[4]

    def stopTest(self, test, interrupted=False):
        super(TestResult, self).stopTest(test)

        test.stopped = os.times()

        starttime = test.started
        endtime = test.stopped
        origin = self._firststarttime
        self.times.append(
            (
                test.name,
                endtime[2] - starttime[2],  # user space CPU time
                endtime[3] - starttime[3],  # sys  space CPU time
                endtime[4] - starttime[4],  # real time
                starttime[4] - origin,  # start date in run context
                endtime[4] - origin,  # end date in run context
            )
        )

        if interrupted:
            with iolock:
                self.stream.writeln(
                    'INTERRUPTED: %s (after %d seconds)'
                    % (test.name, self.times[-1][3])
                )


def getTestResult():
    """
    Returns the relevant test result
    """

        testresultmodule = __import__(os.environ["CUSTOM_TEST_RESULT"])
        return testresultmodule.TestResult
    else:
        return TestResult


class TestSuite(unittest.TestSuite):
    """Custom unittest TestSuite that knows how to execute Mercurial tests."""

    def __init__(
        self,
        testdir,
        jobs=1,
        whitelist=None,
        blacklist=None,
        retest=False,
        keywords=None,
        loop=False,
        runs_per_test=1,
        loadtest=None,
        showchannels=False,
        *args,
        **kwargs
    ):
        """Create a new instance that can run tests with a configuration.

        testdir specifies the directory where tests are executed from. This
        is typically the ``tests`` directory from Mercurial's source
        repository.

        # here instead of inside Test because it makes the running logic for
        # Test simpler.
        tests = []
        num_tests = [0]
        for test in self._tests:

            def get():
                num_tests[0] += 1
                if getattr(test, 'should_reload', False):
                    return self._loadtest(test, num_tests[0])
                return test

            if not os.path.exists(test.path):
                result.addSkip(test, "Doesn't exist")
                continue

            if not (self._whitelist and test.bname in self._whitelist):

            try:
                test(result)
                done.put(None)
            except KeyboardInterrupt:
                pass
            except:  # re-raises
                done.put(('!', test, 'run-test raised an error, see traceback'))
                raise
            finally:
                try:
                    channels[channel] = ''

                with iolock:
                    sys.stdout.write(d + '  ')
                    sys.stdout.flush()
                for x in xrange(10):
                    if channels:
                        time.sleep(0.1)
                count += 1

        stoppedearly = False

        if self._showchannels:

                if tests and not running == self._jobs:
                    test = tests.pop(0)
                    if self._loop:
                        if getattr(test, 'should_reload', False):
                            num_tests[0] += 1
                            tests.append(self._loadtest(test, num_tests[0]))
                        else:
                            tests.append(test)
                    if self._jobs == 1:
                        job(test, result)
                    else:
                        t = threading.Thread(
                            target=job, name=test.name, args=(test, result)
                        )
                        t.start()
                    running += 1

            # If we stop early we still need to wait on started tests to
            # finish. Otherwise, there is a race between the test completing

        channels = []

        return result


# Save the most recent 5 wall-clock runtimes of each test to a
# human-readable text file named .testtimes. Tests are sorted
# alphabetically, while times for each test are listed from oldest to
# newest.


def loadtimes(outputdir):
    times = []
    try:
        with open(os.path.join(outputdir, b'.testtimes')) as fp:
            for line in fp:
                m = re.match('(.*?) ([0-9. ]+)', line)
                times.append(
                    (m.group(1), [float(t) for t in m.group(2).split()])
                )
    except IOError as err:
        if err.errno != errno.ENOENT:
            raise
    return times


def savetimes(outputdir, result):
    saved = dict(loadtimes(outputdir))
    maxruns = 5
    skipped = set([str(t[0]) for t in result.skipped])

        if test not in skipped:
            ts = saved.setdefault(test, [])
            ts.append(real)
            ts[:] = ts[-maxruns:]

    fd, tmpname = tempfile.mkstemp(
        prefix=b'.testtimes', dir=outputdir, text=True
    )
    with os.fdopen(fd, 'w') as fp:
        for name, ts in sorted(saved.items()):
            fp.write('%s %s\n' % (name, ' '.join(['%.3f' % (t,) for t in ts])))
    timepath = os.path.join(outputdir, b'.testtimes')
    try:

    try:
        os.rename(tmpname, timepath)
    except OSError:
        pass


class TextTestRunner(unittest.TextTestRunner):
    """Custom unittest test runner that uses appropriate settings."""

    def __init__(self, runner, *args, **kwargs):
        super(TextTestRunner, self).__init__(*args, **kwargs)

        self._runner = runner

        self._result = getTestResult()(
            self._runner.options, self.stream, self.descriptions, self.verbosity
        )

    def listtests(self, test):
        test = sorted(test, key=lambda t: t.name)

        self._result.onStart(test)

        with iolock:
            self.stream.writeln('')

            if not self._runner.options.noskips:
                for test, msg in sorted(
                    self._result.skipped, key=lambda s: s[0].name
                ):
                    formatted = 'Skipped %s: %s\n' % (test.name, msg)
                    msg = highlightmsg(formatted, self._result.color)
                    self.stream.write(msg)
            for test, msg in sorted(
                self._result.failures, key=lambda f: f[0].name
            ):
                formatted = 'Failed %s: %s\n' % (test.name, msg)
                self.stream.write(highlightmsg(formatted, self._result.color))
            for test, msg in sorted(
                self._result.errors, key=lambda e: e[0].name
            ):
                self.stream.writeln('Errored %s: %s' % (test.name, msg))

            if self._runner.options.xunit:
                with open(self._runner.options.xunit, "wb") as xuf:
                    self._writexunit(self._result, xuf)

            if failed and self._runner.options.known_good_rev:
                self._bisecttests(t for t, m in self._result.failures)
            self.stream.writeln(
                '# Ran %d tests, %d skipped, %d failed.'
                % (self._result.testsRun, skipped + ignored, failed)
            )
            if failed:
                self.stream.writeln(
                    'python hash seed: %s' % os.environ['PYTHONHASHSEED']
                )
            if self._runner.options.time:
                self.printtimes(self._result.times)

            if self._runner.options.exceptions:
                exceptions = aggregateexceptions(
                    os.path.join(self._runner._outputdir, b'exceptions')
                )

                self.stream.writeln('Exceptions Report:')
                self.stream.writeln(
                    '%d total from %d frames'
                    % (exceptions['total'], len(exceptions['exceptioncounts']))
                )
                combined = exceptions['combined']
                for key in sorted(combined, key=combined.get, reverse=True):
                    frame, line, exc = key
                    totalcount, testcount, leastcount, leasttest = combined[key]

                    self.stream.writeln(
                        '%d (%d tests)\t%s: %s (%s - %d total)'
                        % (
                            totalcount,
                            testcount,
                            frame,
                            exc,
                            leasttest,
                            leastcount,
                        )
                    )

            self.stream.flush()

        return self._result

    def _bisecttests(self, tests):
        bisectcmd = ['hg', 'bisect']
        bisectrepo = self._runner.options.bisect_repo
        if bisectrepo:
            bisectcmd.extend(['-R', os.path.abspath(bisectrepo)])

        def pread(args):
            env = os.environ.copy()
            env['HGPLAIN'] = '1'
            p = subprocess.Popen(
                args, stderr=subprocess.STDOUT, stdout=subprocess.PIPE, env=env
            )
            data = p.stdout.read()
            p.wait()
            return data

        for test in tests:
            pread(bisectcmd + ['--reset']),
            pread(bisectcmd + ['--bad', '.'])
            pread(bisectcmd + ['--good', self._runner.options.known_good_rev])
            # TODO: we probably need to forward more options
            # that alter hg's behavior inside the tests.
            opts = ''
            withhg = self._runner.options.with_hg
            if withhg:
                opts += ' --with-hg=%s ' % shellquote(_strpath(withhg))
            rtc = '%s %s %s %s' % (sysexecutable, sys.argv[0], opts, test)
            data = pread(bisectcmd + ['--command', rtc])
            m = re.search(
                (
                    br'\nThe first (?P<goodbad>bad|good) revision '
                    br'is:\nchangeset: +\d+:(?P<node>[a-f0-9]+)\n.*\n'
                    br'summary: +(?P<summary>[^\n]+)\n'
                ),
                data,
                (re.MULTILINE | re.DOTALL),
            )
            if m is None:
                self.stream.writeln(
                    'Failed to identify failure point for %s' % test
                )
                continue
            dat = m.groupdict()
            verb = 'broken' if dat['goodbad'] == b'bad' else 'fixed'
            self.stream.writeln(
                '%s %s by %s (%s)'
                % (
                    test,
                    verb,
                    dat['node'].decode('ascii'),
                    dat['summary'].decode('utf8', 'ignore'),
                )
            )

    def printtimes(self, times):
        # iolock held by run
        self.stream.writeln('# Producing time report')
        times.sort(key=lambda t: (t[3]))
        cols = '%7.3f %7.3f %7.3f %7.3f %7.3f   %s'
        self.stream.writeln(
            '%-7s %-7s %-7s %-7s %-7s   %s'
            % ('start', 'end', 'cuser', 'csys', 'real', 'Test')
        )
        for tdata in times:
            test = tdata[0]
            cuser, csys, real, start, end = tdata[1:6]
            self.stream.writeln(cols % (start, end, cuser, csys, real, test))

    def _writexunit(result, outf):
        # See http://llg.cubic.org/docs/junit/ for a reference.
        timesd = dict((t[0], t[3]) for t in result.times)
        doc = minidom.Document()
        s = doc.createElement('testsuite')
        s.setAttribute('errors', "0")  # TODO
        s.setAttribute('failures', str(len(result.failures)))
        s.setAttribute('name', 'run-tests')
        s.setAttribute(
            'skipped', str(len(result.skipped) + len(result.ignored))
        )
        s.setAttribute('tests', str(result.testsRun))
        doc.appendChild(s)
        for tc in result.successes:
            t = doc.createElement('testcase')
            t.setAttribute('name', tc.name)

        for tdata in result.times:
            test = tdata[0]
            timesd[test] = tdata[1:]

        outcome = {}
        groups = [
            ('success', ((tc, None) for tc in result.successes)),
            ('failure', result.failures),
            ('skip', result.skipped),
        ]
        for res, testcases in groups:
            for tc, __ in testcases:
                if tc.name in timesd:
                    diff = result.faildata.get(tc.name, b'')
                    try:
                        diff = diff.decode('unicode_escape')
                    except UnicodeDecodeError as e:
                        diff = '%r decoding diff, sorry' % e
                    tres = {
                        'result': res,
                        'time': ('%0.3f' % timesd[tc.name][2]),
                        'cuser': ('%0.3f' % timesd[tc.name][0]),
                        'csys': ('%0.3f' % timesd[tc.name][1]),
                        'start': ('%0.3f' % timesd[tc.name][3]),
                        'end': ('%0.3f' % timesd[tc.name][4]),
                        'diff': diff,
                    }
                else:
                    # blacklisted test
                    tres = {'result': res}

                outcome[tc.name] = tres
        jsonout = json.dumps(
            outcome, sort_keys=True, indent=4, separators=(',', ': ')
        )
        outf.writelines(("testreport =", jsonout))


def sorttests(testdescs, previoustimes, shuffle=False):
    """Do an in-place sort of tests."""
    if shuffle:
        random.shuffle(testdescs)
        return

    if previoustimes:

        def sortkey(f):
            f = f['path']
            if f in previoustimes:
                # Use most recent time as estimate
                return -(previoustimes[f][-1])
            else:
                # Default to a rather arbitrary value of 1 second for new tests
                return -1.0

    else:
        # keywords for slow tests
        slow = {
            b'svn': 10,
            b'cvs': 10,
            b'hghave': 10,
            b'largefiles-update': 10,
            b'run-tests': 10,
            b'corruption': 10,
            b'race': 10,
            b'i18n': 10,
            b'check': 100,
            b'gendoc': 100,
            b'contrib-perf': 200,
            b'merge-combination': 100,
        }
        perf = {}

        def sortkey(f):
            # run largest tests first, as they tend to take the longest
            f = f['path']

                perf[f] = val / 1000.0
                return perf[f]

    testdescs.sort(key=sortkey)


class TestRunner(object):
    """Holds context for executing tests.

    Tests rely on a lot of state. This object holds it for them.
    """

            self._checktools()
            testdescs = self.findtests(tests)
            if options.profile_runner:
                import statprof

                statprof.start()
            result = self._run(testdescs)
            if options.profile_runner:
                statprof.stop()
                statprof.display()

                # without this, we get the default temp dir location, but
                # in all lowercase, which causes troubles with paths (issue3490)
                d = osenvironb.get(b'TMP', None)
            tmpdir = tempfile.mkdtemp(b'', b'hgtests.', d)

        self._hgtmp = osenvironb[b'HGTMP'] = os.path.realpath(tmpdir)

        if self.options.with_hg:
            self._installdir = None
            whg = self.options.with_hg
            self._bindir = os.path.dirname(os.path.realpath(whg))

                os.unlink(os.path.join(exceptionsdir, f))

            osenvironb[b'HGEXCEPTIONSDIR'] = exceptionsdir
            logexceptions = os.path.join(self._testdir, b'logexceptions.py')
            self.options.extra_config_opt.append(
                'extensions.logexceptions=%s' % logexceptions.decode('utf-8')
            )

        vlog("# Using TESTDIR", self._testdir)
        vlog("# Using RUNTESTDIR", osenvironb[b'RUNTESTDIR'])
        vlog("# Using HGTMP", self._hgtmp)
        vlog("# Using PATH", os.environ["PATH"])

        vlog("# Writing to directory", self._outputdir)

        try:
            return self._runtests(testdescs) or 0
        finally:
            time.sleep(0.1)
            self._cleanup()

    def findtests(self, args):
        """Finds possible test files from arguments.

        If you wish to inject custom tests into the test harness, this would
        be a good function to monkeypatch or override in a derived class.
        """
        if not args:
            if self.options.changed:
                proc = Popen4(
                    b'hg st --rev "%s" -man0 .'
                    % _bytespath(self.options.changed),
                    None,
                    0,
                )
                stdout, stderr = proc.communicate()
                args = stdout.strip(b'\0').split(b'\0')
            else:
                args = os.listdir(b'.')

            else:
                expanded_args.append(arg)
        args = expanded_args

        testcasepattern = re.compile(
            br'([\w-]+\.t|py)(?:#([a-zA-Z0-9_\-\.#]+))'
        )
        tests = []
        for t in args:
            case = []

            if not (
                os.path.basename(t).startswith(b'test-')
                and (t.endswith(b'.py') or t.endswith(b'.t'))
            ):

                m = testcasepattern.match(os.path.basename(t))
                if m is not None:
                    t_basename, casestr = m.groups()
                    t = os.path.join(os.path.dirname(t), t_basename)

            if t.endswith(b'.t'):
                # .t file may contain multiple test cases
                casedimensions = parsettestcases(t)
                if casedimensions:
                    cases = []

                    def addcases(case, casedimensions):
                        if not casedimensions:
                            cases.append(case)
                        else:
                            for c in casedimensions[0]:
                                addcases(case + [c], casedimensions[1:])

                    addcases([], casedimensions)
                    if case and case in cases:
                        cases = [case]
                    elif case:
                        # Ignore invalid cases

            failed = False
            kws = self.options.keywords
            if kws is not None and PYTHON3:
                kws = kws.encode('utf-8')

            suite = TestSuite(
                self._testdir,
                jobs=jobs,
                whitelist=self.options.whitelisted,
                blacklist=self.options.blacklist,
                retest=self.options.retest,
                keywords=kws,
                loop=self.options.loop,
                runs_per_test=self.options.runs_per_test,
                showchannels=self.options.showchannels,
                tests=tests,
                loadtest=_reloadtest,
            )
            verbosity = 1
            if self.options.list_tests:
                verbosity = 0
            elif self.options.verbose:
                verbosity = 2

                    self._usecorrectpython()
                if self.options.chg:
                    assert self._installdir
                    self._installchg()

                log(
                    'running %d tests using %d parallel processes'
                    % (num_tests, jobs)
                )

                result = runner.run(suite)

            if result.failures or result.errors:
                failed = True

        if failed:
            return 1

    def _getport(self, count):
        port = self._ports.get(count)  # do we have a cached entry?
        if port is None:
            portneeded = 3
            # above 100 tries we just give up and let test reports failure
            for tries in xrange(100):
                allfree = True

        tmpdir = os.path.join(self._hgtmp, b'child%d' % count)

        # extra keyword parameters. 'case' is used by .t tests
        kwds = dict((k, testdesc[k]) for k in ['case'] if k in testdesc)

        t = testcls(
            refpath,
            self._outputdir,
            tmpdir,
            keeptmpdir=self.options.keep_tmpdir,
            debug=self.options.debug,
            first=self.options.first,
            timeout=self.options.timeout,
            startport=self._getport(count),
            extraconfigopts=self.options.extra_config_opt,
            py3warnings=self.options.py3_warnings,
            shell=self.options.shell,
            hgcommand=self._hgcommand,
            usechg=bool(self.options.with_chg or self.options.chg),
            useipv6=useipv6,
            **kwds
        )
        t.should_reload = True
        return t

    def _cleanup(self):
        """Clean up state from this test invocation."""

        pyexename = sys.platform == 'win32' and b'python.exe' or b'python'

        # os.symlink() is a thing with py3 on Windows, but it requires
        # Administrator rights.
        if getattr(os, 'symlink', None) and os.name != 'nt':
            vlog(
                "# Making python executable in test path a symlink to '%s'"
                % sysexecutable
            )
            mypython = os.path.join(self._tmpbindir, pyexename)
            try:
                if os.readlink(mypython) == sysexecutable:
                    return
                os.unlink(mypython)

                    # child processes may race, which is harmless
                    if err.errno != errno.EEXIST:
                        raise
        else:
            exedir, exename = os.path.split(sysexecutable)
            vlog(
                "# Modifying search path to find %s as %s in '%s'"
                % (exename, pyexename, exedir)
            )
            path = os.environ['PATH'].split(os.pathsep)
            while exedir in path:
                path.remove(exedir)
            os.environ['PATH'] = os.pathsep.join([exedir] + path)
            if not self._findprogram(pyexename):

            # The --home="" trick works only on OS where os.sep == '/'
            # because of a distutils convert_path() fast-path. Avoid it at
            # least on Windows for now, deal with .pydistutils.cfg bugs
            # when they happen.
            nohome = b''
        cmd = (
            b'"%(exe)s" setup.py %(pure)s clean --all'
            b' build %(compiler)s --build-base="%(base)s"'
            b' install --force --prefix="%(prefix)s"'
            b' --install-lib="%(libdir)s"'
            b' --install-scripts="%(bindir)s" %(nohome)s >%(logfile)s 2>&1'
            % {
                b'exe': exe,
                b'pure': pure,
                b'compiler': compiler,
                b'base': os.path.join(self._hgtmp, b"build"),
                b'prefix': self._installdir,
                b'libdir': self._pythondir,
                b'bindir': self._bindir,
                b'nohome': nohome,
                b'logfile': installerrs,
            }
        )

        # setuptools requires install directories to exist.
        def makedirs(p):
            try:
                os.makedirs(p)
            except OSError as e:
                if e.errno != errno.EEXIST:
                    raise

        makedirs(self._pythondir)
        makedirs(self._bindir)

        vlog("# Running", cmd)
        if subprocess.call(_strpath(cmd), shell=True) == 0:

            # hg.bat expects to be put in bin/scripts while run-tests.py
            # installation layout put it in bin/ directly. Fix it
            with open(hgbat, 'rb') as f:
                data = f.read()
            if br'"%~dp0..\python" "%~dp0hg" %*' in data:
                data = data.replace(
                    br'"%~dp0..\python" "%~dp0hg" %*',
                    b'"%~dp0python" "%~dp0hg" %*',
                )
                with open(hgbat, 'wb') as f:
                    f.write(data)
            else:
                print('WARNING: cannot fix hg.bat reference to python.exe')

            os.environ['COVERAGE_DIR'] = covdir

    def _checkhglib(self, verb):
        """Ensure that the 'mercurial' package imported by python is
        the one we expect it to be.  If not, print a warning to stderr."""
        if (self._bindir == self._pythondir) and (
            self._bindir != self._tmpbindir
        ):
            # The pythondir has been inferred from --with-hg flag.
            # We cannot expect anything sensible here.
            return
        expecthg = os.path.join(self._pythondir, b'mercurial')
        actualhg = self._gethgpath()
        if os.path.abspath(actualhg) != os.path.abspath(expecthg):
            sys.stderr.write(
                'warning: %s with unexpected mercurial lib: %s\n'
                '         (expected %s)\n' % (verb, actualhg, expecthg)
            )

    def _gethgpath(self):
        """Return the path to the mercurial package that is actually found by
        the current Python interpreter."""
        if self._hgpath is not None:
            return self._hgpath

    def _installchg(self):
        """Install chg into the test environment"""
        vlog('# Performing temporary installation of CHG')
        assert os.path.dirname(self._bindir) == self._installdir
        assert self._hgroot, 'must be called after _installhg()'
        cmd = b'"%(make)s" clean install PREFIX="%(prefix)s"' % {
            b'make': b'make',  # TODO: switch by option or environment?
            b'prefix': self._installdir,
        }
        cwd = os.path.join(self._hgroot, b'contrib', b'chg')
        vlog("# Running", cmd)
        proc = subprocess.Popen(
            cmd,
            shell=True,
            cwd=cwd,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
        )
        out, _err = proc.communicate()
        if proc.returncode != 0:
            if PYTHON3:
                sys.stdout.buffer.write(out)
            else:

            sys.exit(1)

    def _outputcoverage(self):
        """Produce code coverage output."""
        import coverage

        coverage = coverage.coverage

        vlog('# Producing coverage report')
        # chdir is the easiest way to get short, relative paths in the
        # output.

                p += b'.exe'
            found = self._findprogram(p)
            if found:
                vlog("# Found prerequisite", p, "at", found)
            else:
                print(
                    "WARNING: Did not find prerequisite tool: %s "
                    % p.decode("utf-8")
                )


def aggregateexceptions(path):
    exceptioncounts = collections.Counter()
    testsbyfailure = collections.defaultdict(set)
    failuresbytest = collections.defaultdict(set)

    # Create a combined counter so we can sort by total occurrences and
    # impacted tests.
    combined = {}
    for key in exceptioncounts:
        combined[key] = (
            exceptioncounts[key],
            len(testsbyfailure[key]),
            leastfailing[key][0],
            leastfailing[key][1],
        )

    return {
        'exceptioncounts': exceptioncounts,
        'total': sum(exceptioncounts.values()),
        'combined': combined,
        'leastfailing': leastfailing,
        'byfailure': testsbyfailure,
        'bytest': failuresbytest,
    }


if __name__ == '__main__':
    runner = TestRunner()

    try:
        import msvcrt

        msvcrt.setmode(sys.stdin.fileno(), os.O_BINARY)
        msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)
        msvcrt.setmode(sys.stderr.fileno(), os.O_BINARY)
    except ImportError:
        pass