#!/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.)

import argparse

import collections

import contextlib

import difflib

import errno

import functools

import json

import multiprocessing

import os

import platform

import queue

import random

import re

import shlex

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

if sys.version_info < (3, 8, 0):

    print(

        '%s is only supported on Python 3.8+, not %s'

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

    )

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

MACOS = sys.platform == 'darwin'

WINDOWS = os.name == r'nt'

shellquote = shlex.quote

# The number of HGPORTx ports allocated to each test.

HGPORT_COUNT = 4

RUNTEST_DIR = os.path.abspath(os.path.dirname(__file__.encode('utf-8')))

RUNTEST_DIR_FORWARD_SLASH = RUNTEST_DIR.replace(os.sep.encode('utf-8'), b'/')

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

    from pygments.console import ansiformat

    if WINDOWS:

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

    def colorize(text: str, style: str, color: bool = True) -> str:

        if not color:

            return text

        return ansiformat(style, text)

except ImportError:

    def colorize(text: str, style: str, color: bool = True) -> str:

        return text

progress_type = {}

if pygmentspresent:

    _T_ERROR = token.string_to_tokentype("Token.Generic.Error")

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

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

    _T_IGNORED = token.string_to_tokentype("Token.Generic.Ignored")

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

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

    _T_SKIPPED_DOT = token.string_to_tokentype("Token.Generic.SkippedDot")

    _T_SUCCESS = token.string_to_tokentype("Token.Generic.Success")

    _T_TIMEDOUT = token.string_to_tokentype("Token.Generic.TimedOut")

    class TestRunnerStyle(style.Style):

        default_style = ""

        styles = {

            _T_ERROR: '#cd00cd',

            _T_FAILED: '#7f0000',

            _T_FNAME: '#ff0000',

            _T_IGNORED: '#cdcd00',

            _T_SKIPPED: '#e5e5e5',

            _T_SNAME: '#00ffff',

            _T_SKIPPED_DOT: '#00ffff',

            _T_SUCCESS: '#00cd00',

            _T_TIMEDOUT: '#ff00ff',

        }

    class TestRunnerLexer(lexer.RegexLexer):

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

        tokens = {

            'root': [

                (r'^Skipped', _T_SKIPPED, 'skipped'),

                (r'^Failed ', _T_FAILED, 'failed'),

                (r'^ERROR: ', _T_FAILED, 'failed'),

            ],

            'skipped': [

                (testpattern, _T_SNAME),

                (r':.*', _T_SKIPPED),

            ],

            'failed': [

                (testpattern, _T_FNAME),

                (r'(:| ).*', _T_FAILED),

            ],

        }

    progress_type['.'] = _T_SUCCESS

    progress_type['s'] = _T_SKIPPED_DOT

    progress_type['i'] = _T_IGNORED

    progress_type['!'] = _T_FAILED

    progress_type['E'] = _T_ERROR

    progress_type['t'] = _T_TIMEDOUT

    class progressLexer(lexer.RegexLexer):

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

        tokens = {

            'root': [

                (r'^Skipped', _T_SKIPPED, 'skipped'),

                (r'^Failed ', _T_FAILED, 'failed'),

                (r'^ERROR: ', _T_FAILED, 'failed'),

            ],

            'skipped': [

                (testpattern, _T_SNAME),

                (r':.*', _T_SKIPPED),

            ],

            'failed': [

                (testpattern, _T_FNAME),

                (r'(:| ).*', _T_FAILED),

            ],

        }

    runnerformatter = formatters.Terminal256Formatter(style=TestRunnerStyle)

    runnerlexer = TestRunnerLexer()

origenviron = os.environ.copy()

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

original_env = os.environ.copy()

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:

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

    getcwdb = lambda: _sys2bytes(os.getcwd())

if WINDOWS:

    _getcwdb = getcwdb

    def getcwdb():

        cwd = _getcwdb()

        if re.match(b'^[a-z]:', cwd):

            # os.getcwd() is inconsistent on the capitalization of the drive

            # letter, so adjust it. see https://bugs.python.org/issue40368

            cwd = cwd[0:1].upper() + cwd[1:]

        return cwd

# 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, OSError) as exc:

        if exc.errno == errno.EADDRINUSE:

            return True

        elif exc.errno in (

            errno.EADDRNOTAVAIL,

            errno.EPROTONOSUPPORT,

            errno.EAFNOSUPPORT,

        ):

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

        return False

    except socket.error as exc:

        if WINDOWS and exc.errno == errno.WSAEACCES:

            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

default_defaults = {

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

    'timeout': ('HGTEST_TIMEOUT', 360 if not WINDOWS else 360 * 4),

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

            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:

            for l in f:

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

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

    except FileNotFoundError:

        pass

    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(

        "--shard-total",

        type=int,

        default=None,

        help="total number of shard to use (enable sharding)",

    )

    selection.add_argument(

        "--shard-index",

        type=int,

        default=None,

        help="index of this shard [1-N]",

    )

    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(

        "-d",

        "--debug",

        action="store_true",