#!/usr/bin/env python
"""
Support package for the testsuite
"""

import unittest
import os
import shutil
import sys
import re
import subprocess
import difflib
import platform

exe_suffix = ''
pathsep = ":"

cygwin = 'uname' in os.__dict__ and os.uname()[0].find('CYGWIN') != -1
windows = cygwin or platform.system().find("Windows") != -1

if cygwin:
    exe_suffix = '.exe'
    pathsep = ":"
elif windows:
    exe_suffix = '.exe'
    pathsep = ";"

verbosity = 0

valgrind = None
redirect = True
if os.getenv("VALGRIND") == "leaks":
    valgrind = ["valgrind", "--tool=memcheck", "--num-callers=30",
                "-q",
                "--leak-check=full",
                "--show-reachable=no",
                "--leak-resolution=high",
                "--suppressions=%s/valgrind.supp" % os.getcwd(),
                "--suppressions=%s/valgrind-python.supp" % os.getcwd(),
                "--gen-suppressions=no"]
elif os.getenv("VALGRIND") == "yes":
    valgrind = ["valgrind", "--tool=memcheck", "--num-callers=30",
                "-q",
                "--suppressions=%s/valgrind.supp" % os.getcwd(),
                "--suppressions=%s/valgrind-python.supp" % os.getcwd(),
                "--gen-suppressions=no"]
elif os.getenv("GDB") == "yes":
    valgrind = ["gdb"]
    redirect = False


def find_on_path(execname):
    """Search for EXECNAME on the PATH, or return None"""

    dirs = os.getenv('PATH').split(pathsep)
    if pathsep != ":":
        # It seems to be difficult on Windows to know whether we are using
        # : or ;, so we just try both
        dirs.extend(os.getenv('PATH').split(":"))

    dirs.insert(0, os.getcwd())

    for dir in dirs:
        p = os.path.join(os.path.abspath(dir), execname)
        if os.path.exists(p) and not os.path.isdir(p):
            return p

        p = os.path.join(os.path.abspath(dir), execname + exe_suffix)
        if os.path.exists(p) and not os.path.isdir(p):
            return p

    return None


def cpp_supports_dump_xref():
    # Check whether we have "g++" next to "gcc", and assume it supports
    # "-fdump-xref" in this case
    p = find_on_path("gcc")
    if p:
        if os.path.exists(os.path.join(os.path.dirname(p), "g++")):
            return True
    return False


has_sqlite = file("../Makefile.conf").read().find("WITH_SQLITE=no") < 0
has_postgres = file("../Makefile.conf").read().find("WITH_POSTGRES=yes") > 0
has_iconv = file(
    os.path.join(os.getcwd(), "../gnatcoll_shared.gpr")).read().find(
        'Iconv : Yes_No := "yes";') >= 0
has_cpp = cpp_supports_dump_xref()
sqlite_shell = None
own_sqlite_shell = None


def requires_own_sqlite_shell(func):
    """
    This requires our own sqlite3 executable (so that versions match)
    """
    global own_sqlite_shell
    if own_sqlite_shell is None:
        own_sqlite_shell = os.path.join(
            os.getcwd(),
            "../src/sqlite/amalgamation/sqlite3_for_gps")
        if not os.path.exists(own_sqlite_shell):
            own_sqlite_shell = ''
        os.environ["SQLITE3_SHELL"] = own_sqlite_shell

    def wrapped(*args):
        if own_sqlite_shell:
            return func(*args)
        if hasattr(unittest, 'SkipTest'):
            raise unittest.SkipTest('Need sqlite3 shell')
    wrapped.__name__ = func.__name__
    return wrapped


def requires_sqlite_shell(func):
    """
    Search for the sqlite3 executable on the path, and run the test
    if found
    """
    global sqlite_shell
    if sqlite_shell is None:
        sqlite_shell = os.path.join(
            os.getcwd(),
            "../src/sqlite/amalgamation/sqlite3_for_gps")
        if not os.path.exists(sqlite_shell):
            sqlite_shell = find_on_path('sqlite3')
        os.environ["SQLITE3_SHELL"] = sqlite_shell

    def wrapped(*args):
        if sqlite_shell:
            return func(*args)
        if hasattr(unittest, 'SkipTest'):
            raise unittest.SkipTest('Need sqlite3 shell')
    wrapped.__name__ = func.__name__
    return wrapped


def requires_not_windows(func):
    """Only execute the test if not running on windows"""
    def wrapped(*args):
        if not windows:
            return func(*args)
        if hasattr(unittest, "SkipTest"):
            raise unittest.SkipTest("Not supported on Windows")
    wrapped.__name__ = func.__name__
    return wrapped


def requires_windows(func):
    """Only execute the test if running on windows"""
    def wrapped(*args):
        if windows:
            return func(*args)
        if hasattr(unittest, "SkipTest"):
            raise unittest.SkipTest("Supported only on Windows")
    wrapped.__name__ = func.__name__
    return wrapped


def requires_cpp(func):
    """Only execute the test if c++ is available"""
    def wrapped(*args):
        if has_cpp:
            return func(*args)
        if hasattr(unittest, "SkipTest"):
            raise unittest.SkipTest("Requires a c++ compiler")
    wrapped.__name__ = func.__name__
    return wrapped


def requires_iconv(func):
    """Only execute the test if iconv is available"""
    def wrapped(*args):
        if has_iconv:
            return func(*args)
        if hasattr(unittest, "SkipTest"):
            raise unittest.SkipTest("Requires iconv")
    wrapped.__name__ = func.__name__
    return wrapped


def requires_sqlite(func):
    """A decorator that skips a test if sqlite is not installed"""
    def wrapped(*args):
        if has_sqlite:
            return func(*args)
        if hasattr(unittest, "SkipTest"):
            raise unittest.SkipTest("No sqlite support")
    wrapped.__name__ = func.__name__
    return wrapped


def requires_postgresql(func):
    """A decorator that skips a test if postgres is not installed"""
    def wrapped(*args):
        if has_postgres:
            return func(*args)
        if hasattr(unittest, "SkipTest"):
            raise unittest.SkipTest("No postgreSQL support")
    wrapped.__name__ = func.__name__
    return wrapped


def not_in_continuous_builder(func):
    """A decorator that skips a test when in the continuous builder"""
    def wrapped(*args):
        if os.getenv("GPS_MONITOR") == "true":
            raise unittest.SkipTest("Not run in continuous builder")
        else:
            return func(*args)
    wrapped.__name__ = func.__name__
    return wrapped


class GNATCOLL_TestResult(unittest._TextTestResult):
    """Custom class to save test results for GAIA"""

    def __init__(self, stream, verbosity):
        super(GNATCOLL_TestResult, self).__init__(
            stream, 1, verbosity=verbosity)

        self.result_dir = os.path.join(os.getcwd(), "results")
        try:
            shutil.rmtree(self.result_dir, ignore_errors=True)
            os.makedirs(self.result_dir)
        except:
            pass

    def _do_out(self, test, short, err=None):
        name = test.id()
        file(os.path.join(self.result_dir, name + ".result"), "w").write(
            short)
        file(os.path.join(self.result_dir, "results"), "a").write(
            name + ":" + short + "\n")
        if err:
            file(os.path.join(self.result_dir, name + ".out"), "w").write(
                str(err[1]))

    def addSuccess(self, test):
        super(GNATCOLL_TestResult, self).addSuccess(test)
        self._do_out(test, "OK:test passed")

    def addError(self, test, err):
        super(GNATCOLL_TestResult, self).addError(test, err)
        self._do_out(test, "FAILED:driver error", err)

    def addFailure(self, test, err):
        super(GNATCOLL_TestResult, self).addFailure(test, err)
        self._do_out(test, "DIFF:test failed", err)

    def addSkip(self, test, reason):
        if sys.version_info >= (2, 7):
            super(GNATCOLL_TestResult, self).addSkip(test, reason)
        self._do_out(test, "DEAD:" + reason)

    def addExpectedFailure(self, test, err):
        if sys.version_info >= (2, 7):
            super(GNATCOLL_TestResult, self).addExpectedFailure(test, err)
        self._do_out(test, "XFAIL:test should have failed", err=err)


class GNATCOLL_TestRunner(unittest.TextTestRunner):
    def _makeResult(self):
        return GNATCOLL_TestResult(self.stream, verbosity=self.verbosity)

    def run(self, test):
        result = self._makeResult()
        test(result)
        if verbosity >= 1:
            self.stream.write("(%d tests) " % result.testsRun)
        if not result.wasSuccessful():
            self.stream.writeln("FAILED")
        result.printErrors()


class GNATCOLL_TestCase(unittest.TestCase):

    def find_on_path(self, execname):
        """Search for EXECNAME on the PATH"""

        p = find_on_path(execname)
        if p is None:
            self.assertFalse(True, "Exec not found on PATH: %s" % execname)
            return execname
        return p

    def gprbuild(self, project="default.gpr"):
        self.runexec(['gprbuild', '-q', '-m', '-p', '-g', '-P%s' % project])

    def gcc(self, file):
        self.runexec(['gcc', '-c', '-gnatc', file])

    def runexec(self, cmdline, expected=None, msg="", sorted=False,
                capture_output=True,
                shell=False, expectedStatus=0, input='',
                customFilter=None):
        """Run the givne command, and compare its output to EXPECTED. The
           latter can also be the name of a file.
           First element in CMDLINE is converted to full path if necessary.
           No comparison done if EXPECTED is None.
           If SORTED is True, the actual output lines are first sorted before
           comparing with the expected output.
           :param customFilter: see diff()
        """

        if isinstance(cmdline, str):
            cmdline = [cmdline]
        if os.path.exists(cmdline[0]):
            if os.path.dirname(cmdline[0]) == '':
                cmdline[0] = "./%s" % cmdline[0]
        else:
            cmdline[0] = self.find_on_path(cmdline[0])

        if valgrind and "gprbuild" not in cmdline[0]:
            cmdline = valgrind + cmdline

        if input:
            stdin = subprocess.PIPE
        else:
            stdin = None

        if capture_output and redirect:
            stdout = subprocess.PIPE
            stderr = subprocess.STDOUT
        else:
            stdout = stderr = None

        try:
            if verbosity >= 3:
                print "Running %s" % cmdline
            proc = subprocess.Popen(cmdline, stdout=stdout,
                                    shell=shell,
                                    stdin=stdin,
                                    stderr=stderr)
        except:
            self.assertTrue(False, "Could not execute %s" % cmdline)

        out = proc.communicate(input=input)[0]

        status = proc.wait()
        self.assertEqual(
            expectedStatus, status, "Failed to run %s\n%s\nstatus=%s" % (
                cmdline, out, status))

        if expected is not None:
            self.diff(expected=expected, actual=out, msg=msg, sorted=sorted,
                      customFilter=customFilter)

        if verbosity >= 3:
            print out

        return out

    def diff(self, expected, actual, msg="", sorted=False, customFilter=None):
        """
          :param customFilter: a subprogram that takes the actual output, and
             returns a cleaned up version.
        """
        if os.path.isfile(expected):
            expected = file(expected).read()

        pwd = os.getcwd() + "/"
        actual = actual.replace(pwd, "<pwd>")

        if windows:
            pwd = pwd.replace("/", "\\")
            pwd = re.sub(r'^[cC]:', '', pwd)
            actual = actual.replace(pwd, "<pwd>")
            actual = re.sub(r'[cC]:<pwd>', '<pwd>', actual)
            actual = actual.replace("\r", "")

        if customFilter:
            actual = customFilter(actual)

        act = actual.splitlines()

        if sorted:
            act.sort()

        d = difflib.unified_diff(expected.splitlines(), act)
        d = "\n".join(d)
        if d != "":
            self.assertTrue(False, msg + "\n" + d)

    def unlink_if_exists(self, files):
        if isinstance(files, str):
            files = [files]
        for f in files:
            try:
                os.unlink(f)
            except OSError:
                pass

    def create_fake_bb_compiler(
            self, comp_dir, comp_target, gnat_version, gcc_version,
            runtimes=["zfp", "ravenscar"]):
        """
        Create a fake compiler in a local directory. This is mostly used
        to emulate having multiple compilers on the PATH, for instance
        bareboard.
        """

        comp_prefix = comp_target + '-'

        comp_dict = {'comp_dir': comp_dir,
                     'comp_target': comp_target,
                     'gnat_version': gnat_version,
                     'gcc_version': gcc_version,
                     'comp_prefix': comp_prefix,
                     'exeext': exe_suffix}

        try:
            os.makedirs(os.path.join(comp_dir, 'bin'))
        except OSError:
            pass
        gnatls_adb = open(os.path.join(comp_dir, 'bin', 'gnatls.adb'), 'w')
        gnatls_adb.write("""
    with Text_IO; use Text_IO;
    with GNAT.Command_Line; use GNAT.Command_Line;
    procedure gnatls is
       RTS : access String;
    begin
       loop
          case Getopt ("v -RTS=:") is
             when ASCII.NUL => exit;
             when 'v' =>
                Put_Line ("GNATLS Pro %(gnat_version)s (20070123-34)");
             when '-' =>
                RTS := new String'(Parameter);
             when others =>
                null;
          end case;
       end loop;
       if RTS /= null then
          Put_Line ("Source Search Path:");
          Put_Line ("   <Current_Directory>");
          Put_Line
             ("   %(comp_dir)s/%(comp_target)s/lib/gnat/" & RTS.all
              & "/adainclude");
          New_Line;
          Put_Line ("Object Search Path:");
          Put_Line ("   <Current_Directory>");
          Put_Line
             ("   %(comp_dir)s/%(comp_target)s/lib/gnat/" & RTS.all
              & "/adalib");
       end if;
    exception
       when others => null;
    end gnatls;
    """ % comp_dict)
        gnatls_adb.close()

        gcc_adb = open(os.path.join(comp_dir, 'bin', 'gcc.adb'), 'w')
        gcc_adb.write("""
    with Ada.Text_IO; use Ada.Text_IO;
    with Ada.Command_Line; use Ada.Command_Line;
    procedure gcc is
    begin
       if Argument_Count >= 1 and then Argument (1) = "-v" then
            Put_Line ("gcc version %(gcc_version)s 20131008 for GNAT Pro");
       elsif Argument_Count >= 1 and then Argument (1) = "--version" then
            Put_Line ("gcc (GCC) %(gcc_version)s");
       elsif Argument_Count >= 1 and then Argument (1) = "-dumpmachine" then
            Put_Line ("%(comp_target)s");
       else
             Put ("Running gcc");
             for J in 1 .. Argument_Count loop
                 Put (" " & Argument (J));
             end loop;
       end if;
    end gcc;
""" % comp_dict)
        gcc_adb.close()

        gnatmake_adb = open(os.path.join(comp_dir, 'bin', 'gnatmake.adb'), 'w')
        gnatmake_adb.write("""
    with Ada.Text_IO; use Ada.Text_IO;
    with Ada.Command_Line; use Ada.Command_Line;
    procedure gnatmake is
    begin
             Put ("Running gcc");
             for J in 1 .. Argument_Count loop
                 Put (" " & Argument (J));
             end loop;
    end gnatmake;
    """)
        gnatmake_adb.close()

        # Build the dummy tools (rename them in a second those, so as not
        # to interfer with the build process of the other tools)
        pwd = os.getcwd()
        os.chdir(os.path.join(comp_dir, "bin"))
        for tool in ['gnatmake', 'gcc', 'gnatls']:
            comp_dict['bin'] = tool
            self.runexec(
                ['gnatmake', tool + '.adb', '-o',
                 'dummy-%(comp_prefix)s%(bin)s%(exeext)s' % comp_dict])
        for tool in ['gnatmake', 'gcc', 'gnatls']:
            comp_dict['bin'] = tool
            os.rename(
                'dummy-%(comp_prefix)s%(bin)s%(exeext)s' % comp_dict,
                '%(comp_prefix)s%(bin)s%(exeext)s' % comp_dict)
        os.chdir(pwd)

        # We need access to a native gprconfig
        gprconfig = find_on_path('gprconfig')
        dest = os.path.join(comp_dir, 'bin', 'gprconfig')
        self.unlink_if_exists(dest)
        os.symlink(gprconfig, dest)

        dest = os.path.join(comp_dir, 'share')
        self.unlink_if_exists(dest)
        os.symlink(os.path.join(gprconfig, '..', '..', 'share'), dest)

        libdir = os.path.join(comp_dir, comp_target, 'lib', 'gnat')

        for runtime in runtimes:
            for dir in ("adalib", "adainclude"):
                try:
                    os.makedirs(os.path.join(libdir, runtime, dir))
                except OSError:
                    pass
            open(os.path.join(libdir, runtime, 'adainclude', 'system.ads'),
                 'w').write("")


def chdir(in_dir):
    """A decorator that will temporarily change to another directory
       The directory is relative to the containing module's directory.
    """
    def wrap(func):
        def wrapped(*args):
            try:
                pwd = os.getcwd()
                d = os.path.join(func.__module__, in_dir)
                if verbosity >= 3:
                    print "chdir %s" % d
                os.chdir(d)
                func(*args)
            finally:
                if verbosity >= 3:
                    print "chdir %s" % pwd
                os.chdir(pwd)
        wrapped.__name__ = func.__name__
        return wrapped
    return wrap