#!/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, 5, 0):

    print(

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

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

    )

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

WINDOWS = os.name == r'nt'

shellquote = shlex.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 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()

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

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:

        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 socket.error as exc:

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

            return False

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

        elif 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 which(exe):

    # shutil.which only accept bytes from 3.8

    cmd = _bytes2sys(exe)

    real_exec = shutil.which(cmd)

    return _sys2bytes(real_exec)

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 = []