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

try:

    import subprocess

    subprocess.Popen  # trigger ImportError early

    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

except ImportError:

    subprocess = None

    from popen2 import Popen4

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

FAILED_PREFIX  = 'hghave check failed: '

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

defaults = {

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

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

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

}

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

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

    # Run installer in hg root

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

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

    global hgpkg

    hgpkg = _hgpath()

    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 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' or sys.platform.startswith('java'):

        tochild, fromchild = os.popen4(cmd)

        tochild.close()

        output = fromchild.read()

        ret = fromchild.close()

        if ret == None:

            ret = 0

    else:

        proc = 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))

        if not nodiff:

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