#!/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, incorporated herein by reference.

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

#

# (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 difflib

import errno

import optparse

import os

import subprocess

import shutil

import signal

import sys

import tempfile

import time

closefds = os.name == 'posix'

def Popen4(cmd, bufsize=-1):

    p = subprocess.Popen(cmd, shell=True, bufsize=bufsize,

                         close_fds=closefds,

                         stdin=subprocess.PIPE, stdout=subprocess.PIPE,

                         stderr=subprocess.STDOUT)

    p.fromchild = p.stdout

    p.tochild = p.stdin

    p.childerr = p.stderr

    return p

# reserved exit code to skip test (used by hghave)

SKIPPED_STATUS = 80

SKIPPED_PREFIX = 'skipped: '

FAILED_PREFIX  = 'hghave check failed: '

PYTHON = sys.executable

requiredtools = ["python", "diff", "grep", "unzip", "gunzip", "bunzip2", "sed"]

defaults = {

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

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

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

}

def parseargs():

    parser = optparse.OptionParser("%prog [options] [tests]")

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

        help="output files annotated with coverage")

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

        help="run as child process, summary to given fd")

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

        help="print a test coverage report")

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

        help="exit on the first test failure")

    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"

             " (best used with --tmpdir)")

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

        help="restart at last error")

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

        help="port on which servers should listen"

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

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

        help="retest failed tests")

    parser.add_option("-s", "--cover_stdlib", action="store_true",

        help="print a test coverage report inc. standard libraries")

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

        help="kill errant tests after TIMEOUT seconds"

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

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

        help="run tests in the given temporary directory")

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

        help="output verbose messages")

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

        help="skip showing test changes")

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

        metavar="HG",

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

             "temporary installation")

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

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

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

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

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

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

    for option, default in defaults.items():

        defaults[option] = int(os.environ.get(*default))

    parser.set_defaults(**defaults)

    (options, args) = parser.parse_args()

    if options.with_hg:

        if not (os.path.isfile(options.with_hg) and

                os.access(options.with_hg, os.X_OK)):

            parser.error('--with-hg must specify an executable hg script')

        if not os.path.basename(options.with_hg) == 'hg':

            sys.stderr.write('warning: --with-hg should specify an hg script')

    if options.local:

        testdir = os.path.dirname(os.path.realpath(sys.argv[0]))

        hgbin = os.path.join(os.path.dirname(testdir), 'hg')

        if not os.access(hgbin, os.X_OK):

            parser.error('--local specified, but %r not found or not executable'

                         % hgbin)

        options.with_hg = hgbin

    options.anycoverage = (options.cover or

                           options.cover_stdlib or

                           options.annotate)

    if options.anycoverage and options.with_hg:

        # I'm not sure if this is a fundamental limitation or just a

        # bug.  But I don't want to waste people's time and energy doing

        # test runs that don't give the results they want.

        parser.error("sorry, coverage options do not work when --with-hg "

                     "or --local specified")

    global vlog

    if options.verbose:

        if options.jobs > 1 or options.child is not None:

            pid = "[%d]" % os.getpid()

        else:

            pid = None

        def vlog(*msg):

            if pid:

                print pid,

            for m in msg:

                print m,

            print

    else:

        vlog = lambda *msg: None

    if options.jobs < 1:

        print >> sys.stderr, 'ERROR: -j/--jobs must be positive'

        sys.exit(1)

    if options.interactive and options.jobs > 1:

        print '(--interactive overrides --jobs)'

        options.jobs = 1

    if options.py3k_warnings:

        if sys.version_info[:2] < (2, 6) or sys.version_info[:2] >= (3, 0):

            print 'ERROR: Py3k warnings switch can only be used on Python 2.6+'

            sys.exit(1)

    return (options, args)

def rename(src, dst):

    """Like os.rename(), trade atomicity and opened files friendliness

    for existing destination support.

    """

    shutil.copy(src, dst)

    os.remove(src)

def splitnewlines(text):

    '''like str.splitlines, but only split on newlines.

    keep line endings.'''

    i = 0

    lines = []

    while True:

        n = text.find('\n', i)

        if n == -1:

            last = text[i:]

            if last:

                lines.append(last)

            return lines

        lines.append(text[i:n+1])

        i = n + 1

def parsehghaveoutput(lines):

    '''Parse hghave log lines.

    Return tuple of lists (missing, failed):

      * the missing/unknown features

      * the features for which existence check failed'''

    missing = []

    failed = []

    for line in lines:

        if line.startswith(SKIPPED_PREFIX):

            line = line.splitlines()[0]

            missing.append(line[len(SKIPPED_PREFIX):])

        elif line.startswith(FAILED_PREFIX):

            line = line.splitlines()[0]

            failed.append(line[len(FAILED_PREFIX):])

    return missing, failed

def showdiff(expected, output):

    for line in difflib.unified_diff(expected, output,

            "Expected output", "Test output"):

        sys.stdout.write(line)

def findprogram(program):

    """Search PATH for a executable program"""

    for p in os.environ.get('PATH', os.defpath).split(os.pathsep):

        name = os.path.join(p, program)

        if os.access(name, os.X_OK):

            return name

    return None

def checktools():

    # Before we go any further, check for pre-requisite tools

    # stuff from coreutils (cat, rm, etc) are not tested

    for p in requiredtools:

        if os.name == 'nt':

            p += '.exe'

        found = findprogram(p)

        if found:

            vlog("# Found prerequisite", p, "at", found)

        else:

            print "WARNING: Did not find prerequisite tool: "+p

def cleanup(options):

    if not options.keep_tmpdir:

        vlog("# Cleaning up HGTMP", HGTMP)

        shutil.rmtree(HGTMP, True)

def usecorrectpython():

    # some tests run python interpreter. they must use same

    # interpreter we use or bad things will happen.

    exedir, exename = os.path.split(sys.executable)

    if exename == 'python':

        path = findprogram('python')

        if os.path.dirname(path) == exedir:

            return

    vlog('# Making python executable in test path use correct Python')

    mypython = os.path.join(BINDIR, 'python')

    try:

        os.symlink(sys.executable, mypython)

    except AttributeError:

        # windows fallback

        shutil.copyfile(sys.executable, mypython)

        shutil.copymode(sys.executable, mypython)

def installhg(options):

    vlog("# Performing temporary installation of HG")

    installerrs = os.path.join("tests", "install.err")

    pure = options.pure and "--pure" or ""

    # Run installer in hg root

    script = os.path.realpath(sys.argv[0])

    hgroot = os.path.dirname(os.path.dirname(script))

    os.chdir(hgroot)

    cmd = ('%s setup.py %s clean --all'

           ' install --force --prefix="%s" --install-lib="%s"'

           ' --install-scripts="%s" >%s 2>&1'

           % (sys.executable, pure, INST, PYTHONDIR, BINDIR, installerrs))

    vlog("# Running", cmd)

    if os.system(cmd) == 0:

        if not options.verbose:

            os.remove(installerrs)

    else:

        f = open(installerrs)

        for line in f:

            print line,

        f.close()

        sys.exit(1)

    os.chdir(TESTDIR)

    usecorrectpython()

    vlog("# Installing dummy diffstat")

    f = open(os.path.join(BINDIR, 'diffstat'), 'w')

    f.write('#!' + sys.executable + '\n'

            'import sys\n'

            'files = 0\n'

            'for line in sys.stdin:\n'

            '    if line.startswith("diff "):\n'

            '        files += 1\n'

            'sys.stdout.write("files patched: %d\\n" % files)\n')

    f.close()

    os.chmod(os.path.join(BINDIR, 'diffstat'), 0700)

    if options.py3k_warnings and not options.anycoverage:

        vlog("# Updating hg command to enable Py3k Warnings switch")

        f = open(os.path.join(BINDIR, 'hg'), 'r')

        lines = [line.rstrip() for line in f]

        lines[0] += ' -3'

        f.close()

        f = open(os.path.join(BINDIR, 'hg'), 'w')

        for line in lines:

            f.write(line + '\n')

        f.close()

    if options.anycoverage:

        vlog("# Installing coverage wrapper")

        os.environ['COVERAGE_FILE'] = COVERAGE_FILE

        if os.path.exists(COVERAGE_FILE):

            os.unlink(COVERAGE_FILE)

        # Create a wrapper script to invoke hg via coverage.py

        os.rename(os.path.join(BINDIR, "hg"), os.path.join(BINDIR, "_hg.py"))

        f = open(os.path.join(BINDIR, 'hg'), 'w')

        f.write('#!' + sys.executable + '\n')

        f.write('import sys, os; os.execv(sys.executable, [sys.executable, '

                '"%s", "-x", "-p", "%s"] + sys.argv[1:])\n' %

                (os.path.join(TESTDIR, 'coverage.py'),

                 os.path.join(BINDIR, '_hg.py')))

        f.close()

        os.chmod(os.path.join(BINDIR, 'hg'), 0700)

def outputcoverage(options):

    vlog('# Producing coverage report')

    os.chdir(PYTHONDIR)

    def covrun(*args):

        start = sys.executable, os.path.join(TESTDIR, 'coverage.py')

        cmd = '"%s" "%s" %s' % (start[0], start[1], ' '.join(args))

        vlog('# Running: %s' % cmd)

        os.system(cmd)

    omit = [BINDIR, TESTDIR, PYTHONDIR]

    if not options.cover_stdlib:

        # Exclude as system paths (ignoring empty strings seen on win)

        omit += [x for x in sys.path if x != '']

    omit = ','.join(omit)

    covrun('-c') # combine from parallel processes

    for fn in os.listdir(TESTDIR):

        if fn.startswith('.coverage.'):

            os.unlink(os.path.join(TESTDIR, fn))

    covrun('-i', '-r', '"--omit=%s"' % omit) # report

    if options.annotate:

        adir = os.path.join(TESTDIR, 'annotated')

        if not os.path.isdir(adir):

            os.mkdir(adir)

        covrun('-i', '-a', '"--directory=%s"' % adir, '"--omit=%s"' % omit)

class Timeout(Exception):

    pass