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

# Use is subject to license terms.

import testutils

if __name__ == "__main__":

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

import os

import time

import unittest

import sys

from stat import *

import pkg.client.api as api

import pkg.client.api_errors as api_errors

import pkg.client.progress as progress

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 /tmp/libc.so.1 mode=0555 owner=root group=bin path=/lib/libc.so.1

            close """

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

        def setUp(self):

                testutils.SingleDepotTestCase.setUp(self)

                for p in self.misc_files:

                        f = open(p, "w")

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

                        # all files have differing contents

                        f.write(p)

                        f.close

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

        def tearDown(self):

                testutils.SingleDepotTestCase.tearDown(self)

                for p in self.misc_files:

                        os.remove(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(False)

                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(False)

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