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

#

import testutils

if __name__ == "__main__":

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

import pkg5unittest

import cStringIO

import os

import pkg.client.api as api

import pkg.client.api_errors as api_errors

import pkg.client.progress as progress

import pkg.facet as facet

import pkg.fmri as fmri

import sys

import tempfile

import time

import unittest

API_VERSION = 46

PKG_CLIENT_NAME = "pkg"

class TestPkgApi(pkg5unittest.SingleDepotTestCase):

        # restart the depot for every test

        persistent_setup = False

        foo10 = """

            open [email protected],5.11-0

            close """

        foo12 = """

            open [email protected],5.11-0

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

            close """

        bar10 = """

            open [email protected],5.11-0

            close """

        baz10 = """

            open [email protected],5.11-0

            add license copyright.baz license=copyright.baz

            close """

        quux10 = """

            open [email protected],5.11-0

            add depend type=require [email protected]

            close """

        # First iteration has just a copyright.

        licensed10 = """

            open [email protected],5.11-0

            add depend type=require [email protected]

            add license copyright.licensed license=copyright.licensed

            close """

        # Second iteration has copyright that must-display and a new license

        # that doesn't require acceptance.

        licensed12 = """

            open [email protected],5.11-0

            add depend type=require [email protected]

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

            add license copyright.licensed license=copyright.licensed must-display=True

            add license license.licensed license=license.licensed

            close """

        # Third iteration now requires acceptance of license.

        licensed13 = """

            open [email protected],5.11-0

            add depend type=require [email protected]

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

            add license copyright.licensed license=copyright.licensed must-display=True

            add license license.licensed license=license.licensed must-accept=True

            close """

        p5i_bobcat = """{

  "packages": [

    "pkg:/[email protected],5.11-0", 

    "baz"

  ], 

  "publishers": [

    {

      "alias": "cat", 

      "intermediate_certs": [], 

      "name": "bobcat", 

      "packages": [

        "pkg:/[email protected],5.11-0"

      ], 

      "repositories": [

        {

          "collection_type": "core", 

          "description": "xkcd.net/325", 

          "legal_uris": [

            "http://xkcd.com/license.html"

          ], 

          "mirrors": [], 

          "name": "source", 

          "origins": [

            "%REAL_ORIGIN%/"

          ], 

          "refresh_seconds": 43200, 

          "registration_uri": "", 

          "related_uris": []

        }

      ], 

      "signing_ca_certs": []

    }

  ], 

  "version": 1

}

"""

        misc_files = ["copyright.baz", "copyright.licensed", "libc.so.1",

            "license.licensed", "license.licensed.addendum"]

        def setUp(self):

                pkg5unittest.SingleDepotTestCase.setUp(self, publisher="bobcat")

                self.make_misc_files(self.misc_files)

                self.p5i_bobcat = self.p5i_bobcat.replace("%REAL_ORIGIN%",

                    self.rurl)

        def __try_bad_installs(self, api_obj):

                self.assertRaises(api_errors.PlanExistsException,

                    api_obj.plan_install,["foo"])

                self.assertRaises(api_errors.PlanExistsException,

                    api_obj.plan_uninstall,["foo"], False)

                self.assertRaises(api_errors.PlanExistsException,

                    api_obj.plan_update_all, sys.argv[0])

                try:

                        api_obj.plan_update_all(sys.argv[0])

                except api_errors.PlanExistsException:

                        pass

                else:

                        assert 0

        def __try_bad_combinations_and_complete(self, api_obj):

                self.__try_bad_installs(api_obj)

                self.assertRaises(api_errors.PrematureExecutionException,

                    api_obj.execute_plan)

                api_obj.prepare()

                self.__try_bad_installs(api_obj)

                self.assertRaises(api_errors.AlreadyPreparedException,

                    api_obj.prepare)

                api_obj.execute_plan()

                self.__try_bad_installs(api_obj)

                self.assertRaises(api_errors.AlreadyPreparedException,

                    api_obj.prepare)

                self.assertRaises(api_errors.AlreadyExecutedException,

                    api_obj.execute_plan)

        def test_bad_orderings(self):

                self.pkgsend_bulk(self.rurl, self.foo10)

                api_obj = self.image_create(self.rurl, prefix="bobcat")

                self.assert_(api_obj.describe() is None)

                self.assertRaises(api_errors.PlanMissingException,

                    api_obj.prepare)

                api_obj.plan_install(["foo"])

                self.__try_bad_combinations_and_complete(api_obj)

                api_obj.reset()

                self.assertRaises(api_errors.PlanMissingException,

                    api_obj.prepare)

                self.assert_(api_obj.describe() is None)

                self.pkgsend_bulk(self.rurl, self.foo12)

                api_obj.refresh(immediate=True)

                api_obj.plan_update_all(sys.argv[0])

                self.__try_bad_combinations_and_complete(api_obj)

                api_obj.reset()

                self.assertRaises(api_errors.PlanMissingException,

                    api_obj.prepare)

                self.assert_(api_obj.describe() is None)

                api_obj.plan_uninstall(["foo"], False)

                self.__try_bad_combinations_and_complete(api_obj)

                api_obj.reset()

                self.assertRaises(api_errors.PlanMissingException,

                    api_obj.prepare)

                self.assert_(api_obj.describe() is None)

        def test_reset(self):

                """ Send empty package [email protected], install and uninstall """

                self.pkgsend_bulk(self.rurl, self.foo10)

                api_obj = self.image_create(self.rurl, prefix="bobcat")

                recursive_removal = False

                facets = facet.Facets({ "facet.devel": True })

                api_obj.plan_change_varcets(facets=facets)

                self._api_finish(api_obj)

                api_obj.plan_install(["foo"])

                self.assert_(api_obj.describe() is not None)

                api_obj.reset()

                self.assert_(api_obj.describe() is None)

                api_obj.plan_install(["foo"])

                self.assert_(api_obj.describe() is not None)

                api_obj.prepare()

                api_obj.reset()

                self.assert_(api_obj.describe() is None)

                api_obj.plan_install(["foo"])

                self.assert_(api_obj.describe() is not None)

                api_obj.prepare()

                api_obj.execute_plan()

                api_obj.reset()

                self.assert_(api_obj.describe() is None)

                self.pkg("list")

                self.pkg("verify")

                self.pkgsend_bulk(self.rurl, self.foo12)

                api_obj.refresh(immediate=True)

                api_obj.plan_update_all(sys.argv[0])

                self.assert_(api_obj.describe() is not None)

                api_obj.reset()

                self.assert_(api_obj.describe() is None)

                api_obj.plan_update_all(sys.argv[0])

                self.assert_(api_obj.describe() is not None)

                api_obj.prepare()

                api_obj.reset()

                self.assert_(api_obj.describe() is None)

                api_obj.plan_update_all(sys.argv[0])

                self.assert_(api_obj.describe() is not None)

                api_obj.prepare()

                api_obj.execute_plan()

                api_obj.reset()

                self.assert_(api_obj.describe() is None)

                self.pkg("list")

                self.pkg("verify")

                api_obj.plan_uninstall(["foo"], recursive_removal)

                self.assert_(api_obj.describe() is not None)

                api_obj.reset()

                self.assert_(api_obj.describe() is None)

                api_obj.plan_uninstall(["foo"], recursive_removal)

                self.assert_(api_obj.describe() is not None)

                api_obj.prepare()

                api_obj.reset()

                self.assert_(api_obj.describe() is None)

                api_obj.plan_uninstall(["foo"], recursive_removal)

                self.assert_(api_obj.describe() is not None)

                api_obj.prepare()

                api_obj.execute_plan()

                api_obj.reset()

                self.assert_(api_obj.describe() is None)

                self.pkg("verify")

        def test_refresh_transition(self):

                """Verify that refresh works for a v0 catalog source and that

                if the client transitions from v0 to v1 or back that the correct

                state information is recorded in the image catalog."""

                # This test requires an actual depot due to v0 operation usage.

                # First create the image and get v1 catalog.

                self.dc.start()

                self.pkgsend_bulk(self.durl, (self.foo10, self.quux10))

                try:

                        api_obj = self.image_create(self.durl, prefix="bobcat")

                except api_errors.CatalogRefreshException, e:

                        self.debug("\n".join(str(x[-1]) for x in e.failed))

                        raise

                self.pkg("publisher")

                img = api_obj.img

                kcat = img.get_catalog(img.IMG_CATALOG_KNOWN)

                entry = [e for f, e in kcat.entries()][0]

                states = entry["metadata"]["states"]

                self.assert_(img.PKG_STATE_V1 in states)

                self.assert_(img.PKG_STATE_V0 not in states)

                # Next, disable v1 catalog for the depot and force a client

                # refresh.  Only v0 state should be present.

                self.dc.stop()

                self.dc.set_disable_ops(["catalog/1"])

                self.dc.start()

                # Since the depot state changed while the API object was

                # still active, it needs to be reset to clear the internal

                # transport state cache.

                api_obj.reset()

                api_obj.refresh(immediate=True)

                api_obj.reset()

                img = api_obj.img

                kcat = img.get_catalog(img.IMG_CATALOG_KNOWN)

                entry = [e for f, e in kcat.entries()][0]

                states = entry["metadata"]["states"]

                self.assert_(img.PKG_STATE_V1 not in states)

                self.assert_(img.PKG_STATE_V0 in states)

                # Verify that there is no dependency information present

                # in the known or installed catalog.

                icat = img.get_catalog(img.IMG_CATALOG_INSTALLED)

                for cat in kcat, icat:

                        dpart = cat.get_part("catalog.dependency.C")

                        dep_acts = [

                            acts

                            for t, entry in dpart.tuple_entries()

                            for acts in entry.get("actions", [])

                        ]

                        self.assertEqual(dep_acts, [])

                # Now install a package, and verify that the entries in the

                # known catalog for installed packages exist in the installed

                # catalog and are identical.

                progresstracker = progress.NullProgressTracker()

                api_obj = api.ImageInterface(self.get_img_path(), API_VERSION,

                    progresstracker, lambda x: False, PKG_CLIENT_NAME)

                img = api_obj.img

                # Get image catalogs.

                kcat = img.get_catalog(img.IMG_CATALOG_KNOWN)

                icat = img.get_catalog(img.IMG_CATALOG_INSTALLED)

                # Verify quux package is only in known catalog.

                self.assertTrue("quux" in kcat.names())

                self.assertTrue("foo" in kcat.names())

                self.assertTrue("quux" not in icat.names())

                self.assertTrue("foo" not in icat.names())

                # Install the packages.

                api_obj.plan_install(["[email protected]"])

                api_obj.prepare()

                api_obj.execute_plan()

                api_obj.reset()

                # Get image catalogs.

                kcat = img.get_catalog(img.IMG_CATALOG_KNOWN)

                icat = img.get_catalog(img.IMG_CATALOG_INSTALLED)

                # Verify quux package is in both catalogs.

                self.assertTrue("quux" in kcat.names())

                self.assertTrue("foo" in kcat.names())

                self.assertTrue("quux" in icat.names())

                self.assertTrue("foo" in icat.names())

                # Verify state info.

                for cat in kcat, icat:

                        entry = [e for f, e in cat.entries()][0]

                        states = entry["metadata"]["states"]

                        self.assert_(img.PKG_STATE_INSTALLED in states)

                        self.assert_(img.PKG_STATE_V0 in states)

                # Finally, transition back to v1 catalog.  This requires

                # creating a new api object since transport will think that

                # v1 catalogs are still unsupported.

                self.dc.unset_disable_ops()

                self.dc.stop()

                self.dc.start()

                progresstracker = progress.NullProgressTracker()

                api_obj = api.ImageInterface(self.get_img_path(), API_VERSION,

                    progresstracker, lambda x: False, PKG_CLIENT_NAME)

                api_obj.refresh(immediate=True)

                img = api_obj.img

                # Get image catalogs.

                kcat = img.get_catalog(img.IMG_CATALOG_KNOWN)

                icat = img.get_catalog(img.IMG_CATALOG_INSTALLED)

                # Verify quux package is in both catalogs.

                self.assertTrue("quux" in kcat.names())

                self.assertTrue("foo" in kcat.names())

                self.assertTrue("quux" in icat.names())

                self.assertTrue("foo" in icat.names())

                # Verify state info.

                for f, entry in kcat.entries():

                        states = entry["metadata"]["states"]

                        self.assert_(img.PKG_STATE_V1 in states)

                        self.assert_(img.PKG_STATE_V0 not in states)

                # Verify that there is dependency information present

                # in the known and installed catalog.

                for cat in kcat, icat:

                        dpart = cat.get_part("catalog.dependency.C")

                        entries = [(t, entry) for t, entry in dpart.tuple_entries()]

                        dep_acts = [

                            acts

                            for t, entry in dpart.tuple_entries()

                            for acts in entry.get("actions", [])

                            if t[1] == "quux"

                        ]

                        self.assertNotEqual(dep_acts, [])

                # Verify that every installed package is in known and has

                # identical entries and that every installed package in

                # the installed catalog is in the known catalog and has

                # entries.

                for src, dest in ((kcat, icat), (icat, kcat)):

                        src_base = src.get_part("catalog.base.C",

                            must_exist=True)

                        self.assertNotEqual(src_base, None)

                        for f, bentry in src_base.entries():

                                states = bentry["metadata"]["states"]

                                if img.PKG_STATE_INSTALLED not in states:

                                        continue

                                for name in src.parts:

                                        spart = src.get_part(name,

                                            must_exist=True)

                                        self.assertNotEqual(spart, None)

                                        dpart = dest.get_part(name,

                                            must_exist=True)

                                        self.assertNotEqual(dpart, None)

                                        sentry = spart.get_entry(pfmri=f)

                                        dentry = dpart.get_entry(pfmri=f)

                                        self.assertEqual(sentry, dentry)

        def test_properties(self):

                """Verify that properties of the ImageInterface api object are

                accessible and return expected values."""

                api_obj = self.image_create(self.rurl, prefix="bobcat")

                self.assertEqual(api_obj.root, self.img_path)

        def test_publisher_apis(self):

                """Verify that the publisher api methods work as expected.

                Note that not all methods are tested here as this would be

                redundant since other tests for the client will use those

                methods indirectly."""

                plist = self.pkgsend_bulk(self.rurl, (self.foo10, self.bar10))

                api_obj = self.image_create(self.rurl, prefix="bobcat")

                # Verify that existence tests succeed.

                self.assertTrue(api_obj.has_publisher("bobcat"))

                # Verify preferred publisher prefix is returned correctly.

                self.assertEqual(api_obj.get_preferred_publisher(), "bobcat")

                # Verify that get_publisher returned the correct publisher object.

                pub = api_obj.get_publisher(prefix="bobcat")

                self.assertEqual(pub.prefix, "bobcat")

                # Verify that not specifying matching criteria for get_publisher

                # raises a UnknownPublisher exception.

                self.assertRaises(api_errors.UnknownPublisher,

                    api_obj.get_publisher, "zuul")