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

#

import testutils

if __name__ == "__main__":

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

import pkg5unittest

import json

import os

import pkg.catalog as catalog

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

        licensed_pkg_2 = """

            open pkg://test2/[email protected]

            add license tmp/LICENSE2 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/LICENSE2", "tmp/quux"]

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

                        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 = "{0}_cert.pem".format(c)

                        portable.copyfile(

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

                            os.path.join(dest_dir, name))

        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} -c {cert} -i {i1} -i {i2} " \

                    "-i {i3} -i {i4} -i {i5} -i {i6} {pkg}".format(

                      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 {0} foo".format(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(

                            "{0}.p5p".format(alist[0].pkg_name), repo, alist)

                        setattr(self, "{0}_arc".format(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]

                self.licensed20 = self.pkgsend_bulk(self.licensed_rurl,

                    self.licensed_pkg_2)[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 {0}".format(self.empty_arc), exit=1)

                self.pkg("list -H -g {0} -g {1}".format(self.empty_arc,

                    self.foo_arc), exit=1)

                # Verify graceful failure if source doesn't exist.

                self.pkg("list -H -g {0}".format(self.foo_arc + ".nosuchpkg"),

                    exit=1)

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

                # access temporary source.

                self.pkg("list -H -g {0}".format(self.perm_arc), su_wrap=True, exit=1)

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

                self.pkg("list -H -u -g {0}".format(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 {0}".format(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 {0}".format(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 {0}".format(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 {0} -g {1}".format(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 {0} -g {1} -g {2} -g {3}".format(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 {0} -g {1} -g {2} -g {3}".format(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 {0} -g {1}".format(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 {0} -g {1} -g {2} -g {3}".format(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 {0} [email protected]".format(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 {0} -g {1} -g {2} -g {3}".format(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 {0} -g {1} -g {2} -g {3}".format(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 {0} -g {1} -g {2} -g {3}".format(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 {0} -g {1} -g {2} -g {3}".format(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.

                """

		# because we compare date strings we must run this in

		# a consistent locale, which we made 'C'

		os.environ['LC_ALL'] = 'C'

                # 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 {0} \*".format(self.empty_arc), exit=1)

                self.pkg("info -g {0} -g {1} foo".format(self.empty_arc,

                    self.foo_arc), exit=1)

                # Verify graceful failure if source doesn't exist.

                self.pkg("info -g {0} foo".format(self.foo_arc + ".nosuchpkg"),

                    exit=1)

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

                # access temporary source.

                self.pkg("info -g {0} foo".format(self.perm_arc), su_wrap=True, exit=1)

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

                self.pkg("info -l -g {0} foo".format(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 {0} foo".format(self.foo_arc), su_wrap=True)

                expected = """\

          Name: foo

       Summary: Example package foo.

         State: Not installed

     Publisher: test

       Version: 1.0

        Branch: None

Packaging Date: {pkg_date}

          Size: 41.00 B

          FMRI: {pkg_fmri}

""".format(pkg_date=pd(self.foo10), pkg_fmri=self.foo10.get_fmri(

    include_build=False))

                self.assertEqualDiff(expected, self.output)

                # Again, as prvileged user.

                self.pkg("info -r -g {0} foo".format(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 {0} \*".format(self.all_arc), su_wrap=True)

                expected = """\

          Name: foo

       Summary: Example package foo.

         State: Not installed

     Publisher: test

       Version: 1.0

        Branch: None

Packaging Date: {foo10_pkg_date}

          Size: 41.00 B

          FMRI: {foo10_pkg_fmri}

          Name: incorp

       Summary: Incorporation

         State: Not installed

     Publisher: test

       Version: 2.0

        Branch: None

Packaging Date: {incorp20_pkg_date}

          Size: 0.00 B

          FMRI: {incorp20_pkg_fmri}

          Name: quux

       Summary: Example package quux.

         State: Not installed

     Publisher: test2

       Version: 1.0

        Branch: 0.2