#!/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 difflib

import distutils.version as version

import errno

import json

import optparse

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

if os.environ.get('RTUNICODEPEDANTRY', False):

    try:

        reload(sys)

        sys.setdefaultencoding("undefined")

    except NameError:

        pass

origenviron = os.environ.copy()

osenvironb = getattr(os, 'environb', os.environ)

processlock = threading.Lock()

with_color = 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.formatters as formatters

        with_color = True

    except ImportError:

        pass

if not sys.stderr.isatty(): # check if the terminal is capable

    with_color = False

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

        return p.encode('utf-8')

    def _strpath(p):

        return p.decode('utf-8')

elif sys.version_info >= (3, 0, 0):

    print('%s is only supported on Python 3.5+ and 2.7, not %s' %

          (sys.argv[0], '.'.join(str(v) for v in sys.version_info[:3])))

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

else:

    PYTHON3 = False

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

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

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

    # help keep code consistent between platforms.

    def _bytespath(p):

        return p

    _strpath = _bytespath

# 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(cmd, shell=True, bufsize=-1, cwd=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

PYTHON = _bytespath(sys.executable.replace('\\', '/'))

IMPL_PATH = b'PYTHONPATH'

if 'java' in sys.platform:

    IMPL_PATH = b'JYTHONPATH'

defaults = {

    'jobs': ('HGTEST_JOBS', 1),

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

    'slowtimeout': ('HGTEST_SLOWTIMEOUT', 500),

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

    try:

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

            for l in f:

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

                    cases.update(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 = optparse.OptionParser("%prog [options] [tests]")

    # keep these sorted

    parser.add_option("--blacklist", action="append",

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

    parser.add_option("--whitelist", action="append",

        help="always run tests listed in the specified whitelist file")

    parser.add_option("--changed", type="string",

        help="run tests that are changed in parent rev or working directory")

    parser.add_option("-C", "--annotate", action="store_true",

        help="output files annotated with coverage")

    parser.add_option("-c", "--cover", action="store_true",

        help="print a test coverage report")

    parser.add_option("--color", choices=["always", "auto", "never"],

                      default="auto",

                      help="colorisation: always|auto|never (default: auto)")

    parser.add_option("-d", "--debug", action="store_true",

        help="debug mode: write output of test scripts to console"

             " rather than capturing and diffing it (disables timeout)")

    parser.add_option("-f", "--first", action="store_true",

        help="exit on the first test failure")

    parser.add_option("-H", "--htmlcov", action="store_true",

        help="create an HTML report of the coverage of the files")

    parser.add_option("-i", "--interactive", action="store_true",

        help="prompt to accept changed output")

    parser.add_option("-j", "--jobs", type="int",

        help="number of jobs to run in parallel"

             " (default: $%s or %d)" % defaults['jobs'])

    parser.add_option("--keep-tmpdir", action="store_true",

        help="keep temporary directory after running tests")

    parser.add_option("-k", "--keywords",

        help="run tests matching keywords")

    parser.add_option("--list-tests", action="store_true",

        help="list tests instead of running them")

    parser.add_option("-l", "--local", action="store_true",

        help="shortcut for --with-hg=<testdir>/../hg, "

             "and --with-chg=<testdir>/../contrib/chg/chg if --chg is set")

    parser.add_option("--loop", action="store_true",

        help="loop tests repeatedly")

    parser.add_option("--runs-per-test", type="int", dest="runs_per_test",

        help="run each test N times (default=1)", default=1)

    parser.add_option("-n", "--nodiff", action="store_true",

        help="skip showing test changes")

    parser.add_option("--outputdir", type="string",

        help="directory to write error logs to (default=test directory)")

    parser.add_option("-p", "--port", type="int",

        help="port on which servers should listen"

             " (default: $%s or %d)" % defaults['port'])

    parser.add_option("--compiler", type="string",

        help="compiler to build with")

    parser.add_option("--pure", action="store_true",

        help="use pure Python code instead of C extensions")

    parser.add_option("-R", "--restart", action="store_true",

        help="restart at last error")

    parser.add_option("-r", "--retest", action="store_true",

        help="retest failed tests")

    parser.add_option("-S", "--noskips", action="store_true",

        help="don't report skip tests verbosely")

    parser.add_option("--shell", type="string",

        help="shell to use (default: $%s or %s)" % defaults['shell'])

    parser.add_option("-t", "--timeout", type="int",

        help="kill errant tests after TIMEOUT seconds"

             " (default: $%s or %d)" % defaults['timeout'])

    parser.add_option("--slowtimeout", type="int",

        help="kill errant slow tests after SLOWTIMEOUT seconds"

             " (default: $%s or %d)" % defaults['slowtimeout'])

    parser.add_option("--time", action="store_true",

        help="time how long each test takes")

    parser.add_option("--json", action="store_true",

                      help="store test result data in 'report.json' file")

    parser.add_option("--tmpdir", type="string",

        help="run tests in the given temporary directory"

             " (implies --keep-tmpdir)")

    parser.add_option("-v", "--verbose", action="store_true",

        help="output verbose messages")

    parser.add_option("--xunit", type="string",

                      help="record xunit results at specified path")

    parser.add_option("--view", type="string",

        help="external diff viewer")

    parser.add_option("--with-hg", type="string",

        metavar="HG",

        help="test using specified hg script rather than a "

             "temporary installation")

    parser.add_option("--chg", action="store_true",

                      help="install and use chg wrapper in place of hg")

    parser.add_option("--with-chg", metavar="CHG",

                      help="use specified chg wrapper in place of hg")

    parser.add_option("--ipv6", action="store_true",

                      help="prefer IPv6 to IPv4 for network related tests")

    parser.add_option("-3", "--py3k-warnings", action="store_true",

        help="enable Py3k warnings on Python 2.7+")

    # This option should be deleted once test-check-py3-compat.t and other

    # Python 3 tests run with Python 3.

    parser.add_option("--with-python3", metavar="PYTHON3",

                      help="Python 3 interpreter (if running under Python 2)"

                           " (TEMPORARY)")

    parser.add_option('--extra-config-opt', action="append",

                      help='set the given config opt in the test hgrc')

    parser.add_option('--random', action="store_true",

                      help='run tests in random order')

    parser.add_option('--profile-runner', action='store_true',

                      help='run statprof on run-tests')

    parser.add_option('--allow-slow-tests', action='store_true',

                      help='allow extremely slow tests')

    parser.add_option('--showchannels', action='store_true',

                      help='show scheduling channels')

    parser.add_option('--known-good-rev', type="string",

                      metavar="known_good_rev",

                      help=("Automatically bisect any failures using this "

                            "revision as a known-good revision."))

    for option, (envvar, default) in defaults.items():

        defaults[option] = type(default)(os.environ.get(envvar, default))

    parser.set_defaults(**defaults)