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

if __name__ == "__main__":

        testutils.setup_environment("../../../proto")

import pkg5unittest

import httplib

import os

import shutil

import tempfile

import time

import unittest

import urllib

import urllib2

import urlparse

import pkg.client.publisher as publisher

import pkg.depotcontroller as dc

import pkg.fmri as fmri

import pkg.misc as misc

import pkg.server.repository as sr

import pkg.p5i as p5i

class TestPkgDepot(pkg5unittest.SingleDepotTestCase):

        # Only start/stop the depot once (instead of for every test)

        persistent_setup = True

        foo10 = """

            open [email protected],5.11-0

            add dir path=tmp/foo mode=0755 owner=root group=bin

            close """

        bar10 = """

            open [email protected],5.11-0

            add dir path=tmp/bar mode=0755 owner=root group=bin

            close """

        quux10 = """

            open [email protected],5.11-0

            add dir mode=0755 owner=root group=bin path=/bin

            add file tmp/cat mode=0555 owner=root group=bin path=/bin/cat

            add file tmp/libc.so.1 mode=0555 owner=root group=bin path=/lib/libc.so.1

            close """

        info10 = """

            open [email protected],5.11-0

            close """

        update10 = """

            open [email protected],5.11-0

            close """

        update11 = """

            open [email protected],5.11-0

            close """

        system10 = """

            open system/[email protected],5.11-0

            add set name="description" value="Package to test package names with slashes"

            add dir path=tmp/foo mode=0755 owner=root group=bin

            add depend type=require fmri=pkg:/SUNWcsl

            close """

        misc_files = [ "tmp/libc.so.1", "tmp/cat" ]

        def setUp(self):

                # This suite, for obvious reasons, actually needs a depot.

                pkg5unittest.SingleDepotTestCase.setUp(self, start_depot=True)

                self.make_misc_files(self.misc_files)

        def test_depot_ping(self):

                """ Ping the depot several times """

                self.assert_(self.dc.is_alive())

                self.assert_(self.dc.is_alive())

                self.assert_(self.dc.is_alive())

                self.assert_(self.dc.is_alive())

        def testStartStop(self):

                """ Start and stop the depot several times """

                self.dc.stop()

                for i in range(0, 5):

                        self.dc.start()

                        self.assert_(self.dc.is_alive())

                        self.dc.stop()

                        self.assert_(not self.dc.is_alive())

                self.dc.start()

        def test_bug_1876(self):

                """ Send package [email protected] an action at a time, restarting the

                    depot server after each one is sent, to ensure that

                    transactions work across depot restart. Then verify that

                    the package was successfully added by performing some

                    basic operations. """

                durl = self.dc.get_depot_url()

                for line in self.quux10.split("\n"):

                        line = line.strip()

                        if line == "":

                                continue

                        try:

                                self.pkgsend(durl, line, exit = 0)

                        except:

                                self.pkgsend(durl, "close -A", exit = 0)

                                raise

                        if not line == "close":

                                self.restart_depots()

                self.image_create(durl)

                self.pkg("list -a")

                self.pkg("list", exit=1)

                self.pkg("install quux")

                self.pkg("list")

                self.pkg("verify")

                self.pkg("uninstall quux")

                self.pkg("verify")

        def test_bad_fmris(self):

                durl = self.dc.get_depot_url()

                self.pkgsend(durl, "open foo@", exit=1)

                self.pkgsend(durl, "open [email protected]", exit=1)

                self.pkgsend(durl, "open [email protected],-2.0", exit=1)

        def test_bug_3365(self):

                durl = self.dc.get_depot_url()

                depotpath = self.dc.get_repodir()

                dir_file = os.path.join(depotpath, "search.dir")

                pag_file = os.path.join(depotpath, "search.pag")

                self.assert_(not os.path.exists(dir_file))

                self.assert_(not os.path.exists(pag_file))

                f = open(dir_file, "w")

                f.close()

                f = open(pag_file, "w")

                f.close()

                self.assert_(os.path.exists(dir_file))

                self.assert_(os.path.exists(pag_file))

                self.dc.stop()

                self.dc.start()

                self.pkgsend_bulk(durl, self.quux10)

                self.assert_(not os.path.exists(dir_file))

                self.assert_(not os.path.exists(pag_file))

        def test_bug_4489(self):

                """Publish a package and then verify that the depot /info

                operation doesn't fail."""

                depot_url = self.dc.get_depot_url()

                plist = self.pkgsend_bulk(depot_url, self.info10)

                repourl = urlparse.urljoin(depot_url, "info/0/%s" % plist[0])

                urllib2.urlopen(repourl)

        def test_bug_3739(self):

                """Verify that a depot will return a 400 (Bad Request) error

                whenever it is provided malformed FMRIs."""

                durl = self.dc.get_depot_url()

                for operation in ("info", "manifest"):

                        for entry in ("BRCMbnx", "BRCMbnx%40a",

                            "BRCMbnx%400.5.11%2C5.11-0.101%3A20081119T231649a"):

                                try:

                                        urllib2.urlopen("%s/%s/0/%s" % (durl,

                                            operation, entry))

                                except urllib2.HTTPError, e:

                                        if e.code != httplib.BAD_REQUEST:

                                                raise

        def test_bug_5366(self):

                """Publish a package with slashes in the name, and then verify

                that the depot manifest and info operations work regardless of

                the encoding."""

                depot_url = self.dc.get_depot_url()

                plist = self.pkgsend_bulk(depot_url, self.system10)

                # First, try it un-encoded.

                repourl = urlparse.urljoin(depot_url, "info/0/%s" % plist[0])

                urllib2.urlopen(repourl)

                repourl = urlparse.urljoin(depot_url, "manifest/0/%s" %

                    plist[0])

                urllib2.urlopen(repourl)

                # Second, try it encoded.

                repourl = urlparse.urljoin(depot_url, "info/0/%s" %

                    urllib.quote(plist[0]))

                urllib2.urlopen(repourl)

                repourl = urlparse.urljoin(depot_url, "manifest/0/%s" %

                    urllib.quote(plist[0]))

                urllib2.urlopen(repourl)

        def test_bug_5707(self):

                """Testing depotcontroller.refresh()."""

                depot_url = self.dc.get_depot_url()

                self.pkgsend_bulk(depot_url, self.foo10)

                self.image_create(depot_url)

                self.pkg("install foo")

                self.pkg("verify")

                depot_file_url = "file://%s" % self.dc.get_repodir()

                self.pkgsend_bulk(depot_url, self.bar10)

                self.pkg("refresh")

                self.pkg("install bar")

                self.pkg("verify")

                self.dc.refresh()

                self.pkg("refresh")

                self.pkg("install bar", exit=4) # nothing to do

                self.pkg("verify")

        def test_face_root(self):

                """Verify that files outside of the package content web root

                cannot be accessed, and that files inside can be."""

                depot_url = self.dc.get_depot_url()

                # Since /usr/share/lib/pkg/web/ is the content web root,

                # any attempts to go outside that directory should fail

                # with a 404 error.

                try:

                        urllib2.urlopen("%s/../../../../bin/pkg" % depot_url)

                except urllib2.HTTPError, e:

                        if e.code != httplib.NOT_FOUND:

                                raise

                f = urllib2.urlopen("%s/robots.txt" % depot_url)

                self.assert_(len(f.read()))

                f.close()

        def test_repo_create(self):

                """Verify that starting a depot server in readonly mode with

                a non-existent or empty repo_dir fails and that permissions

                errors are handled correctly during creation.  Then verify

                that starting a depot with the same directory in publishing

                mode works and then a readonly depot again after that works.

                """

                dpath = os.path.join(self.test_root, "repo_create")

                opath = self.dc.get_repodir()

                self.dc.set_repodir(dpath)

                # First, test readonly mode with a repo_dir that doesn't exist.

                self.dc.set_readonly()

                self.dc.stop()

                self.dc.start_expected_fail()

                self.assert_(not self.dc.is_alive())

                # Next, test readonly mode with a repo_dir that is empty.

                os.makedirs(dpath, misc.PKG_DIR_MODE)

                self.dc.set_readonly()

                self.dc.start_expected_fail()

                self.assert_(not self.dc.is_alive())

                # Next, test readwrite (publishing) mode with a non-existent

                # repo_dir.

                shutil.rmtree(dpath)

                self.dc.set_readwrite()

                self.dc.start()

                self.assert_(self.dc.is_alive())

                self.dc.stop()

                self.assert_(not self.dc.is_alive())

                # Next, test readwrite (publishing) mode with a non-existent

                # repo_dir for an unprivileged user.

                shutil.rmtree(dpath)

                self.dc.set_readwrite()

                wr_start, wr_end = self.dc.get_wrapper()

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

                try:

                        self.dc.set_wrapper([su_wrap], su_end)

                        self.dc.start_expected_fail(exit=1)

                finally:

                        # Even if this test fails, this wrapper must be reset.

                        self.dc.set_wrapper(wr_start, wr_end)

                self.assert_(not self.dc.is_alive())

                # Next, test readwrite (publishing) mode with an empty repo_dir.

                os.makedirs(dpath, misc.PKG_DIR_MODE)

                self.dc.set_readwrite()

                self.dc.start()

                self.assert_(self.dc.is_alive())

                self.dc.stop()

                self.assert_(not self.dc.is_alive())

                # Finally, re-test readonly mode now that the repository has

                # been created.

                self.dc.set_readonly()

                self.dc.start()

                self.assert_(self.dc.is_alive())

                self.dc.stop()

                self.assert_(not self.dc.is_alive())

                # Cleanup.

                shutil.rmtree(dpath)

                self.dc.set_repodir(opath)

        def test_repo_file_only(self):

                """Test that if a depot is created with only --file-root

                supplied, it comes up in mirror mode, with only file content

                available."""

                if self.dc.is_alive():

                        self.dc.stop()

                fpath = os.path.join(self.test_root, "var/pkg/download")

                os.makedirs(fpath, misc.PKG_DIR_MODE)

                opath = self.dc.get_repodir()

                self.dc.set_repodir(None)

                self.dc.set_file_root(fpath)

                self.dc.set_readwrite()

                self.dc.start()

                self.assert_(self.dc.is_alive())

                durl = self.dc.get_depot_url()

                verdata = urllib2.urlopen("%s/versions/0/" % durl)

                verlines = verdata.readlines()

                verdict = dict(

                    s.split(None, 1)

                    for s in (l.strip() for l in verlines)

                )

                self.assert_("file" in verdict)

                self.assert_("catalog" not in verdict)

                self.assert_("manifest" not in verdict)

                self.dc.stop()

                self.dc.set_repodir(opath)

                self.dc.set_file_root(None)

        def test_append_reopen(self):

                """Test that if a depot has a partially finished append

                transaction, that it reopens it correctly."""

                durl = self.dc.get_depot_url()

                plist = self.pkgsend_bulk(durl, self.foo10)

                self.pkgsend(durl, "append %s" % plist[0])

                self.dc.stop()

                self.dc.start()

                self.pkgsend(durl, "close")

        def test_nonsig_append(self):

                """Test that sending a non-signature action to an append

                transaction results in an error."""

                durl = self.dc.get_depot_url()

                plist = self.pkgsend_bulk(durl, self.foo10)

                self.pkgsend(durl, "append %s" % plist[0])

                self.pkgsend(durl, "add dir path=tmp/foo mode=0755 "

                    "owner=root group=bin", exit=1)

        def test_root_link(self):

                """Verify that the depot server accepts a link to a

                directory as a repository root."""

                if self.dc.started:

                        self.dc.stop()

                # Create a link to the repository and verify that

                # the depot server allows it.

                lsrc = self.dc.get_repodir()

                ltarget = os.path.join(self.test_root, "depot_link")

                os.symlink(lsrc, ltarget)

                self.dc.set_repodir(ltarget)

                self.dc.start()

                # Reset for any tests that might execute afterwards.

                os.unlink(ltarget)

                self.dc.stop()

                self.dc.set_repodir(lsrc)

                self.dc.start()

class TestDepotController(pkg5unittest.CliTestCase):

        def setUp(self):

                pkg5unittest.CliTestCase.setUp(self)

                self.__dc = dc.DepotController()

                self.__pid = os.getpid()

                self.__dc.set_property("publisher", "prefix", "test")

                self.__dc.set_depotd_path(pkg5unittest.g_proto_area + \

                    "/usr/lib/pkg.depotd")

                self.__dc.set_depotd_content_root(pkg5unittest.g_proto_area + \

                    "/usr/share/lib/pkg")

                repopath = os.path.join(self.test_root, "repo")

                logpath = os.path.join(self.test_root, self.id())

                self.create_repo(repopath, properties={ "publisher": {

                    "prefix": "test" }})

                self.__dc.set_repodir(repopath)

                self.__dc.set_logpath(logpath)

        def _get_repo_index_dir(self):

                depotpath = self.__dc.get_repodir()

                repo = self.__dc.get_repo()

                rstore = repo.get_pub_rstore("test")

                return rstore.index_root

        def _get_repo_writ_dir(self):

                depotpath = self.__dc.get_repodir()

                repo = self.__dc.get_repo()

                rstore = repo.get_pub_rstore("test")

                return rstore.writable_root

        def tearDown(self):

                pkg5unittest.CliTestCase.tearDown(self)

                self.__dc.kill()

        def testStartStop(self):

                for i in range(0, 5):

                        self.__dc.start()

                        self.assert_(self.__dc.is_alive())

                        self.__dc.stop()

                        self.assert_(not self.__dc.is_alive())

        def test_cfg_file(self):

                cfg_file = os.path.join(self.test_root, "cfg2")

                fh = open(cfg_file, "w")

                fh.close()

                self.__dc.set_cfg_file(cfg_file)

                self.__dc.start()

        def test_writable_root(self):

                """Tests whether the index and feed cache file are written to

                the writable root parameter."""

                self.make_misc_files(TestPkgDepot.misc_files)

                writable_root = os.path.join(self.test_root,

                    "writ_root")

                o_index_dir = os.path.join(self._get_repo_index_dir(), "index")

                timeout = 10

                def check_state(check_feed):

                        index_dir = os.path.join(self._get_repo_writ_dir(),

                            "index")

                        feed = os.path.join(writable_root, "publisher", "test",

                            "feed.xml")

                        found = not os.path.exists(o_index_dir) and \

                            os.path.isdir(index_dir) and \

                            (not check_feed or os.path.isfile(feed))

                        start_time = time.time()

                        while not found and time.time() - start_time < timeout:

                                time.sleep(1)

                                found = not os.path.exists(o_index_dir) and \

                                    os.path.isdir(index_dir) and \

                                    (not check_feed or os.path.isfile(feed))

                        self.assert_(not os.path.exists(o_index_dir))

                        self.assert_(os.path.isdir(index_dir))

                        if check_feed:

                                try:

                                        self.assert_(os.path.isfile(feed))

                                except:

                                        raise RuntimeError("Feed cache file "

                                            "not found at '%s'." % feed)

                def get_feed(durl, pub=""):