#!/usr/bin/python

# CDDL HEADER START

#

# The contents of this file are subject to the terms of the

# Common Development and Distribution License (the "License").

# You may not use this file except in compliance with the License.

#

# You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE

# or http://www.opensolaris.org/os/licensing.

# See the License for the specific language governing permissions

# and limitations under the License.

#

# When distributing Covered Code, include this CDDL HEADER in each

# file and include the License file at usr/src/OPENSOLARIS.LICENSE.

# If applicable, add the following below this CDDL HEADER, with the

# fields enclosed by brackets "[]" replaced with your own identifying

# information: Portions Copyright [yyyy] [name of copyright owner]

#

# CDDL HEADER END

#

# Copyright (c) 2008, 2011, Oracle and/or its affiliates. All rights reserved.

import baseline

import ConfigParser

import copy

import difflib

import errno

import gettext

import hashlib

import logging

import os

import pprint

import shutil

import signal

import stat

import subprocess

import sys

import tempfile

import time

import unittest

import platform

import pwd

import re

import textwrap

EmptyI = tuple()

EmptyDict = dict()

# relative to our proto area

path_to_pub_util = "../../src/util/publish"

#

# These are initialized by pkg5testenv.setup_environment.

#

g_proto_area = "TOXIC"

# User's value for TEMPDIR

g_tempdir = "/tmp"

g_debug_output = False

if "DEBUG" in os.environ:

        g_debug_output = True

#

# XXX?

#

gettext.install("pkg", "/usr/share/locale")

OUTPUT_DOTS = 0         # Dots ...

OUTPUT_VERBOSE = 1      # Verbose

OUTPUT_PARSEABLE = 2    # Machine readable

class TestStopException(Exception):

        """An exception used to signal that all testing should cease.

        This is a framework-internal exception that tests should not

        raise"""

        pass

class TestSkippedException(Exception):

        """An exception used to signal that a test was skipped.

        Should be initialized with a string giving a more detailed

        reason.  Test cases can raise this to the framework

        that some prerequisite of the test is unsatisfied.  A string

        explaining the error should be passed at construction.  """

        def __str__(self):

                return "Test Skipped: " + " ".join(self.args)

#

# Errors for which the traceback is likely not useful.

#

import pkg.depotcontroller as depotcontroller

import pkg.portable as portable

import pkg.client.api

import pkg.client.progress

# Version test suite is known to work with.

PKG_CLIENT_NAME = "pkg"

ELIDABLE_ERRORS = [ TestSkippedException, depotcontroller.DepotStateException ]

class Pkg5CommonException(AssertionError):

        def __init__(self, com = ""):

                Pkg5TestCase.failureException.__init__(self, com)

        topdivider = \

        ",---------------------------------------------------------------------\n"

        botdivider = \

        "`---------------------------------------------------------------------\n"

        def format_comment(self, comment):

                if comment is not None:

                        comment = comment.expandtabs()

                        comm = ""

                        for line in comment.splitlines():

                                line = line.strip()

                                if line == "":

                                        continue

                                comm += "  " + line + "\n"

                        return comm + "\n"

                else:

                        return "<no comment>\n\n"

        def format_output(self, command, output):

                str = "  Output Follows:\n"

                str += self.topdivider

                if command is not None:

                        str += "| $ " + command + "\n"

                if output is None or output == "":

                        str += "| <no output>\n"

                else:

                        for line in output.split("\n"):

                                str += "| " + line.rstrip() + "\n"

                str += self.botdivider

                return str

        def format_debug(self, output):

                str = "  Debug Buffer Follows:\n"

                str += self.topdivider

                if output is None or output == "":

                        str += "| <no debug buffer>\n"

                else:

                        for line in output.split("\n"):

                                str += "| " + line.rstrip() + "\n"

                str += self.botdivider

                return str

class AssFailException(Pkg5CommonException):

        def __init__(self, comment = None, debug=None):

                Pkg5CommonException.__init__(self, comment)

                self.__comment = comment

                self.__debug = debug

        def __str__(self):

                str = ""

                if self.__comment is None:

                        str += Exception.__str__(self)

                else:

                        str += self.format_comment(self.__comment)

                if self.__debug is not None and self.__debug != "":

                        str += self.format_debug(self.__debug)

                return str

class DebugLogHandler(logging.Handler):

        """This class is a special log handler to redirect logger output to

        the test case class' debug() method.

        """

        def __init__(self, test_case):

                self.test_case = test_case

                logging.Handler.__init__(self)

        def emit(self, record):

                self.test_case.debug(record)

def setup_logging(test_case):

        # Ensure logger messages output by unit tests are redirected

        # to debug output so they are not shown by default.

        from pkg.client import global_settings

        log_handler = DebugLogHandler(test_case)

        global_settings.info_log_handler = log_handler

        global_settings.error_log_handler = log_handler

class Pkg5TestCase(unittest.TestCase):

        # Needed for compatability

        failureException = AssertionError

        bogus_url = "test.invalid"

        __debug_buf = ""

        def __init__(self, methodName='runTest'):

                super(Pkg5TestCase, self).__init__(methodName)

                self.__test_root = None

                self.__pid = os.getpid()

                self.__pwd = os.getcwd()

                self.__didteardown = False

                setup_logging(self)

        def __str__(self):

                return "%s.py %s.%s" % (self.__class__.__module__,

                    self.__class__.__name__, self._testMethodName)

        #

        # Uses property() to implements test_root as a read-only attribute.

        #

        test_root = property(fget=lambda self: self.__test_root)

        def __get_ro_data_root(self):

                if not self.__test_root:

                        return None

                return os.path.join(self.__test_root, "ro_data")

        ro_data_root = property(fget=__get_ro_data_root)

        def cmdline_run(self, cmdline, comment="", coverage=True, exit=0,

            handle=False, out=False, prefix="", raise_error=True, su_wrap=None,

            stderr=False):

                wrapper = ""

                if coverage:

                        wrapper = self.coverage_cmd

                su_wrap, su_end = self.get_su_wrapper(su_wrap=su_wrap)

                cmdline = "%s%s%s %s%s" % (prefix, su_wrap, wrapper,

                    cmdline, su_end)

                self.debugcmd(cmdline)

                newenv = os.environ.copy()

                if coverage:

                        newenv.update(self.coverage_env)

                p = subprocess.Popen(cmdline,

                    env=newenv,

                    shell=True,

                    stdout=subprocess.PIPE,

                    stderr=subprocess.PIPE)

                if handle:

                        # Do nothing more.

                        return p

                self.output, self.errout = p.communicate()

                retcode = p.returncode

                self.debugresult(retcode, exit, self.output)

                if self.errout != "":

                        self.debug(self.errout)

                if raise_error and retcode == 99:

                        raise TracebackException(cmdline, self.output +

                            self.errout, comment)

                if not isinstance(exit, list):

                        exit = [exit]

                if raise_error and retcode not in exit:

                        raise UnexpectedExitCodeException(cmdline,

                            exit, retcode, self.output + self.errout,

                            comment)

                if out:

                        if stderr:

                                return retcode, self.output, self.errout

                        return retcode, self.output

                return retcode

        def debug(self, s):

                s = str(s)

                for x in s.splitlines():

                        if g_debug_output:

                                print >> sys.stderr, "# %s" % x

                        self.__debug_buf += x + "\n"

        def debugcmd(self, cmdline):

                wrapper = textwrap.TextWrapper(initial_indent="$ ",

                    subsequent_indent="\t",

                    break_long_words=False,

                    break_on_hyphens=False)

                res = wrapper.wrap(cmdline.strip())

                self.debug(" \\\n".join(res))

        def debugfilecreate(self, content, path):

                lines = content.splitlines()

                if lines == []:

                        lines = [""]

                if len(lines) > 1:

                        ins = " [+%d lines...]" % (len(lines) - 1)

                else:

                        ins = ""

                self.debugcmd(

                    "echo '%s%s' > %s" % (lines[0], ins, path))

        def debugresult(self, retcode, expected, output):

                if output.strip() != "":

                        self.debug(output.strip())

                if not isinstance(expected, list):

                        expected = [expected]

                if retcode is None or retcode != 0 or \

                    retcode not in expected:

                        self.debug("[exited %s, expected %s]" %

                            (retcode, ", ".join(str(e) for e in expected)))

        def get_debugbuf(self):

                return self.__debug_buf

        def get_su_wrapper(self, su_wrap=None):

                if su_wrap:

                        if su_wrap == True:

                                su_wrap = get_su_wrap_user()

                        cov_env = " ".join(

                            ("%s=%s" % e for e in self.coverage_env.items()))

                        su_wrap = "su %s -c 'LD_LIBRARY_PATH=%s %s " % \

                            (su_wrap, os.getenv("LD_LIBRARY_PATH", ""), cov_env)

                        su_end = "'"

                else:

                        su_wrap = ""

                        su_end = ""

                return su_wrap, su_end

        def getTeardownFunc(self):

                return (self, self.tearDown)

        def getSetupFunc(self):

                return (self, self.setUp)

        def setUp(self):

                self.__test_root = os.path.join(g_tempdir,

                    "ips.test.%d" % self.__pid)

                self.__didtearDown = False

                try:

                        os.makedirs(self.__test_root, 0755)

                except OSError, e:

                        if e.errno != errno.EEXIST:

                                raise e

                test_relative = os.path.sep.join(["..", "..", "src", "tests"])

                test_src = os.path.join(g_proto_area, test_relative)

                shutil.copytree(os.path.join(test_src, "ro_data"),

                    self.ro_data_root)

                #

                # TMPDIR affects the behavior of mkdtemp and mkstemp.

                # Setting this here should ensure that tests will make temp

                # files and dirs inside the test directory rather than

                # polluting /tmp.

                #

                os.environ["TMPDIR"] = self.__test_root

                tempfile.tempdir = self.__test_root

                setup_logging(self)

                self.configure_rcfile( "%s/usr/share/lib/pkg/pkglintrc" %

                    g_proto_area,

                    {"info_classification_path":

                    "%s/usr/share/package-manager/data/opensolaris.org.sections" %

                    g_proto_area}, self.test_root, section="pkglint")

        def impl_tearDown(self):

                # impl_tearDown exists so that we can ensure that this class's

                # teardown is actually called.  Sometimes, subclasses will

                # implement teardown but forget to call the superclass teardown.

                if self.__didteardown:

                        return

                self.__didteardown = True

                try:

                        os.chdir(self.__pwd)

                except OSError:

                        # working directory of last resort.

                        os.chdir(g_tempdir)

                #

                # Kill depots before blowing away test dir-- otherwise

                # the depot can race with the shutil.rmtree()

                #

                if hasattr(self, "killalldepots"):

                        try:

                                self.killalldepots()

                        except Exception, e:

                                print >> sys.stderr, str(e)

                #

                # We have some sloppy subclasses which don't call the superclass

                # setUp-- in which case the dir might not exist.  Tolerate it.

                #

                if self.__test_root is not None and \

                    os.path.exists(self.__test_root):

                        shutil.rmtree(self.__test_root)

        def tearDown(self):

                # In reality this call does nothing.

                unittest.TestCase.tearDown(self)

                self.impl_tearDown()

        def run(self, result=None):

                if result is None:

                        result = self.defaultTestResult()

                pwd = os.getcwd()

                result.startTest(self)

                testMethod = getattr(self, self._testMethodName)

                if getattr(result, "coverage", None) is not None:

                        self.coverage_cmd, self.coverage_env = result.coverage

                else:

                        self.coverage_cmd, self.coverage_env = "", {}

                try:

                        needtodie = False

                        try:

                                self.setUp()

                        except KeyboardInterrupt:

                                # Try hard to make sure we've done a teardown.

                                try:

                                        self.tearDown()

                                except:

                                        pass

                                self.impl_tearDown()

                                raise TestStopException

                        except:

                                # teardown could fail too, esp. if setup failed...

                                try:

                                        self.tearDown()

                                except:

                                        pass

                                # Try hard to make sure we've done a teardown.

                                self.impl_tearDown()

                                result.addError(self, sys.exc_info())

                                return

                        ok = False

                        error_added = False

                        try:

                                testMethod()

                                ok = True

                        except self.failureException:

                                result.addFailure(self, sys.exc_info())

                        except KeyboardInterrupt:

                                # Try hard to make sure we've done a teardown.

                                needtodie = True

                        except:

                                error_added = True

                                result.addError(self, sys.exc_info())

                        try:

                                self.tearDown()

                        except KeyboardInterrupt:

                                needtodie = True

                        except:

                                # Try hard to make sure we've done a teardown.

                                self.impl_tearDown()

                                # Make sure we don't mark this error'd twice.

                                if not error_added:

                                        result.addError(self, sys.exc_info())

                                ok = False

                        if needtodie:

                                try:

                                        self.impl_tearDown()

                                except:

                                        pass

                                raise TestStopException

                        if ok:

                                result.addSuccess(self)

                finally:

                        result.stopTest(self)

                        # make sure we restore our directory if it still exists.

                        try:

                                os.chdir(pwd)

                        except OSError, e:

                                # If directory doesn't exist anymore it doesn't

                                # matter.

                                if e.errno != errno.ENOENT:

                                        raise

        #

        # The following are utility functions for use by testcases.

        #

        def c_compile(self, prog_text, opts, outputfile):

                """Given a C program (as a string), compile it into the

                executable given by outputfile.  Outputfile should be

                given as a relative path, and will be located below the

                test prefix path.  Additional compiler options should be

                passed in 'opts'.  Suitable for compiling small test

                programs."""

                #

                # We use a series of likely compilers.  At present we support

                # this testing with SunStudio.

                #

                if os.path.dirname(outputfile) != "":