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

import difflib

import errno

import optparse

import os

import popen2

import shutil

import signal

import sys

import tempfile

import time

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

SKIPPED_STATUS = 80

SKIPPED_PREFIX = 'skipped: '

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

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

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

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

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("--with-hg", type="string",

    help="test existing install at given location")

parser.set_defaults(jobs=1, port=20059, timeout=180)

(options, args) = parser.parse_args()

verbose = options.verbose

coverage = options.cover or options.cover_stdlib or options.annotate

python = sys.executable

if options.jobs < 1:

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

    sys.exit(1)

if options.interactive and options.jobs > 1:

    print >> sys.stderr, 'ERROR: cannot mix -interactive and --jobs > 1'

    sys.exit(1)

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 vlog(*msg):

    if verbose:

        for m in msg:

            print m,

        print

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 extract_missing_features(lines):

    '''Extract missing/unknown features log lines as a list'''

    missing = []

    for line in lines:

        if not line.startswith(SKIPPED_PREFIX):

            continue

        line = line.splitlines()[0]

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

    return missing

def show_diff(expected, output):

    for line in difflib.unified_diff(expected, output,

            "Expected output", "Test output"):

        sys.stdout.write(line)

def find_program(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 check_required_tools():

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

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

    for p in required_tools:

        if os.name == 'nt':

            p += '.exe'

        found = find_program(p)

        if found:

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

        else:

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

def cleanup_exit():

    if not options.keep_tmpdir:

        if verbose:

            print "# Cleaning up HGTMP", HGTMP

        shutil.rmtree(HGTMP, True)

def use_correct_python():

    # 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 = find_program('python')

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

            return

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

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

    try:

        os.symlink(sys.executable, my_python)

    except AttributeError:

        # windows fallback

        shutil.copyfile(sys.executable, my_python)

        shutil.copymode(sys.executable, my_python)

def install_hg():

    global python

    vlog("# Performing temporary installation of HG")

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

    # Run installer in hg root

    os.chdir(os.path.join(os.path.dirname(sys.argv[0]), '..'))

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

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

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

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

    vlog("# Running", cmd)

    if os.system(cmd) == 0:

        if not verbose:

            os.remove(installerrs)

    else:

        f = open(installerrs)

        for line in f:

            print line,

        f.close()

        sys.exit(1)

    os.chdir(TESTDIR)

    os.environ["PATH"] = "%s%s%s" % (BINDIR, os.pathsep, os.environ["PATH"])

    pydir = os.pathsep.join([PYTHONDIR, TESTDIR])

    pythonpath = os.environ.get("PYTHONPATH")

    if pythonpath:

        pythonpath = pydir + os.pathsep + pythonpath

    else:

        pythonpath = pydir

    os.environ["PYTHONPATH"] = pythonpath

    use_correct_python()

    if coverage:

        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", "%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)

        python = '"%s" "%s" -x' % (sys.executable,

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

def output_coverage():

    vlog("# Producing coverage report")

    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)

    os.chdir(PYTHONDIR)

    cmd = '"%s" "%s" -i -r "--omit=%s"' % (

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

    vlog("# Running: "+cmd)

    os.system(cmd)

    if options.annotate:

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

        if not os.path.isdir(adir):

            os.mkdir(adir)

        cmd = '"%s" "%s" -i -a "--directory=%s" "--omit=%s"' % (

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

            adir, omit)

        vlog("# Running: "+cmd)

        os.system(cmd)

class Timeout(Exception):

    pass

def alarmed(signum, frame):

    raise Timeout

def run(cmd):

    """Run command in a sub-process, capturing the output (stdout and stderr).

    Return the exist code, and output."""

    # TODO: Use subprocess.Popen if we're running on Python 2.4

    if os.name == 'nt':

        tochild, fromchild = os.popen4(cmd)

        tochild.close()

        output = fromchild.read()

        ret = fromchild.close()

        if ret == None:

            ret = 0

    else:

        proc = popen2.Popen4(cmd)

        try:

            output = ''

            proc.tochild.close()

            output = proc.fromchild.read()

            ret = proc.wait()

            if os.WIFEXITED(ret):

                ret = os.WEXITSTATUS(ret)

        except Timeout:

            vlog('# Process %d timed out - killing it' % proc.pid)

            os.kill(proc.pid, signal.SIGTERM)

            ret = proc.wait()

            if ret == 0:

                ret = signal.SIGTERM << 8

            output += ("\n### Abort: timeout after %d seconds.\n"

                       % options.timeout)

    return ret, splitnewlines(output)

def run_one(test, skips, fails):

    '''tristate output:

    None -> skipped

    True -> passed

    False -> failed'''

    def skip(msg):

        if not verbose:

            skips.append((test, msg))

        else:

            print "\nSkipping %s: %s" % (test, msg)

        return None

    def fail(msg):

        fails.append((test, msg))

        print "\nERROR: %s %s" % (test, msg)

        return None

    vlog("# Test", test)

    # create a fresh hgrc

    hgrc = file(HGRCPATH, 'w+')

    hgrc.write('[ui]\n')

    hgrc.write('slash = True\n')

    hgrc.write('[defaults]\n')

    hgrc.write('backout = -d "0 0"\n')

    hgrc.write('commit = -d "0 0"\n')

    hgrc.write('debugrawcommit = -d "0 0"\n')

    hgrc.write('tag = -d "0 0"\n')

    hgrc.close()

    err = os.path.join(TESTDIR, test+".err")

    ref = os.path.join(TESTDIR, test+".out")

    testpath = os.path.join(TESTDIR, test)

    if os.path.exists(err):

        os.remove(err)       # Remove any previous output files

    # Make a tmp subdirectory to work in

    tmpd = os.path.join(HGTMP, test)

    os.mkdir(tmpd)

    os.chdir(tmpd)

    try:

        tf = open(testpath)

        firstline = tf.readline().rstrip()

        tf.close()

    except:

        firstline = ''

    lctest = test.lower()

    if lctest.endswith('.py') or firstline == '#!/usr/bin/env python':

        cmd = '%s "%s"' % (python, testpath)

    elif lctest.endswith('.bat'):

        # do not run batch scripts on non-windows

        if os.name != 'nt':

            return skip("batch script")

        # To reliably get the error code from batch files on WinXP,

        # the "cmd /c call" prefix is needed. Grrr

        cmd = 'cmd /c call "%s"' % testpath

    else:

        # do not run shell scripts on windows

        if os.name == 'nt':

            return skip("shell script")

        # do not try to run non-executable programs

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

            return skip("not executable")

        cmd = '"%s"' % testpath

    if options.timeout > 0:

        signal.alarm(options.timeout)

    vlog("# Running", cmd)

    ret, out = run(cmd)

    vlog("# Ret was:", ret)

    if options.timeout > 0:

        signal.alarm(0)

    skipped = (ret == SKIPPED_STATUS)

    diffret = 0

    # If reference output file exists, check test output against it

    if os.path.exists(ref):

        f = open(ref, "r")

        ref_out = splitnewlines(f.read())

        f.close()

    else:

        ref_out = []

    if not skipped and out != ref_out:

        diffret = 1

        fail("output changed")

        show_diff(ref_out, out)

    if skipped:

        missing = extract_missing_features(out)

        if not missing:

            missing = ['irrelevant']

        skip(missing[-1])