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

import testutils

if __name__ == "__main__":

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

import pkg5unittest

import os

import pkg.fmri as fmri

import pkg.portable as portable

import pkg.misc as misc

import pkg.p5p

import shutil

import stat

import tempfile

import unittest

class TestPkgTempSources(pkg5unittest.ManyDepotTestCase):

        # Don't discard repository or setUp() every test.

        persistent_setup = True

        # Tests in this suite use the read only data directory.

        need_ro_data = True

        foo_pkg = """

            open pkg://test/[email protected]

            add set name=pkg.summary value="Example package foo."

            add set name=variant.debug.foo value=true value=false

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

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

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

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

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

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

            add dir mode=0755 owner=root group=bin path=usr/share/doc

            add dir mode=0755 owner=root group=bin path=usr/share/doc/foo

            add dir mode=0755 owner=root group=bin path=usr/share/man

            add dir mode=0755 owner=root group=bin path=usr/share/man/man1

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

            add file tmp/libfoo.so.1 mode=0755 owner=root group=bin path=lib/libfoo.so.1 variant.debug.foo=false

            add file tmp/libfoo_debug.so.1 mode=0755 owner=root group=bin path=lib/libfoo.so.1 variant.debug.foo=true

            add file tmp/foo.1 mode=0444 owner=root group=bin path=usr/share/man/man1/foo.1 facet.doc.man=true

            add file tmp/README mode=0444 owner=root group=bin path=/usr/share/doc/foo/README

            add link path=usr/local/bin/soft-foo target=usr/bin/foo

            add hardlink path=usr/local/bin/hard-foo target=/usr/bin/foo

            close """

        incorp_pkg = """

            open pkg://test/[email protected]

            add set name=pkg.summary value="Incorporation"

            add depend type=incorporate [email protected],5.11-0.1

            close

            open pkg://test/[email protected]

            add set name=pkg.summary value="Incorporation"

            add depend type=incorporate [email protected],5.11-0.2

            close """

        licensed_pkg = """

            open pkg://test2/[email protected]

            add license tmp/LICENSE license=sample_license

            close """

        signed_pkg = """

            open pkg://test/[email protected]

            add depend type=require [email protected]

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

            add file tmp/quux mode=0755 owner=root group=bin path=usr/bin/quark

            add set name=authorized.species value=bobcat

            close """

        quux_pkg = """

            open pkg://test2/[email protected],5.11-0.1

            add set name=pkg.summary value="Example package quux."

            add depend type=require fmri=pkg:/incorp

            close

            open pkg://test2/[email protected],5.11-0.2

            add set name=pkg.summary value="Example package quux."

            add depend type=require fmri=pkg:/incorp

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

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

            add file tmp/quux mode=0755 owner=root group=bin path=usr/bin/quux

            close """

        misc_files = ["tmp/foo", "tmp/libfoo.so.1", "tmp/libfoo_debug.so.1",

            "tmp/foo.1", "tmp/README", "tmp/LICENSE", "tmp/quux"]

        def __seed_ta_dir(self, certs, dest_dir=None):

                if isinstance(certs, basestring):

                        certs = [certs]

                if not dest_dir:

                        dest_dir = self.ta_dir

                self.assert_(dest_dir)

                self.assert_(self.raw_trust_anchor_dir)

                for c in certs:

                        name = "%s_cert.pem" % c

                        portable.copyfile(

                            os.path.join(self.raw_trust_anchor_dir, name),

                            os.path.join(dest_dir, name))

        def image_create(self, *args, **kwargs):

                pkg5unittest.ManyDepotTestCase.image_create(self,

                    *args, **kwargs)

                self.ta_dir = os.path.join(self.img_path(), "etc/certs/CA")

                os.makedirs(self.ta_dir)

        def __publish_packages(self, rurl):

                """Private helper function to publish packages needed for

                testing.

                """

                pkgs = "".join([self.foo_pkg, self.incorp_pkg, self.signed_pkg,

                    self.quux_pkg])

                # Publish packages needed for tests.

                plist = self.pkgsend_bulk(rurl, pkgs)

                # Sign the 'signed' package.

                r = self.get_repo(self.dcs[1].get_repodir())

                sign_args = "-k %(key)s -c %(cert)s -i %(i1)s -i %(i2)s " \

                    "-i %(i3)s -i %(i4)s -i %(i5)s -i %(i6)s %(pkg)s" % {

                      "key": os.path.join(self.keys_dir, "cs1_ch5_ta1_key.pem"),

                      "cert": os.path.join(self.cs_dir, "cs1_ch5_ta1_cert.pem"),

                      "i1": os.path.join(self.chain_certs_dir,

                          "ch1_ta1_cert.pem"),

                      "i2": os.path.join(self.chain_certs_dir,

                          "ch2_ta1_cert.pem"),

                      "i3": os.path.join(self.chain_certs_dir,

                          "ch3_ta1_cert.pem"),

                      "i4": os.path.join(self.chain_certs_dir,

                          "ch4_ta1_cert.pem"),

                      "i5": os.path.join(self.chain_certs_dir,

                          "ch5_ta1_cert.pem"),

                      "i6": os.path.join(self.chain_certs_dir,

                          "ch1_ta3_cert.pem"),

                      "pkg": plist[3]

                    }

                self.pkgsign(rurl, sign_args)

                # This is just a test assertion to verify that the

                # package was signed as expected.

                self.image_create(rurl, prefix=None)

                self.__seed_ta_dir("ta1")

                self.pkg("set-property signature-policy verify")

                self.pkg("install signed")

                self.image_destroy()

                return [

                    fmri.PkgFmri(sfmri)

                    for sfmri in plist

                ]

        def __archive_packages(self, arc_name, repo, plist):

                """Private helper function to archive packages needed for

                testing.

                """

                arc_path = os.path.join(self.test_root, arc_name)

                assert not os.path.exists(arc_path)

                arc = pkg.p5p.Archive(arc_path, mode="w")

                for pfmri in plist:

                        arc.add_repo_package(pfmri, repo)

                arc.close()

                return arc_path

        def setUp(self):

                pkg5unittest.ManyDepotTestCase.setUp(self, ["test", "test",

                    "empty", "test2"])

                self.make_misc_files(self.misc_files)

                # First repository will contain all packages.

                self.all_rurl = self.dcs[1].get_repo_url()

                # Second repository will contain only foo.

                self.foo_rurl = self.dcs[2].get_repo_url()

                # Third will be empty.

                self.empty_rurl = self.dcs[3].get_repo_url()

                # Fourth will be for license packages only.

                self.licensed_rurl = self.dcs[4].get_repo_url()

                # Setup base test paths.

                self.path_to_certs = os.path.join(self.ro_data_root,

                    "signing_certs", "produced")

                self.keys_dir = os.path.join(self.path_to_certs, "keys")

                self.cs_dir = os.path.join(self.path_to_certs,

                    "code_signing_certs")

                self.chain_certs_dir = os.path.join(self.path_to_certs,

                    "chain_certs")

                self.raw_trust_anchor_dir = os.path.join(self.path_to_certs,

                    "trust_anchors")

                self.crl_dir = os.path.join(self.path_to_certs, "crl")

                # Publish packages.

                plist = self.__publish_packages(self.all_rurl)

                # Copy foo to second repository.

                self.pkgrecv(self.all_rurl, "-d %s foo" % self.foo_rurl)

                # Now create a package archive containing all packages, and

                # then one for each.

                repo = self.dcs[1].get_repo()

                self.all_arc = self.__archive_packages("all_pkgs.p5p", repo,

                    plist)

                for alist in ([plist[0]], [plist[1], plist[2]], [plist[3]],

                    [plist[4], plist[5]]):

                        arc_path = self.__archive_packages(

                            "%s.p5p" % alist[0].pkg_name, repo, alist)

                        setattr(self, "%s_arc" % alist[0].pkg_name, arc_path)

                self.ta_dir = None

                # Copy an archive and set its permissions to 0000 for testing

                # unprivileged user access attempts.

                self.perm_arc = os.path.join(self.test_root, "noaccess.p5p")

                portable.copyfile(self.foo_arc, self.perm_arc)

                os.chmod(self.perm_arc, 0)

                # Create an empty archive.

                arc_path = os.path.join(self.test_root, "empty.p5p")

                arc = pkg.p5p.Archive(arc_path, mode="w")

                arc.close()

                self.empty_arc = arc_path

                # Store FMRIs for later use.

                self.foo10 = plist[0]

                self.incorp10 = plist[1]

                self.incorp20 = plist[2]

                self.signed10 = plist[3]

                self.quux01 = plist[4]

                self.quux10 = plist[5]

                # Handle license package specially.

                self.licensed10 = self.pkgsend_bulk(self.licensed_rurl,

                    self.licensed_pkg)[0]

        def test_00_list(self):

                """Verify that the list operation works as expected for

                temporary origins.

                """

                # Create an image and verify no packages are known.

                self.image_create(self.empty_rurl, prefix=None)

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

                # Verify graceful failure for an empty source alone or in

                # combination with another temporary source.

                self.pkg("list -H -g %s" % self.empty_arc, exit=1)

                self.pkg("list -H -g %s -g %s" % (self.empty_arc,

                    self.foo_arc), exit=1)

                # Verify graceful failure if source doesn't exist.

                self.pkg("list -H -g %s" % (self.foo_arc + ".nosuchpkg"),

                    exit=1)

                # Verify graceful failure if user doesn't have permission to

                # access temporary source.

                self.pkg("list -H -g %s" % self.perm_arc, su_wrap=True, exit=1)

                # Verify graceful list failure if -u is used with -g.

                self.pkg("list -H -u -g %s" % self.foo_arc, exit=2)

                # Verify list output for a single package temporary source.

                # -a is used here to verify that even though -a is implicit,

                # it is not an error to specify it.

                self.pkg("list -aH -g %s" % self.foo_arc)

                expected = "foo (test) 1.0 ---\n"

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                # Note that -a is implicit when -g is used, so all of

                # the following tests purposefully omit it.

                # Verify list output for a multiple package temporary source

                # as an unprivileged user.

                self.pkg("list -fH -g %s" % self.all_arc, su_wrap=True)

                expected = \

                    ("foo (test) 1.0 ---\n"

                    "incorp (test) 2.0 ---\n"

                    "incorp (test) 1.0 ---\n"

                    "quux (test2) 1.0-0.2 ---\n"

                    "quux (test2) 0.1-0.1 ---\n"

                    "signed (test) 1.0 ---\n")

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                # Verify list output for a multiple package temporary source.

                self.pkg("list -fH -g %s" % self.all_arc)

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                # Verify list output for multiple temporary sources using

                # different combinations of archives and repositories.

                self.pkg("list -fH -g %s -g %s" % (self.signed_arc,

                    self.foo_rurl))

                expected = \

                    ("foo (test) 1.0 ---\n"

                    "signed (test) 1.0 ---\n")

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                self.pkg("list -fH -g %s -g %s -g %s -g %s" % (self.signed_arc,

                    self.incorp_arc, self.quux_arc, self.foo_arc))

                expected = \

                    ("foo (test) 1.0 ---\n"

                    "incorp (test) 2.0 ---\n"

                    "incorp (test) 1.0 ---\n"

                    "quux (test2) 1.0-0.2 ---\n"

                    "quux (test2) 0.1-0.1 ---\n"

                    "signed (test) 1.0 ---\n")

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                self.pkg("list -fH -g %s -g %s -g %s -g %s" % (self.signed_arc,

                    self.quux_arc, self.incorp_arc, self.foo_rurl))

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                self.pkg("list -fH -g %s -g %s" % (self.all_arc,

                    self.all_rurl))

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                # Verify list -g without -f.

                self.pkg("list -H -g %s -g %s -g %s -g %s" % (self.signed_arc,

                    self.quux_arc, self.incorp_arc, self.foo_rurl))

                expected = \

                    ("foo (test) 1.0 ---\n"

                    "incorp (test) 2.0 ---\n"

                    "quux (test2) 1.0-0.2 ---\n"

                    "signed (test) 1.0 ---\n")

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                # Verify package installed from archive shows in default list

                # output.

                self.pkg("install -g %s [email protected]" % self.incorp_arc)

                self.pkg("list -H")

                expected = "incorp (test) 1.0 i--\n"

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                # Verify list -g with an incorp installed without -f.

                self.pkg("list -H -g %s -g %s -g %s -g %s" % (self.signed_arc,

                    self.quux_arc, self.incorp_arc, self.foo_rurl))

                expected = \

                    ("foo (test) 1.0 ---\n"

                    "incorp (test) 1.0 i--\n"

                    "quux (test2) 0.1-0.1 ---\n"

                    "signed (test) 1.0 ---\n")

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                # Verify output again as unprivileged user.

                self.pkg("list -H -g %s -g %s -g %s -g %s" % (self.signed_arc,

                    self.quux_arc, self.incorp_arc, self.foo_rurl),

                    su_wrap=True)

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                # Verify list -g with an incorp installed with -f.

                self.pkg("list -fH -g %s -g %s -g %s -g %s" % (self.signed_arc,

                    self.quux_arc, self.incorp_arc, self.foo_rurl))

                expected = \

                    ("foo (test) 1.0 ---\n"

                    "incorp (test) 2.0 ---\n"

                    "incorp (test) 1.0 i--\n"

                    "quux (test2) 1.0-0.2 ---\n"

                    "quux (test2) 0.1-0.1 ---\n"

                    "signed (test) 1.0 ---\n")

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                # Verify list -g with an incorp installed and -n.

                self.pkg("list -nH -g %s -g %s -g %s -g %s" % (self.signed_arc,

                    self.quux_arc, self.incorp_arc, self.foo_rurl))

                expected = \

                    ("foo (test) 1.0 ---\n"

                    "incorp (test) 2.0 ---\n"

                    "quux (test2) 1.0-0.2 ---\n"

                    "signed (test) 1.0 ---\n")

                output = self.reduceSpaces(self.output)

                self.assertEqualDiff(expected, output)

                # Uninstall all packages and verify there are no known packages.

                self.pkg("uninstall \*")

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

                # Cleanup.

                self.image_destroy()

        def test_01_info(self):

                """Verify that the info operation works as expected for

                temporary origins.

                """

                # Create an image and verify no packages are known.

                self.image_create(self.empty_rurl, prefix=None)

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

                # Verify graceful failure for an empty source alone or in

                # combination with another temporary source.

                self.pkg("info -g %s \*" % self.empty_arc, exit=1)

                self.pkg("info -g %s -g %s foo" % (self.empty_arc,

                    self.foo_arc), exit=1)

                # Verify graceful failure if source doesn't exist.

                self.pkg("info -g %s foo" % (self.foo_arc + ".nosuchpkg"),

                    exit=1)

                # Verify graceful failure if user doesn't have permission to

                # access temporary source.

                self.pkg("info -g %s foo" % self.perm_arc, su_wrap=True, exit=1)

                # Verify graceful failure if -l is used with -g.

                self.pkg("info -l -g %s foo" % self.foo_arc, exit=2)

                # Verify output for a single package temporary source.

                # -r is used here to verify that even though -r is implicit,

                # it is not an error to specify it.

                def pd(pfmri):

                        return pfmri.version.get_timestamp().strftime("%c")

                self.pkg("info -r -g %s foo" % self.foo_arc, su_wrap=True)

                expected = """\

          Name: foo

       Summary: Example package foo.

         State: Not installed

     Publisher: test

       Version: 1.0

 Build Release: 5.11

        Branch: None

Packaging Date: %(pkg_date)s

          Size: 41.00 B

          FMRI: %(pkg_fmri)s

""" % { "pkg_date": pd(self.foo10), "pkg_fmri": self.foo10 }

                self.assertEqualDiff(expected, self.output)

                # Again, as prvileged user.

                self.pkg("info -r -g %s foo" % self.foo_arc)

                self.assertEqualDiff(expected, self.output)

                # Note that -r is implicit when -g is used, so all of

                # the following tests purposefully omit it.

                # Verify info output for a multiple package temporary source

                # as an unprivileged user.

                self.pkg("info -g %s \*" % self.all_arc, su_wrap=True)

                expected = """\

          Name: foo

       Summary: Example package foo.

         State: Not installed

     Publisher: test

       Version: 1.0

 Build Release: 5.11

        Branch: None

Packaging Date: %(foo10_pkg_date)s

          Size: 41.00 B

          FMRI: %(foo10_pkg_fmri)s

          Name: incorp

       Summary: Incorporation

         State: Not installed

     Publisher: test

       Version: 2.0

 Build Release: 5.11

        Branch: None

Packaging Date: %(incorp20_pkg_date)s

          Size: 0.00 B

          FMRI: %(incorp20_pkg_fmri)s

          Name: quux

       Summary: Example package quux.

         State: Not installed

     Publisher: test2

       Version: 1.0

 Build Release: 5.11

        Branch: 0.2

Packaging Date: %(quux10_pkg_date)s

          Size: 8.00 B

          FMRI: %(quux10_pkg_fmri)s

          Name: signed

       Summary: 

         State: Not installed

     Publisher: test

       Version: 1.0

 Build Release: 5.11

        Branch: None

Packaging Date: %(signed10_pkg_date)s

          FMRI: %(signed10_pkg_fmri)s

""" % { "foo10_pkg_date": pd(self.foo10), "foo10_pkg_fmri": self.foo10,

    "incorp20_pkg_date": pd(self.incorp20), "incorp20_pkg_fmri": self.incorp20,