#!/usr/bin/env python3

#

# run-tests.py - Run a set of tests on Mercurial

#

# Copyright 2006 Olivia Mackall <olivia@selenic.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

# Modifying this script is tricky because it has many modes:

#   - serial (default) vs parallel (-jN, N > 1)

#   - no coverage (default) vs coverage (-c, -C, -s)

#   - temp install (default) vs specific hg script (--with-hg, --local)

#   - tests are a mix of shell scripts and Python scripts

#

# If you change this script, it is recommended that you ensure you

# haven't broken it by running it in various modes with a representative

# sample of test scripts.  For example:

#

#  1) serial, no coverage, temp install:

#      ./run-tests.py test-s*

#  2) serial, no coverage, local hg:

#      ./run-tests.py --local test-s*

#  3) serial, coverage, temp install:

#      ./run-tests.py -c test-s*

#  4) serial, coverage, local hg:

#      ./run-tests.py -c --local test-s*      # unsupported

#  5) parallel, no coverage, temp install:

#      ./run-tests.py -j2 test-s*

#  6) parallel, no coverage, local hg:

#      ./run-tests.py -j2 --local test-s*

#  7) parallel, coverage, temp install:

#      ./run-tests.py -j2 -c test-s*          # currently broken

#  8) parallel, coverage, local install:

#      ./run-tests.py -j2 -c --local test-s*  # unsupported (and broken)

#  9) parallel, custom tmp dir:

#      ./run-tests.py -j2 --tmpdir /tmp/myhgtests

#  10) parallel, pure, tests that call run-tests:

#      ./run-tests.py --pure `grep -l run-tests.py *.t`

#

# (You could use any subset of the tests: test-s* happens to match

# enough that it's worth doing parallel runs, few enough that it

# completes fairly quickly, includes both shell and Python scripts, and

# includes some scripts that run daemon processes.)

from __future__ import absolute_import, print_function

import argparse

import collections

import contextlib

import difflib

import distutils.version as version

import errno

import json

import multiprocessing

import os

import platform

import random

import re

import shutil

import signal

import socket

import subprocess

import sys

import sysconfig

import tempfile

import threading

import time

import unittest

import uuid

import xml.dom.minidom as minidom

try:

    import Queue as queue

except ImportError:

    import queue

try:

    import shlex

    shellquote = shlex.quote

except (ImportError, AttributeError):

    import pipes

    shellquote = pipes.quote

processlock = threading.Lock()

pygmentspresent = False

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

    if os.name == 'nt':

        hgpath = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

        sys.path.append(hgpath)

        try:

            from mercurial import win32  # pytype: disable=import-error

            # Don't check the result code because it fails on heptapod, but

            # something is able to convert to color anyway.

            win32.enablevtmode()

        finally:

            sys.path = sys.path[:-1]

    pygmentspresent = True

    difflexer = lexers.DiffLexer()

    terminal256formatter = formatters.Terminal256Formatter()

except ImportError:

    pass

if pygmentspresent:

    class TestRunnerStyle(style.Style):

        default_style = ""

        skipped = token.string_to_tokentype("Token.Generic.Skipped")

        failed = token.string_to_tokentype("Token.Generic.Failed")

        skippedname = token.string_to_tokentype("Token.Generic.SName")

        failedname = token.string_to_tokentype("Token.Generic.FName")

        styles = {

            skipped: '#e5e5e5',

            skippedname: '#00ffff',

            failed: '#7f0000',

            failedname: '#ff0000',

        }

    class TestRunnerLexer(lexer.RegexLexer):

        testpattern = r'[\w-]+\.(t|py)(#[a-zA-Z0-9_\-\.]+)?'

        tokens = {

            'root': [

                (r'^Skipped', token.Generic.Skipped, 'skipped'),

                (r'^Failed ', token.Generic.Failed, 'failed'),

                (r'^ERROR: ', token.Generic.Failed, 'failed'),

            ],

            'skipped': [

                (testpattern, token.Generic.SName),

                (r':.*', token.Generic.Skipped),

            ],

            'failed': [

                (testpattern, token.Generic.FName),

                (r'(:| ).*', token.Generic.Failed),

            ],

        }

    runnerformatter = formatters.Terminal256Formatter(style=TestRunnerStyle)

    runnerlexer = TestRunnerLexer()

origenviron = os.environ.copy()

if sys.version_info > (3, 5, 0):

    PYTHON3 = True

    xrange = range  # we use xrange in one place, and we'd rather not use range

    def _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)

    if osenvironb is None:

        # Windows lacks os.environb, for instance.  A proxy over the real thing

        # instead of a copy allows the environment to be updated via bytes on

        # all platforms.

        class environbytes(object):

            def __init__(self, strenv):

                self.__len__ = strenv.__len__

                self.clear = strenv.clear

                self._strenv = strenv

            def __getitem__(self, k):

                v = self._strenv.__getitem__(_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]))

    )

    sys.exit(70)  # EX_SOFTWARE from `man 3 sysexit`

else:

    PYTHON3 = False

    # In python 2.x, path operations are generally done using

    # bytestrings by default, so we don't have to do any extra

    # fiddling there. We define the wrapper functions anyway just to

    # help keep code consistent between platforms.

    def _sys2bytes(p):

        return p

    _bytes2sys = _sys2bytes

    osenvironb = os.environ

    getcwdb = os.getcwd

# For Windows support

wifexited = getattr(os, "WIFEXITED", lambda x: False)

# Whether to use IPv6

def checksocketfamily(name, port=20058):

    """return true if we can listen on localhost using family=name

    name should be either 'AF_INET', or 'AF_INET6'.

    port being used is okay - EADDRINUSE is considered as successful.

    """

    family = getattr(socket, name, None)

    if family is None:

        return False

    try:

        s = socket.socket(family, socket.SOCK_STREAM)

        s.bind(('localhost', port))

        s.close()

        return True

    except socket.error as exc:

        if exc.errno == errno.EADDRINUSE:

            return True

        elif exc.errno in (errno.EADDRNOTAVAIL, errno.EPROTONOSUPPORT):

            return False

        else:

            raise

    else:

        return False

# useipv6 will be set by parseargs

useipv6 = None

def checkportisavailable(port):

    """return true if a port seems free to bind on localhost"""

    if useipv6:

        family = socket.AF_INET6

    else:

        family = socket.AF_INET

    try:

        with contextlib.closing(socket.socket(family, socket.SOCK_STREAM)) as s:

            s.bind(('localhost', port))

        return True

    except socket.error as exc:

        if os.name == 'nt' and exc.errno == errno.WSAEACCES:

            return False

        elif PYTHON3:

            # TODO: make a proper exception handler after dropping py2.  This

            #       works because socket.error is an alias for OSError on py3,

            #       which is also the baseclass of PermissionError.

            if isinstance(exc, PermissionError):

                return False

        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(

        _bytes2sys(cmd),

        shell=True,

        bufsize=-1,

        cwd=_bytes2sys(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

            vlog('# Timout reached for process %d' % p.pid)

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

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'

default_defaults = {

    'jobs': ('HGTEST_JOBS', multiprocessing.cpu_count()),

    'timeout': ('HGTEST_TIMEOUT', 360),

    'slowtimeout': ('HGTEST_SLOWTIMEOUT', 1500),

    'port': ('HGTEST_PORT', 20059),

    'shell': ('HGTEST_SHELL', 'sh'),

}

defaults = default_defaults.copy()

def canonpath(path):

    return os.path.realpath(os.path.expanduser(path))

def parselistfiles(files, listtype, warn=True):

    entries = dict()

    for filename in files:

        try:

            path = os.path.expanduser(os.path.expandvars(filename))

            f = open(path, "rb")

        except IOError as err:

            if err.errno != errno.ENOENT:

                raise

            if warn:

                print("warning: no such %s file: %s" % (listtype, filename))

            continue

        for line in f.readlines():

            line = line.split(b'#', 1)[0].strip()

            if line:

                # Ensure path entries are compatible with os.path.relpath()

                entries[os.path.normpath(line)] = filename

        f.close()

    return entries

def parsettestcases(path):

    """read a .t test file, return a set of test case names

    If path does not exist, return an empty set.

    """

    cases = []

    try:

        with open(path, 'rb') as f: