#!/usr/bin/python2.4

#

# 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 2009 Sun Microsystems, Inc.  All rights reserved.

# Use is subject to license terms.

#

import testutils

if __name__ == "__main__":

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

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.fmri as fmri

import sys

import tempfile

import time

import unittest

API_VERSION = 15

PKG_CLIENT_NAME = "pkg"

class TestPkgApi(testutils.SingleDepotTestCase):

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

        persistent_depot = False

        foo10 = """

            open [email protected],5.11-0

            close """

        foo12 = """

            open [email protected],5.11-0

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

            close """

        bar10 = """

            open [email protected],5.11-0

            close """

        p5i_bobcat = """{

  "packages": [

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

    "baz"

  ], 

  "publishers": [

    {

      "alias": "cat", 

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

            "http://localhost:12001/"

          ], 

          "refresh_seconds": 43200, 

          "registration_uri": "", 

          "related_uris": []

        }

      ]

    }

  ], 

  "version": 1

        misc_files = [ "libc.so.1" ]

        def setUp(self):

                testutils.SingleDepotTestCase.setUp(self)

                self.foo12 = self.foo12.replace("$test_prefix",

                    self.get_test_prefix())

                for p in self.misc_files:

                        fpath = os.path.join(self.get_test_prefix(), p)

                        f = open(fpath, "wb")

                        # write the name of the file into the file, so that

                        # all files have differing contents

                        f.write(fpath)

                        f.close()

                        self.debug("wrote %s" % fpath)

        def tearDown(self):

                testutils.SingleDepotTestCase.tearDown(self)

                for p in self.misc_files:

                        os.remove(os.path.join(self.get_test_prefix(), p))

        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):

                durl = self.dc.get_depot_url()

                self.pkgsend_bulk(durl, self.foo10)

                self.image_create(durl)

                progresstracker = progress.NullProgressTracker()

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

                    progresstracker, lambda x: False, PKG_CLIENT_NAME)

                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(durl, 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 """

                durl = self.dc.get_depot_url()

                self.pkgsend_bulk(durl, self.foo10)

                self.image_create(durl)

                progresstracker = progress.NullProgressTracker()

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

                    progresstracker, lambda x: False, PKG_CLIENT_NAME)

                recursive_removal = False

                filters = []

                api_obj.plan_install(["foo"], filters)

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

                api_obj.reset()

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

                api_obj.plan_install(["foo"], filters)

                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"], filters)

                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(durl, 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_properties(self):

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

                accessible and return expected values."""

                durl = self.dc.get_depot_url()

                self.image_create(durl)

                progresstracker = progress.NullProgressTracker()

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

                    progresstracker, lambda x: False, PKG_CLIENT_NAME)

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

                durl = self.dc.get_depot_url()

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

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

                progresstracker = progress.NullProgressTracker()

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

                    progresstracker, lambda x: False, PKG_CLIENT_NAME)

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

                self.assertRaises(api_errors.UnknownPublisher,

                    api_obj.get_publisher)

                # Verify that publisher objects returned from get_publishers

                # match those returned by get_publisher.

                pubs = api_obj.get_publishers()

                self.assertEqual(pub.prefix, pubs[0].prefix)

                self.assertEqual(id(pub), id(pubs[0]))

                # Verify that duplicate actually creates duplicates.

                cpub = api_obj.get_publisher(prefix="bobcat", duplicate=True)

                self.assertNotEqual(id(pub), id(cpub))

                # Now modify publisher information and update.

                cpub.alias = "cat"

                repo = cpub.selected_repository

                repo.name = "source"

                repo.description = "xkcd.net/325"

                repo.legal_uris = ["http://xkcd.com/license.html"]

                repo.refresh_seconds = 43200

                repo.registered = False

                api_obj.update_publisher(cpub)

                # Verify that the update happened.

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

                self.assertEqual(pub.alias, "cat")

                repo = pub.selected_repository

                self.assertEqual(repo.name, "source")

                self.assertEqual(repo.description, "xkcd.net/325")

                self.assertEqual(repo.legal_uris[0],

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

                self.assertEqual(repo.refresh_seconds, 43200)

                self.assertEqual(repo.registered, False)

                # Verify that duplicates match their original.

                cpub = api_obj.get_publisher(alias=pub.alias, duplicate=True)

                for p in ("alias", "prefix", "meta_root"):

                        self.assertEqual(getattr(pub, p), getattr(cpub, p))

                for p in ("collection_type", "description", "legal_uris",

                    "mirrors", "name", "origins", "refresh_seconds",

                    "registered", "registration_uri", "related_uris",

                    "sort_policy"):

                        srepo = pub.selected_repository

                        crepo = cpub.selected_repository

                        self.assertEqual(getattr(srepo, p), getattr(crepo, p))

                cpub = None

                cpubs = api_obj.get_publishers(duplicate=True)

                self.assertNotEqual(id(pub), id(cpubs[0]))

                cpubs = None

                # Verify that publisher_last_update_time returns a value.

                self.assertTrue(

                    api_obj.get_publisher_last_update_time("bobcat"))

                # Verify that p5i export and parse works as expected.

                # Ensure that PackageInfo, PkgFmri, and strings are all

                # supported properly.

                # Strip timestamp information so that comparison with

                # pre-generated test data will succeed.

                ffoo = fmri.PkgFmri(plist[0])

                sfoo = str(ffoo).replace(":%s" % ffoo.version.timestr, "")

                ffoo = fmri.PkgFmri(sfoo)

                fbar = fmri.PkgFmri(plist[1])

                sbar = str(fbar).replace(":%s" % fbar.version.timestr, "")

                fbar = fmri.PkgFmri(sbar)

                # Build a simple list of packages.

                pnames = {

                    "bobcat": (api.PackageInfo(ffoo),),

                    "": [fbar, "baz"],

                }

                # Dump the p5i data.

                fobj = cStringIO.StringIO()

                api_obj.write_p5i(fobj, pkg_names=pnames, pubs=[pub])

                # Verify that output matches expected output.

                fobj.seek(0)

                output = fobj.read()

                self.assertEqual(output, self.p5i_bobcat)

                def validate_results(results):

                        # First result should be 'bobcat' publisher and its

                        # pkg_names.

                        pub, pkg_names = results[0]

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

                        self.assertEqual(pub.alias, "cat")

                        repo = pub.selected_repository

                        self.assertEqual(repo.name, "source")

                        self.assertEqual(repo.description, "xkcd.net/325")

                        self.assertEqual(repo.legal_uris[0],

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

                        self.assertEqual(repo.refresh_seconds, 43200)

                        self.assertEqual(pkg_names, [sfoo])

                        # Last result should be no publisher and a list of

                        # pkg_names.

                        pub, pkg_names = results[1]

                        self.assertEqual(pub, None)

                        self.assertEqual(pkg_names, [sbar, "baz"])

                # Verify that parse returns the expected object and information

                # when provided a fileobj.

                fobj.seek(0)

                validate_results(api_obj.parse_p5i(fileobj=fobj))

                # Verify that an add of the parsed object works (the name has to

                # be changed to prevent a duplicate error here).

                fobj.seek(0)

                results = api_obj.parse_p5i(fileobj=fobj)

                pub, pkg_names = results[0]

                pub.prefix = "p5icat"

                pub.alias = "copycat"

                api_obj.add_publisher(pub, refresh_allowed=False)

                # Now verify that we can retrieve the added publisher.

                api_obj.get_publisher(prefix=pub.prefix)

                cpub = api_obj.get_publisher(alias=pub.alias, duplicate=True)

                # Now update the publisher and set it to disabled, to verify

                # that api functions still work as expected.

                cpub.disabled = True

                api_obj.update_publisher(cpub)

                cpub = api_obj.get_publisher(alias=cpub.alias, duplicate=True)

                self.assertTrue(cpub.disabled)

                self.assertTrue(api_obj.has_publisher(prefix=cpub.prefix))

                # Now attempt to update the disabled publisher.

                cpub = api_obj.get_publisher(alias=cpub.alias, duplicate=True)

                cpub.alias = "copycopycat"

                api_obj.update_publisher(cpub)

                cpub = None

                # Verify that parse returns the expected object and information

                # when provided a file path.

                fobj.seek(0)

                (fd1, path1) = tempfile.mkstemp(dir=self.get_test_prefix())

                os.write(fd1, fobj.read())

                validate_results(api_obj.parse_p5i(location=path1))

                # Verify that parse returns the expected object and information

                # when provided a file URI.

                validate_results(api_obj.parse_p5i(location="file://" + path1))

                fobj.close()

                fobj = None

                # Verify that appropriate exceptions are raised for p5i

                # information that can't be retrieved (doesn't exist).

                nefpath = os.path.join(self.get_test_prefix(), "non-existent")

                self.assertRaises(api_errors.RetrievalError,

                    api_obj.parse_p5i, location="file://%s" % nefpath)

                self.assertRaises(api_errors.RetrievalError,

                    api_obj.parse_p5i, location=nefpath)

                # Verify that appropriate exceptions are raised for invalid

                # p5i information.

                lcpath = os.path.join(self.get_test_prefix(), "libc.so.1")

                self.assertRaises(api_errors.InvalidP5IFile, api_obj.parse_p5i,

                    location="file://%s" % lcpath)

                self.assertRaises(api_errors.InvalidP5IFile, api_obj.parse_p5i,

                    location=lcpath)