comparison tests/run-tests.py @ 44448:55c443fcb4fc

tests: rename _bytespath() to _sys2bytes() and _strpath() to _sys2str()

The names were not general enough because e.g. _strpath() was used for
converting IP addresses.

44447:fc70291f3d24

if sys.version_info > (3, 5, 0):
    PYTHON3 = True
    xrange = range  # we use xrange in one place, and we'd rather not use range

    def _bytespath(p):
        if p is None:
            return p
        return p.encode('utf-8')

    def _strpath(p):
        if p is None:
            return p
        return p.decode('utf-8')

    osenvironb = getattr(os, 'environb', None)

                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]))

    # In python 2.x, path operations are generally done using
    # bytestrings by default, so we don't have to do any extra
    # fiddling there. We define the wrapper functions anyway just to
    # help keep code consistent between platforms.
    def _bytespath(p):
        return p

    _strpath = _bytespath
    osenvironb = os.environ
    getcwdb = os.getcwd

# For Windows support
wifexited = getattr(os, "WIFEXITED", lambda x: False)

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,

elif os.environ.get('PYTHON'):
    sysexecutable = os.environ['PYTHON']
else:
    raise AssertionError('Could not find Python interpreter')

PYTHON = _bytespath(sysexecutable.replace('\\', '/'))
IMPL_PATH = b'PYTHONPATH'
if 'java' in sys.platform:
    IMPL_PATH = b'JYTHONPATH'

defaults = {

        options.pure = True

    if options.local:
        if options.with_hg or options.with_chg:
            parser.error('--local cannot be used with --with-hg or --with-chg')
        testdir = os.path.dirname(_bytespath(canonpath(sys.argv[0])))
        reporootdir = os.path.dirname(testdir)
        pathandattrs = [(b'hg', 'with_hg')]
        if options.chg:
            pathandattrs.append((b'contrib/chg/chg', 'with_chg'))
        for relpath, attr in pathandattrs:

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

            startport = defaults['port']
        if slowtimeout is None:
            slowtimeout = defaults['slowtimeout']
        self.path = path
        self.bname = os.path.basename(path)
        self.name = _strpath(self.bname)
        self._testdir = os.path.dirname(path)
        self._outputdir = outputdir
        self._tmpname = os.path.basename(path)
        self.errpath = os.path.join(self._outputdir, b'%s.err' % self.bname)

        self._first = first
        self._timeout = timeout
        self._slowtimeout = slowtimeout
        self._startport = startport
        self._extraconfigopts = extraconfigopts or []
        self._shell = _bytespath(shell)
        self._hgcommand = hgcommand or b'hg'
        self._usechg = usechg
        self._useipv6 = useipv6

        self._aborted = False

    def _genrestoreenv(self, testenv):
        """Generate a script that can be used by tests to restore the original
        environment."""
        # Put the restoreenv script inside self._threadtmp
        scriptpath = os.path.join(self._threadtmp, b'restoreenv.sh')
        testenv['HGTEST_RESTOREENV'] = _strpath(scriptpath)

        # Only restore environment variable names that the shell allows
        # us to export.
        name_regex = re.compile('^[a-zA-Z][a-zA-Z0-9_]*$')

            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
        env['HOME'] = _strpath(self._testtmp)
        # 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)"'
        )

        if extraextensions:
            env['HGTESTEXTRAEXTENSIONS'] = b' '.join(extraextensions)

        # LOCALIP could be ::1 or 127.0.0.1. Useful for tests that require raw
        # IP addresses.
        env['LOCALIP'] = _strpath(self._localip())

        # This has the same effect as Py_LegacyWindowsStdioFlag in exewrapper.c,
        # but this is needed for testing python instances like dummyssh,
        # dummysmtpd.py, and dumbhttp.py.
        if PYTHON3 and os.name == 'nt':

        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)

        self._case = case
        self._allcases = {x for y in parsettestcases(path) for x in y}
        super(TTest, self).__init__(path, *args, **kwds)
        if case:
            casepath = b'#'.join(case)
            self.name = '%s#%s' % (self.name, _strpath(casepath))
            self.errpath = b'%s#%s.err' % (self.errpath[:-4], casepath)
            self._tmpname += b'-%s' % casepath
        self._have = {}

    @property

        if allreqs in self._have:
            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,

                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

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

        """Run the test suite."""
        oldmask = os.umask(0o22)
        try:
            parser = parser or getparser()
            options = parseargs(args, parser)
            tests = [_bytespath(a) for a in options.tests]
            if options.test_list is not None:
                for listfile in options.test_list:
                    with open(listfile, 'rb') as f:
                        tests.extend(t for t in f.read().splitlines() if t)
            self.options = options

            pathname = os.path.dirname(testdescs[0]['path'])
            if pathname:
                testdir = os.path.join(testdir, pathname)
        self._testdir = osenvironb[b'TESTDIR'] = testdir
        if self.options.outputdir:
            self._outputdir = canonpath(_bytespath(self.options.outputdir))
        else:
            self._outputdir = getcwdb()
            if testdescs and pathname:
                self._outputdir = os.path.join(self._outputdir, pathname)
        previoustimes = {}

            # we do the randomness ourself to know what seed is used
            os.environ['PYTHONHASHSEED'] = str(random.getrandbits(32))

        if self.options.tmpdir:
            self.options.keep_tmpdir = True
            tmpdir = _bytespath(self.options.tmpdir)
            if os.path.exists(tmpdir):
                # Meaning of tmpdir has changed since 1.3: we used to create
                # HGTMP inside tmpdir; now HGTMP is tmpdir.  So fail if
                # tmpdir already exists.
                print("error: temp dir %r already exists" % tmpdir)

            self._hgcommand = os.path.basename(whg)
            self._tmpbindir = os.path.join(self._hgtmp, b'install', b'bin')
            os.makedirs(self._tmpbindir)

            normbin = os.path.normpath(os.path.abspath(whg))
            normbin = normbin.replace(_bytespath(os.sep), b'/')

            # Other Python scripts in the test harness need to
            # `import mercurial`. If `hg` is a Python script, we assume
            # the Mercurial modules are relative to its path and tell the tests
            # to load Python modules from its directory.

            self._hgcommand = os.path.basename(self.options.with_chg)

        osenvironb[b"BINDIR"] = self._bindir
        osenvironb[b"PYTHON"] = PYTHON

        fileb = _bytespath(__file__)
        runtestdir = os.path.abspath(os.path.dirname(fileb))
        osenvironb[b'RUNTESTDIR'] = runtestdir
        if PYTHON3:
            sepb = _bytespath(os.pathsep)
        else:
            sepb = os.pathsep
        path = [self._bindir, runtestdir] + osenvironb[b"PATH"].split(sepb)
        if os.path.islink(__file__):
            # test helper will likely be at the end of the symlink

            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", _strpath(self._testdir))
        vlog("# Using RUNTESTDIR", _strpath(osenvironb[b'RUNTESTDIR']))
        vlog("# Using HGTMP", _strpath(self._hgtmp))
        vlog("# Using PATH", os.environ["PATH"])
        vlog(
            "# Using", _strpath(IMPL_PATH), _strpath(osenvironb[IMPL_PATH]),
        )
        vlog("# Writing to directory", _strpath(self._outputdir))

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

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

    def _cleanup(self):
        """Clean up state from this test invocation."""
        if self.options.keep_tmpdir:
            return

        vlog("# Cleaning up HGTMP", _strpath(self._hgtmp))
        shutil.rmtree(self._hgtmp, True)
        for f in self._createdfiles:
            try:
                os.remove(f)
            except OSError:

        # Run installer in hg root
        script = os.path.realpath(sys.argv[0])
        exe = sysexecutable
        if PYTHON3:
            compiler = _bytespath(compiler)
            script = _bytespath(script)
            exe = _bytespath(exe)
        hgroot = os.path.dirname(os.path.dirname(script))
        self._hgroot = hgroot
        os.chdir(hgroot)
        nohome = b'--home=""'
        if os.name == 'nt':

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

        vlog("# Running", cmd.decode("utf-8"))
        if subprocess.call(_strpath(cmd), shell=True) == 0:
            if not self.options.verbose:
                try:
                    os.remove(installerrs)
                except OSError as e:
                    if e.errno != errno.ENOENT:

            return self._hgpath

        cmd = b'"%s" -c "import mercurial; print (mercurial.__path__[0])"'
        cmd = cmd % PYTHON
        if PYTHON3:
            cmd = _strpath(cmd)

        p = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
        out, err = p.communicate()

        self._hgpath = out.strip()

        vlog('# Producing coverage report')
        # chdir is the easiest way to get short, relative paths in the
        # output.
        os.chdir(self._hgroot)
        covdir = os.path.join(_strpath(self._installdir), '..', 'coverage')
        cov = coverage(data_file=os.path.join(covdir, 'cov'))

        # Map install directory paths back to source directory.
        cov.config.paths['srcdir'] = ['.', _strpath(self._pythondir)]

        cov.combine()

        omit = [
            _strpath(os.path.join(x, b'*'))
            for x in [self._bindir, self._testdir]
        ]
        cov.report(ignore_errors=True, omit=omit)

        if self.options.htmlcov:
            htmldir = os.path.join(_strpath(self._outputdir), 'htmlcov')
            cov.html_report(directory=htmldir, omit=omit)
        if self.options.annotate:
            adir = os.path.join(_strpath(self._outputdir), 'annotated')
            if not os.path.isdir(adir):
                os.mkdir(adir)
            cov.annotate(directory=adir, omit=omit)

    def _findprogram(self, program):
        """Search PATH for a executable program"""
        dpb = _bytespath(os.defpath)
        sepb = _bytespath(os.pathsep)
        for p in osenvironb.get(b'PATH', dpb).split(sepb):
            name = os.path.join(p, program)
            if os.name == 'nt' or os.access(name, os.X_OK):
                return name
        return None

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


def aggregateexceptions(path):

44448:55c443fcb4fc

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 _sys2bytes(p):
        if p is None:
            return p
        return p.encode('utf-8')

    def _bytes2sys(p):
        if p is None:
            return p
        return p.decode('utf-8')

    osenvironb = getattr(os, 'environb', None)

                self.__len__ = strenv.__len__
                self.clear = strenv.clear
                self._strenv = strenv

            def __getitem__(self, k):
                v = self._strenv.__getitem__(_bytes2sys(k))
                return _sys2bytes(v)

            def __setitem__(self, k, v):
                self._strenv.__setitem__(_bytes2sys(k), _bytes2sys(v))

            def __delitem__(self, k):
                self._strenv.__delitem__(_bytes2sys(k))

            def __contains__(self, k):
                return self._strenv.__contains__(_bytes2sys(k))

            def __iter__(self):
                return iter([_sys2bytes(k) for k in iter(self._strenv)])

            def get(self, k, default=None):
                v = self._strenv.get(_bytes2sys(k), _bytes2sys(default))
                return _sys2bytes(v)

            def pop(self, k, default=None):
                v = self._strenv.pop(_bytes2sys(k), _bytes2sys(default))
                return _sys2bytes(v)

        osenvironb = environbytes(os.environ)

    getcwdb = getattr(os, 'getcwdb')
    if not getcwdb or os.name == 'nt':
        getcwdb = lambda: _sys2bytes(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]))

    # In python 2.x, path operations are generally done using
    # bytestrings by default, so we don't have to do any extra
    # fiddling there. We define the wrapper functions anyway just to
    # help keep code consistent between platforms.
    def _sys2bytes(p):
        return p

    _bytes2sys = _sys2bytes
    osenvironb = os.environ
    getcwdb = os.getcwd

# For Windows support
wifexited = getattr(os, "WIFEXITED", lambda x: False)

def Popen4(cmd, wd, timeout, env=None):
    processlock.acquire()
    p = subprocess.Popen(
        _bytes2sys(cmd),
        shell=True,
        bufsize=-1,
        cwd=_bytes2sys(wd),
        env=env,
        close_fds=closefds,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,

elif os.environ.get('PYTHON'):
    sysexecutable = os.environ['PYTHON']
else:
    raise AssertionError('Could not find Python interpreter')

PYTHON = _sys2bytes(sysexecutable.replace('\\', '/'))
IMPL_PATH = b'PYTHONPATH'
if 'java' in sys.platform:
    IMPL_PATH = b'JYTHONPATH'

defaults = {

        options.pure = True

    if options.local:
        if options.with_hg or options.with_chg:
            parser.error('--local cannot be used with --with-hg or --with-chg')
        testdir = os.path.dirname(_sys2bytes(canonpath(sys.argv[0])))
        reporootdir = os.path.dirname(testdir)
        pathandattrs = [(b'hg', 'with_hg')]
        if options.chg:
            pathandattrs.append((b'contrib/chg/chg', 'with_chg'))
        for relpath, attr in pathandattrs:

            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, _bytes2sys(binpath))

    if options.with_hg:
        options.with_hg = canonpath(_sys2bytes(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 (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(_sys2bytes(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')

            startport = defaults['port']
        if slowtimeout is None:
            slowtimeout = defaults['slowtimeout']
        self.path = path
        self.bname = os.path.basename(path)
        self.name = _bytes2sys(self.bname)
        self._testdir = os.path.dirname(path)
        self._outputdir = outputdir
        self._tmpname = os.path.basename(path)
        self.errpath = os.path.join(self._outputdir, b'%s.err' % self.bname)

        self._first = first
        self._timeout = timeout
        self._slowtimeout = slowtimeout
        self._startport = startport
        self._extraconfigopts = extraconfigopts or []
        self._shell = _sys2bytes(shell)
        self._hgcommand = hgcommand or b'hg'
        self._usechg = usechg
        self._useipv6 = useipv6

        self._aborted = False

    def _genrestoreenv(self, testenv):
        """Generate a script that can be used by tests to restore the original
        environment."""
        # Put the restoreenv script inside self._threadtmp
        scriptpath = os.path.join(self._threadtmp, b'restoreenv.sh')
        testenv['HGTEST_RESTOREENV'] = _bytes2sys(scriptpath)

        # Only restore environment variable names that the shell allows
        # us to export.
        name_regex = re.compile('^[a-zA-Z][a-zA-Z0-9_]*$')

            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'] = _bytes2sys(self._testtmp)
        env['TESTNAME'] = self.name
        env['HOME'] = _bytes2sys(self._testtmp)
        # 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"] = _bytes2sys(os.path.join(self._threadtmp, b'.hgrc'))
        env["DAEMON_PIDS"] = _bytes2sys(
            os.path.join(self._threadtmp, b'daemon.pids')
        )
        env["HGEDITOR"] = (
            '"' + sysexecutable + '"' + ' -c "import sys; sys.exit(0)"'
        )

        if extraextensions:
            env['HGTESTEXTRAEXTENSIONS'] = b' '.join(extraextensions)

        # LOCALIP could be ::1 or 127.0.0.1. Useful for tests that require raw
        # IP addresses.
        env['LOCALIP'] = _bytes2sys(self._localip())

        # This has the same effect as Py_LegacyWindowsStdioFlag in exewrapper.c,
        # but this is needed for testing python instances like dummyssh,
        # dummysmtpd.py, and dumbhttp.py.
        if PYTHON3 and os.name == 'nt':

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

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

        self._case = case
        self._allcases = {x for y in parsettestcases(path) for x in y}
        super(TTest, self).__init__(path, *args, **kwds)
        if case:
            casepath = b'#'.join(case)
            self.name = '%s#%s' % (self.name, _bytes2sys(casepath))
            self.errpath = b'%s#%s.err' % (self.errpath[:-4], casepath)
            self._tmpname += b'-%s' % casepath
        self._have = {}

    @property

        if allreqs in self._have:
            return self._have.get(allreqs)

        # TODO do something smarter when all other uses of hghave are gone.
        runtestdir = os.path.abspath(os.path.dirname(_sys2bytes(__file__)))
        tdir = runtestdir.replace(b'\\', b'/')
        proc = Popen4(
            b'%s -c "%s/hghave %s"' % (self._shell, tdir, allreqs),
            self._testtmp,
            0,

                pass
            elif self._options.view:
                v = self._options.view
                subprocess.call(
                    r'"%s" "%s" "%s"'
                    % (v, _bytes2sys(test.refpath), _bytes2sys(test.errpath)),
                    shell=True,
                )
            else:
                servefail, lines = getdiff(
                    expected, got, test.refpath, test.errpath

            # 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(_bytes2sys(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 '

        """Run the test suite."""
        oldmask = os.umask(0o22)
        try:
            parser = parser or getparser()
            options = parseargs(args, parser)
            tests = [_sys2bytes(a) for a in options.tests]
            if options.test_list is not None:
                for listfile in options.test_list:
                    with open(listfile, 'rb') as f:
                        tests.extend(t for t in f.read().splitlines() if t)
            self.options = options

            pathname = os.path.dirname(testdescs[0]['path'])
            if pathname:
                testdir = os.path.join(testdir, pathname)
        self._testdir = osenvironb[b'TESTDIR'] = testdir
        if self.options.outputdir:
            self._outputdir = canonpath(_sys2bytes(self.options.outputdir))
        else:
            self._outputdir = getcwdb()
            if testdescs and pathname:
                self._outputdir = os.path.join(self._outputdir, pathname)
        previoustimes = {}

            # we do the randomness ourself to know what seed is used
            os.environ['PYTHONHASHSEED'] = str(random.getrandbits(32))

        if self.options.tmpdir:
            self.options.keep_tmpdir = True
            tmpdir = _sys2bytes(self.options.tmpdir)
            if os.path.exists(tmpdir):
                # Meaning of tmpdir has changed since 1.3: we used to create
                # HGTMP inside tmpdir; now HGTMP is tmpdir.  So fail if
                # tmpdir already exists.
                print("error: temp dir %r already exists" % tmpdir)

            self._hgcommand = os.path.basename(whg)
            self._tmpbindir = os.path.join(self._hgtmp, b'install', b'bin')
            os.makedirs(self._tmpbindir)

            normbin = os.path.normpath(os.path.abspath(whg))
            normbin = normbin.replace(_sys2bytes(os.sep), b'/')

            # Other Python scripts in the test harness need to
            # `import mercurial`. If `hg` is a Python script, we assume
            # the Mercurial modules are relative to its path and tell the tests
            # to load Python modules from its directory.

            self._hgcommand = os.path.basename(self.options.with_chg)

        osenvironb[b"BINDIR"] = self._bindir
        osenvironb[b"PYTHON"] = PYTHON

        fileb = _sys2bytes(__file__)
        runtestdir = os.path.abspath(os.path.dirname(fileb))
        osenvironb[b'RUNTESTDIR'] = runtestdir
        if PYTHON3:
            sepb = _sys2bytes(os.pathsep)
        else:
            sepb = os.pathsep
        path = [self._bindir, runtestdir] + osenvironb[b"PATH"].split(sepb)
        if os.path.islink(__file__):
            # test helper will likely be at the end of the symlink

            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", _bytes2sys(self._testdir))
        vlog("# Using RUNTESTDIR", _bytes2sys(osenvironb[b'RUNTESTDIR']))
        vlog("# Using HGTMP", _bytes2sys(self._hgtmp))
        vlog("# Using PATH", os.environ["PATH"])
        vlog(
            "# Using", _bytes2sys(IMPL_PATH), _bytes2sys(osenvironb[IMPL_PATH]),
        )
        vlog("# Writing to directory", _bytes2sys(self._outputdir))

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

        """
        if not args:
            if self.options.changed:
                proc = Popen4(
                    b'hg st --rev "%s" -man0 .'
                    % _sys2bytes(self.options.changed),
                    None,
                    0,
                )
                stdout, stderr = proc.communicate()
                args = stdout.strip(b'\0').split(b'\0')

    def _cleanup(self):
        """Clean up state from this test invocation."""
        if self.options.keep_tmpdir:
            return

        vlog("# Cleaning up HGTMP", _bytes2sys(self._hgtmp))
        shutil.rmtree(self._hgtmp, True)
        for f in self._createdfiles:
            try:
                os.remove(f)
            except OSError:

        # Run installer in hg root
        script = os.path.realpath(sys.argv[0])
        exe = sysexecutable
        if PYTHON3:
            compiler = _sys2bytes(compiler)
            script = _sys2bytes(script)
            exe = _sys2bytes(exe)
        hgroot = os.path.dirname(os.path.dirname(script))
        self._hgroot = hgroot
        os.chdir(hgroot)
        nohome = b'--home=""'
        if os.name == 'nt':

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

        vlog("# Running", cmd.decode("utf-8"))
        if subprocess.call(_bytes2sys(cmd), shell=True) == 0:
            if not self.options.verbose:
                try:
                    os.remove(installerrs)
                except OSError as e:
                    if e.errno != errno.ENOENT:

            return self._hgpath

        cmd = b'"%s" -c "import mercurial; print (mercurial.__path__[0])"'
        cmd = cmd % PYTHON
        if PYTHON3:
            cmd = _bytes2sys(cmd)

        p = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True)
        out, err = p.communicate()

        self._hgpath = out.strip()

        vlog('# Producing coverage report')
        # chdir is the easiest way to get short, relative paths in the
        # output.
        os.chdir(self._hgroot)
        covdir = os.path.join(_bytes2sys(self._installdir), '..', 'coverage')
        cov = coverage(data_file=os.path.join(covdir, 'cov'))

        # Map install directory paths back to source directory.
        cov.config.paths['srcdir'] = ['.', _bytes2sys(self._pythondir)]

        cov.combine()

        omit = [
            _bytes2sys(os.path.join(x, b'*'))
            for x in [self._bindir, self._testdir]
        ]
        cov.report(ignore_errors=True, omit=omit)

        if self.options.htmlcov:
            htmldir = os.path.join(_bytes2sys(self._outputdir), 'htmlcov')
            cov.html_report(directory=htmldir, omit=omit)
        if self.options.annotate:
            adir = os.path.join(_bytes2sys(self._outputdir), 'annotated')
            if not os.path.isdir(adir):
                os.mkdir(adir)
            cov.annotate(directory=adir, omit=omit)

    def _findprogram(self, program):
        """Search PATH for a executable program"""
        dpb = _sys2bytes(os.defpath)
        sepb = _sys2bytes(os.pathsep)
        for p in osenvironb.get(b'PATH', dpb).split(sepb):
            name = os.path.join(p, program)
            if os.name == 'nt' or os.access(name, os.X_OK):
                return name
        return None

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


def aggregateexceptions(path):