# 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, 2015, Oracle and/or its affiliates. All rights reserved.

#

# Define the basic classes that all test cases are inherited from.

# The currently defined test case classes are:

#

# ApacheDepotTestCase

# CliTestCase

# ManyDepotTestCase

# Pkg5TestCase

# SingleDepotTestCase

# SingleDepotTestCaseCorruptImage

#

import baseline

import ConfigParser

import copy

import difflib

import errno

import gettext

import hashlib

import httplib

import logging

import multiprocessing

import os

import pprint

import shutil

import signal

import simplejson as json

import stat

import subprocess

import sys

import tempfile

import time

import traceback

import unittest

import urllib2

import urlparse

import operator

import platform

import pty

import pwd

import re

import ssl

import StringIO

import textwrap

import threading

import traceback

import types

import pkg.client.api_errors as apx

import pkg.misc as misc

import pkg.client.publisher as publisher

import pkg.portable as portable

import pkg.server.repository as sr

import M2Crypto as m2

from pkg.client.debugvalues import DebugValues

from socket import error as socketerror

EmptyI = tuple()

EmptyDict = dict()

#

# These are initialized by pkg5testenv.setup_environment.

#

g_proto_area = "TOXIC"

g_proto_readable = False

# Location of root of test suite.

g_test_dir = "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

from pkg.client.debugvalues import DebugValues

# Version test suite is known to work with.

PKG_CLIENT_NAME = "pkg"

CLIENT_API_VERSION = 80

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 compatibility

        failureException = AssertionError

        #

        # Some dns servers return results for unknown dns names to redirect

        # callers to a common landing page.  To avoid getting tripped up by

        # these stupid servers make sure that bogus_url actually contains an

        # syntactically invalid dns name so we'll never succeed at the lookup.

        #

        bogus_url = "test.0.invalid"

        __debug_buf = ""

        smf_cmds = { \

            "usr/bin/svcprop" : """\

#!/usr/bin/python

import sys

if __name__ == "__main__":

        sys.exit(1)

"""}

        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

                self.__base_port = None

                self.coverage_cmd = ""

                self.coverage_env = {}

                self.next_free_port = None

                self.ident = None

                self.pkg_cmdpath = "TOXIC"

                self.debug_output = g_debug_output

                setup_logging(self)

                global g_proto_readable

                if not g_proto_readable:

                        self.assertProtoReadable()

                        g_proto_readable = True

        @property

        def methodName(self):

                return self._testMethodName

        @property

        def suite_name(self):

                return self.__suite_name

        def __str__(self):

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

                    self.__class__.__name__, self._testMethodName)

        def __set_base_port(self, port):

                if self.__base_port is not None or \

                    self.next_free_port is not None:

                        raise RuntimeError("Setting the base port twice isn't "

                            "allowed")

                self.__base_port = port

                self.next_free_port = port

        base_port = property(lambda self: self.__base_port, __set_base_port)

        def assertProtoReadable(self):

                """Ensure proto area is readable by unprivileged user."""

                try:

                        self.cmdline_run("dir {0}".format(g_proto_area),

                            su_wrap=True)

                except:

                        raise TestStopException("proto area '{0} is not "

                            "readable by unprivileged user {1}".format(

                                g_proto_area, get_su_wrap_user()))

        def assertRegexp(self, text, regexp):

                """Test that a regexp search matches text."""

                if re.search(regexp, text):

                        return

                raise self.failureException, \

                    "\"%s\" does not match \"%s\"" % (regexp, text)

        def assertRaisesRegexp(self, excClass, regexp,

            callableObj, *args, **kwargs):

                """Perform the same logic as assertRaises, but then verify

                that the stringified version of the exception contains the

                regexp pattern.

                Introduced in in python 2.7"""

                try:

                        callableObj(*args, **kwargs)

                except excClass, e:

                        if re.search(regexp, str(e)):

                                return

                        raise self.failureException, \

                            "\"%s\" does not match \"%s\"" % (regexp, str(e))

                raise self.failureException, \

                    "%s not raised" % excClass

        def assertRaisesStringify(self, excClass, callableObj, *args, **kwargs):

                """Perform the same logic as assertRaises, but then verify that

                the exception raised can be stringified."""

                try:

                        callableObj(*args, **kwargs)

                except excClass, e:

                        str(e)

                        return

                else:

                        raise self.failureException, "%s not raised" % excClass

        #

        # 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 persistent_setup_copy(self, orig):

                pass

        @staticmethod

        def ptyPopen(args, executable=None, env=None, shell=False):

                """Less featureful but inspired by subprocess.Popen.

                Runs subprocess in a pty"""

                #

                # Note: In theory the right answer here is to subclass Popen,

                # but we found that in practice we'd have to reimplement most

                # of that class, because its handling of file descriptors is

                # too brittle in its _execute_child() code.

                #

                def __drain(masterf, outlist):

                        # Use a list as a way to pass by reference

                        while True:

                                        break

                # This is the arg handling protocol from Popen

                if isinstance(args, types.StringTypes):

                        args = [args]

                else:

                        args = list(args)

                if shell:

                        args = ["/bin/sh", "-c"] + args

                        if executable:

                                args[0] = executable

                if executable is None:

                        executable = args[0]

                pid,fd = pty.fork()

                if pid == 0:

                        try:

                                # Child

                                if env is None:

                                        os.execvp(executable, args)

                                else:

                                        os.execvpe(executable, args, env)

                        except:

                                traceback.print_exc()

                                os._exit(99)

                else:

                        masterf = os.fdopen(fd)

                        outlist = []

                        t = threading.Thread(target=__drain,

                            args=(masterf, outlist))

                        t.start()

                        waitedpid, retcode = os.waitpid(pid, 0)

                        retcode = retcode >> 8

                        t.join()

                return retcode, "".join(outlist)

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

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

            stderr=False, env_arg=None, usepty=False):

                # If caller provides arguments as a string, the shell must

                # process them.

                shell = not isinstance(cmdline, list)

                wrapper = ""

                if coverage:

                        wrapper = self.coverage_cmd

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

                    shell=shell)

                if isinstance(cmdline, list):

                        if wrapper:

                                # Coverage command must be split into arguments.

                                wrapper = wrapper.split()

                                while wrapper:

                                        cmdline.insert(0, wrapper.pop())

                        if su_wrap:

                                # This ensures that all parts of the command

                                # line to be passed to 'su -c' are passed as a

                                # single argument.

                                while su_wrap[-1] != "-c":

                                        cmdline.insert(0, su_wrap.pop())

                                cmdline = [" ".join(cmdline)]

                                while su_wrap:

                                        cmdline.insert(0, su_wrap.pop())

                        if prefix:

                                cmdline.insert(0, prefix)

                else:

                        # Space needed between su_wrap and wrapper.

                        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)

                if env_arg:

                        newenv.update(env_arg)

                if not usepty:

                        p = subprocess.Popen(cmdline,

                            env=newenv,

                            shell=shell,

                            stdout=subprocess.PIPE,

                            stderr=subprocess.PIPE)

                        if handle:

                                # Do nothing more.

                                return p

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

                        retcode = p.returncode

                else:

                        retcode, self.output = self.ptyPopen(cmdline,

                            env=newenv, shell=True)

                        self.errout = ""

                self.debugresult(retcode, exit, self.output + 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