1710N/A#!/usr/bin/python
1710N/A#
1710N/A# CDDL HEADER START
1710N/A#
1710N/A# The contents of this file are subject to the terms of the
1710N/A# Common Development and Distribution License (the "License").
1710N/A# You may not use this file except in compliance with the License.
1710N/A#
1710N/A# You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
1710N/A# or http://www.opensolaris.org/os/licensing.
1710N/A# See the License for the specific language governing permissions
1710N/A# and limitations under the License.
1710N/A#
1710N/A# When distributing Covered Code, include this CDDL HEADER in each
1710N/A# file and include the License file at usr/src/OPENSOLARIS.LICENSE.
1710N/A# If applicable, add the following below this CDDL HEADER, with the
1710N/A# fields enclosed by brackets "[]" replaced with your own identifying
1710N/A# information: Portions Copyright [yyyy] [name of copyright owner]
1710N/A#
1710N/A# CDDL HEADER END
1710N/A#
1710N/A
1710N/A#
3339N/A# Copyright (c) 2010, 2016, Oracle and/or its affiliates. All rights reserved.
1710N/A#
1710N/A
3339N/Afrom . import testutils
1710N/Aif __name__ == "__main__":
1710N/A testutils.setup_environment("../../../proto")
1715N/Aimport pkg5unittest
1710N/A
1710N/Aimport os
1710N/Aimport pkg.client.api_errors as api_errors
1710N/Aimport pkg.client.progress as progress
1710N/Aimport pkg.fmri as fmri
1710N/Aimport pkg.nrlock as nrlock
1710N/Aimport unittest
1710N/A
1710N/A
1715N/Aclass TestPkgApi(pkg5unittest.SingleDepotTestCase):
1710N/A # restart the depot for every test
1710N/A persistent_depot = False
1710N/A
1710N/A foo10 = """
1710N/A open foo@1.0,5.11-0
1710N/A close """
1710N/A
1710N/A foo11 = """
1710N/A open foo@1.1,5.11-0
1710N/A close """
1710N/A
1710N/A def setUp(self):
1715N/A pkg5unittest.SingleDepotTestCase.setUp(self)
1895N/A self.pkgsend_bulk(self.rurl, (self.foo10, self.foo11))
1895N/A self.image_create(self.rurl)
1710N/A
1710N/A def test_api_locking(self):
1710N/A """Verify that a locked image cannot be modified if it is
1710N/A already locked and that the API will raise an appropriate
1710N/A exception."""
1710N/A
1710N/A # Get an image object and tests its manual lock mechanism.
1895N/A api_obj = self.get_img_api_obj()
1710N/A img = api_obj.img
1710N/A
1710N/A # Verify a lock file is created and is not zero size.
1710N/A img.lock()
1710N/A lfpath = os.path.join(img.imgdir, "lock")
1710N/A self.assertTrue(os.path.exists(lfpath))
1710N/A self.assertNotEqual(os.stat(lfpath).st_size, 0)
1710N/A
1710N/A # Verify attempting to re-lock when locked fails.
1710N/A self.assertRaises(nrlock.NRLockException, img.lock)
1710N/A
1710N/A # Verify lock file still exists on failure.
1710N/A self.assertTrue(os.path.exists(lfpath))
1710N/A
1710N/A # Verify that an API function will fail the same way.
2339N/A self.assertRaises(nrlock.NRLockException,
2339N/A lambda *args, **kwargs: list(
2339N/A api_obj.gen_plan_install(*args, **kwargs)),
1710N/A ["foo"])
1710N/A api_obj.reset()
1710N/A
1710N/A # Now verify that after unlocking the image that it will work.
1710N/A img.unlock()
1710N/A
1710N/A # Verify that after unlock, lock file still exists, but is
1710N/A # zero size.
1710N/A self.assertTrue(os.path.exists(lfpath))
1710N/A self.assertEqual(os.stat(lfpath).st_size, 0)
1710N/A
2339N/A for pd in api_obj.gen_plan_install(["foo"]):
2339N/A continue
1710N/A api_obj.reset()
1710N/A
1710N/A # Verify that if a state change occurs at any point after
1710N/A # planning before a plan successfully executes, that an
1710N/A # InvalidPlanError will be raised.
1895N/A api_obj2 = self.get_img_api_obj()
1710N/A
1710N/A # Both of these should succeed since no state change exists yet.
2339N/A for pd in api_obj.gen_plan_install(["foo"]):
2339N/A continue
2339N/A for pd in api_obj2.gen_plan_install(["foo"]):
2339N/A continue
1710N/A
1710N/A # Execute the first plan.
1710N/A api_obj.prepare()
1710N/A api_obj.execute_plan()
1710N/A api_obj.reset()
1710N/A
1710N/A # Now verify preparing second plan fails since first has changed
1710N/A # the image state.
1710N/A self.assertRaises(api_errors.InvalidPlanError, api_obj2.prepare)
1710N/A
1710N/A # Restart plan process.
1710N/A api_obj2.reset()
2382N/A for pd in api_obj2.gen_plan_uninstall(["foo"]):
2339N/A continue
2382N/A for pd in api_obj.gen_plan_uninstall(["foo"]):
2339N/A continue
1710N/A
1710N/A # Prepare second and first plan.
1710N/A api_obj2.prepare()
1710N/A api_obj.prepare()
1710N/A
1710N/A # Execute second plan, which should mean that the first can't
1710N/A # execute due to state change since plan was created.
1710N/A api_obj2.execute_plan()
1710N/A self.assertRaises(api_errors.InvalidPlanError,
1710N/A api_obj.execute_plan)
1710N/A
1710N/A def test_process_locking(self):
1710N/A """Verify that image locking works across processes."""
1710N/A
1710N/A # Get an image object and tests its manual lock mechanism.
1895N/A api_obj = self.get_img_api_obj()
1710N/A img = api_obj.img
1710N/A
1710N/A # Verify a lock file is created.
1710N/A img.lock()
1710N/A lfpath = os.path.join(img.imgdir, "lock")
1710N/A self.assertTrue(os.path.exists(lfpath))
1710N/A
1710N/A # Verify attempting to re-lock when lock fails.
1710N/A self.assertRaises(nrlock.NRLockException, img.lock)
1710N/A
1710N/A # Verify that the pkg process will fail if the image is locked.
1710N/A self.pkg("install foo", exit=7)
1710N/A
1710N/A # Now verify that after unlocking the image that it will work.
1710N/A img.unlock()
1710N/A self.pkg("install foo")
1710N/A
1710N/A # Now plan an uninstall using the API object.
1710N/A api_obj.reset()
2382N/A for pd in api_obj.gen_plan_uninstall(["foo"]):
2339N/A continue
1710N/A api_obj.prepare()
1710N/A
1710N/A # Execute the client to actually uninstall the package, and then
1710N/A # attempt to execute the plan which should fail since the image
1710N/A # state has changed since the plan was created.
1710N/A self.pkg("uninstall foo")
1710N/A self.assertRaises(api_errors.InvalidPlanError,
1710N/A api_obj.execute_plan)
3065N/A
3065N/A def test_symlink_lock(self):
3065N/A """Verify that if a install/refresh operation is performed
3065N/A on lock file, the lock file is not a symlink."""
3065N/A
3065N/A # Get an image object and tests its manual lock mechanism.
3065N/A api_obj = self.get_img_api_obj()
3065N/A img = api_obj.img
3065N/A
3065N/A # Verify a lock file is created.
3065N/A img.lock()
3065N/A lfpath = os.path.join(img.imgdir, "lock")
3065N/A self.assertTrue(os.path.exists(lfpath))
3065N/A img.unlock()
3065N/A os.remove(lfpath)
3065N/A
3065N/A # Make lock file a symlink by pointing it to a random file .
3065N/A tmp_file = os.path.join(img.imgdir, "test_symlink")
3065N/A fo = open(tmp_file, 'wb+')
3065N/A fo.close()
3065N/A os.symlink(tmp_file, lfpath)
3065N/A
3065N/A # Verify that both pkg install and refresh generate an error
3065N/A # if the lock file is a symlink.
3065N/A self.pkg("install foo", su_wrap=True, exit=1)
3065N/A self.assertTrue("contains a symlink" in self.errout)
3065N/A
3065N/A self.pkg("refresh --full", su_wrap=True, exit=1)
3065N/A self.assertTrue("contains a symlink" in self.errout)