#!/usr/bin/env python

#

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

#

# Copyright 2006 Matt Mackall <mpm@selenic.com>

#

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

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

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

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

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

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

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

#

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

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

# sample of test scripts.  For example:

#

#  1) serial, no coverage, temp install:

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

#  2) serial, no coverage, local hg:

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

#  3) serial, coverage, temp install:

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

#  4) serial, coverage, local hg:

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

#  5) parallel, no coverage, temp install:

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

#  6) parallel, no coverage, local hg:

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

#  7) parallel, coverage, temp install:

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

#  8) parallel, coverage, local install:

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

#  9) parallel, custom tmp dir:

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

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

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

#

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

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

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

# includes some scripts that run daemon processes.)

from __future__ import absolute_import, print_function

import argparse

import collections

import difflib

import distutils.version as version

import errno

import json

import multiprocessing

import os

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

# 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 = {

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

        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(

        _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

            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'

defaults = {

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

    'timeout': ('HGTEST_TIMEOUT', 180),

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

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

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

}

def canonpath(path):

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

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

    entries = dict()

    for filename in files:

        try:

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

            f = open(path, "rb")

        except IOError as err:

            if err.errno != errno.ENOENT:

                raise

            if warn:

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

            continue

        for line in f.readlines():

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

            if line:

                entries[line] = filename

        f.close()

    return entries

def parsettestcases(path):

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

    If path does not exist, return an empty set.

    """

    cases = []

    try:

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

            for l in f:

                if l.startswith(b'#testcases '):

                    cases.append(sorted(l[11:].split()))

    except IOError as ex:

        if ex.errno != errno.ENOENT:

            raise

    return cases

def getparser():

    """Obtain the OptionParser used by the CLI."""

    parser = argparse.ArgumentParser(usage='%(prog)s [options] [tests]')

    selection = parser.add_argument_group('Test Selection')

    selection.add_argument(

        '--allow-slow-tests',

        action='store_true',

        help='allow extremely slow tests',

    )

    selection.add_argument(

        "--blacklist",

        action="append",

        help="skip tests listed in the specified blacklist file",

    )