Add RPM class implementation

implements blueprint rpm-integration

Change-Id: I4aa06c75b555d0c9ab54791b25c612bc5c5805fc

requirements.txt

@@ -20,3 +20,4 @@ iso8601
 oslo.config>=1.1.0
 jsonschema>=1.0.0,!=1.4.0,<2
 Jinja2
+pexpect

test-requirements.txt

@@ -16,7 +16,6 @@ http://tarballs.openstack.org/python-troveclient/python-troveclient-master.tar.g
 mock
 mox
 testtools>=0.9.22
-pexpect
 discover
 testrepository>=0.0.8
 mockito

trove/guestagent/manager/mysql_service.py

@@ -14,7 +14,7 @@ from trove.common import utils as utils
 from trove.common import exception
 from trove.guestagent import query
 from trove.guestagent.db import models
-from trove.guestagent.pkg import Package
+from trove.guestagent import pkg
 from trove.instance import models as rd_models
 from trove.openstack.common import log as logging
 from trove.openstack.common.gettextutils import _
@@ -39,7 +39,7 @@ INCLUDE_MARKER_OPERATORS = {
 }
 
 # Create a package impl
-pkg = Package()
+packager = pkg.Package()
 
 
 def generate_random_password():
@@ -678,7 +678,7 @@ class MySqlApp(object):
     def _install_mysql(self):
         """Install mysql server. The current version is 5.5"""
         LOG.debug(_("Installing mysql server"))
-        pkg.pkg_install(self.MYSQL_PACKAGE_VERSION, self.TIME_OUT)
+        packager.pkg_install(self.MYSQL_PACKAGE_VERSION, self.TIME_OUT)
         LOG.debug(_("Finished installing mysql server"))
         #TODO(rnirmal): Add checks to make sure the package got installed
 
@@ -698,6 +698,8 @@ class MySqlApp(object):
             command = command % locals()
         else:
             command = "sudo update-rc.d mysql enable"
+            if pkg.OS == pkg.REDHAT:
+                command = "sudo chkconfig mysql on"
         utils.execute_with_timeout(command, shell=True)
 
     def _disable_mysql_on_boot(self):
@@ -716,6 +718,8 @@ class MySqlApp(object):
             command = command % locals()
         else:
             command = "sudo update-rc.d mysql disable"
+            if pkg.OS == pkg.REDHAT:
+                command = "sudo chkconfig mysql off"
         utils.execute_with_timeout(command, shell=True)
 
     def stop_db(self, update_db=False, do_not_start_on_reboot=False):
@@ -869,7 +873,7 @@ class MySqlApp(object):
 
     def is_installed(self):
         #(cp16net) could raise an exception, does it need to be handled here?
-        version = pkg.pkg_version(self.MYSQL_PACKAGE_VERSION)
+        version = packager.pkg_version(self.MYSQL_PACKAGE_VERSION)
         return not version is None
 
 

trove/guestagent/pkg.py

@@ -34,6 +34,13 @@ LOG = logging.getLogger(__name__)
 OK = 0
 RUN_DPKG_FIRST = 1
 REINSTALL_FIRST = 2
+REDHAT = 'redhat'
+DEBIAN = 'debian'
+
+# The default is debian
+OS = DEBIAN
+if os.path.isfile("/etc/redhat-release"):
+    OS = REDHAT
 
 
 class PkgAdminLockError(exception.TroveError):
@@ -56,34 +63,41 @@ class PkgTimeout(exception.TroveError):
     pass
 
 
-class RedhatPackagerMixin:
+class PkgScriptletError(exception.TroveError):
-
+    pass
-    def pkg_install(self, package_name, time_out):
-        pass
-
-    def pkg_version(self, package_name):
-        return "1.0"
-
-    def pkg_remove(self, package_name, time_out):
-        pass
 
 
-class DebianPackagerMixin:
+class PkgTransactionCheckError(exception.TroveError):
+    pass
 
-    def kill_proc(self, child):
+
+class PkgDownloadError(exception.TroveError):
+    pass
+
+
+class BasePackagerMixin:
+
+    def pexpect_kill_proc(self, child):
         child.delayafterclose = 1
         child.delayafterterminate = 1
         child.close(force=True)
 
-    def wait_and_close_proc(self, child, time_out=-1):
+    def pexpect_wait_and_close_proc(self, child, time_out=-1):
         child.expect(pexpect.EOF, timeout=time_out)
         child.close()
 
-    def _fix(self, time_out):
+    def pexpect_run(self, cmd, output_expects, time_out):
-        """Sometimes you have to run this command before a pkg will install."""
+        child = pexpect.spawn(cmd)
-        #sudo dpkg --configure -a
+        try:
-        child = pexpect.spawn("sudo -E dpkg --configure -a")
+            i = child.expect(output_expects, timeout=time_out)
-        self.wait_and_close_proc(child, time_out)
+            self.pexpect_wait_and_close_proc(child)
+        except pexpect.TIMEOUT:
+            self.pexpect_kill_proc(child)
+            raise PkgTimeout("Process timeout after %i seconds." % time_out)
+        return i
+
+
+class RedhatPackagerMixin(BasePackagerMixin):
 
     def _install(self, package_name, time_out):
         """Attempts to install a package.
@@ -93,37 +107,27 @@ class DebianPackagerMixin:
         Raises an exception if a non-recoverable error or time out occurs.
 
         """
-        child = pexpect.spawn("sudo -E DEBIAN_FRONTEND=noninteractive "
+        cmd = "sudo yum --color=never -y install %s" % package_name
-                              "apt-get -y --allow-unauthenticated install %s"
+        output_expects = ['\[sudo\] password for .*:',
-                              % package_name)
+                          'No package %s available.' % package_name,
-        try:
+                          'Transaction Check Error:',
-            i = child.expect(['.*password*',
+                          '.*scriptlet failed*',
-                              'E: Unable to locate package %s' % package_name,
+                          'HTTP Error',
-                              "Couldn't find package % s" % package_name,
+                          'No more mirrors to try.',
-                              ("dpkg was interrupted, you must manually run "
+                          '.*already installed and latest version',
-                               "'sudo dpkg --configure -a'"),
+                          'Updated:',
-                              "Unable to lock the administration directory",
+                          'Installed:']
-                              "Setting up %s*" % package_name,
+        i = self.pexpect_run(cmd, output_expects, time_out)
-                              "is already the newest version"],
+        if i == 0:
-                             timeout=time_out)
+            raise PkgPermissionError("Invalid permissions.")
-            if i == 0:
+        elif i == 1:
-                raise PkgPermissionError("Invalid permissions.")
+            raise PkgNotFoundError("Could not find pkg %s" % package_name)
-            elif i == 1 or i == 2:
+        elif i == 2:
-                raise PkgNotFoundError("Could not find apt %s" % package_name)
+            raise PkgTransactionCheckError("Transaction Check Error")
-            elif i == 3:
+        elif i == 3:
-                return RUN_DPKG_FIRST
+            raise PkgScriptletError("Package scriptlet failed")
-            elif i == 4:
+        elif i == 4 or i == 5:
-                raise PkgAdminLockError()
+            raise PkgDownloadError("Package download problem")
-        except pexpect.TIMEOUT:
-            self.kill_proc(child)
-            raise PkgTimeout("Process timeout after %i seconds." % time_out)
-        try:
-            self.wait_and_close_proc(child)
-        except pexpect.TIMEOUT as e:
-            LOG.error("wait_and_close_proc failed: %s" % e)
-            #TODO(tim.simpson): As of RDL, and on my machine exclusively (in
-            #                   both Virtual Box and VmWare!) this fails, but
-            #                   the package is installed.
         return OK
 
     def _remove(self, package_name, time_out):
@@ -134,34 +138,117 @@ class DebianPackagerMixin:
         Raises an exception if a non-recoverable error or time out occurs.
 
         """
-        child = pexpect.spawn("sudo -E apt-get -y --allow-unauthenticated "
+        cmd = "sudo yum --color=never -y remove %s" % package_name
-                              "remove %s" % package_name)
+        output_expects = ['\[sudo\] password for .*:',
+                          'No Packages marked for removal',
+                          'Removed:']
+        i = self.pexpect_run(cmd, output_expects, time_out)
+        if i == 0:
+            raise PkgPermissionError("Invalid permissions.")
+        elif i == 1:
+            raise PkgNotFoundError("Could not find pkg %s" % package_name)
+        return OK
+
+    def pkg_install(self, package_name, time_out):
+        result = self._install(package_name, time_out)
+        if result != OK:
+            raise PkgPackageStateError("Package %s is in a bad state."
+                                       % package_name)
+
+    def pkg_version(self, package_name):
+        cmd_list = ["rpm", "-qa", "--qf", "'%{VERSION}-%{RELEASE}\n'",
+                    package_name]
+        p = commands.getstatusoutput(' '.join(cmd_list))
+        # Need to capture the version string
+        # check the command output
+        std_out = p[1]
+        for line in std_out.split("\n"):
+            regex = re.compile("[0-9.]+-.*")
+            matches = regex.match(line)
+            if matches:
+                line = matches.group()
+                return line
+        msg = _("version() saw unexpected output from rpm!")
+        LOG.error(msg)
+
+    def pkg_remove(self, package_name, time_out):
+        """Removes a package."""
+        if self.pkg_version(package_name) is None:
+            return
+        result = self._remove(package_name, time_out)
+        if result != OK:
+            raise PkgPackageStateError("Package %s is in a bad state."
+                                       % package_name)
+
+
+class DebianPackagerMixin(BasePackagerMixin):
+
+    def _fix(self, time_out):
+        """Sometimes you have to run this command before a pkg will install."""
         try:
-            i = child.expect(['.*password*',
+            utils.execute("dpkg", "--configure", "-a", run_as_root=True,
-                              'E: Unable to locate package %s' % package_name,
+                          root_helper="sudo")
-                              'Package is in a very bad inconsistent state',
+        except ProcessExecutionError as e:
-                              ("Sub-process /usr/bin/dpkg returned an error "
+            LOG.error(_("Error fixing dpkg"))
-                               "code"),
+
-                              ("dpkg was interrupted, you must manually run "
+    def _install(self, package_name, time_out):
-                               "'sudo dpkg --configure -a'"),
+        """Attempts to install a package.
-                              "Unable to lock the administration directory",
+
-                              #'The following packages will be REMOVED',
+        Returns OK if the package installs fine or a result code if a
-                              "Removing %s*" % package_name],
+        recoverable-error occurred.
-                             timeout=time_out)
+        Raises an exception if a non-recoverable error or time out occurs.
-            if i == 0:
+
-                raise PkgPermissionError("Invalid permissions.")
+        """
-            elif i == 1:
+        cmd = "sudo -E DEBIAN_FRONTEND=noninteractive " \
-                raise PkgNotFoundError("Could not find pkg %s" % package_name)
+              "apt-get -y --allow-unauthenticated install %s" % package_name
-            elif i == 2 or i == 3:
+        output_expects = ['.*password*',
-                return REINSTALL_FIRST
+                          'E: Unable to locate package %s' % package_name,
-            elif i == 4:
+                          "Couldn't find package % s" % package_name,
-                return RUN_DPKG_FIRST
+                          ("dpkg was interrupted, you must manually run "
-            elif i == 5:
+                           "'sudo dpkg --configure -a'"),
-                raise PkgAdminLockError()
+                          "Unable to lock the administration directory",
-            self.wait_and_close_proc(child)
+                          "Setting up %s*" % package_name,
-        except pexpect.TIMEOUT:
+                          "is already the newest version"]
-            self.kill_proc(child)
+        i = self.pexpect_run(cmd, output_expects, time_out)
-            raise PkgTimeout("Process timeout after %i seconds." % time_out)
+        if i == 0:
+            raise PkgPermissionError("Invalid permissions.")
+        elif i == 1 or i == 2:
+            raise PkgNotFoundError("Could not find apt %s" % package_name)
+        elif i == 3:
+            return RUN_DPKG_FIRST
+        elif i == 4:
+            raise PkgAdminLockError()
+        return OK
+
+    def _remove(self, package_name, time_out):
+        """Removes a package.
+
+        Returns OK if the package is removed successfully or a result code if a
+        recoverable-error occurs.
+        Raises an exception if a non-recoverable error or time out occurs.
+
+        """
+        cmd = "sudo -E apt-get -y --allow-unauthenticated remove %s" \
+              % package_name
+        output_expects = ['.*password*',
+                          'E: Unable to locate package %s' % package_name,
+                          'Package is in a very bad inconsistent state',
+                          'Sub-process /usr/bin/dpkg returned an error code',
+                          ("dpkg was interrupted, you must manually run "
+                           "'sudo dpkg --configure -a'"),
+                          "Unable to lock the administration directory",
+                          "Removing %s*" % package_name]
+        i = self.pexpect_run(cmd, output_expects, time_out)
+        if i == 0:
+            raise PkgPermissionError("Invalid permissions.")
+        elif i == 1:
+            raise PkgNotFoundError("Could not find pkg %s" % package_name)
+        elif i == 2 or i == 3:
+            return REINSTALL_FIRST
+        elif i == 4:
+            return RUN_DPKG_FIRST
+        elif i == 5:
+            raise PkgAdminLockError()
         return OK
 
     def pkg_install(self, package_name, time_out):
@@ -216,7 +303,6 @@ class DebianPackagerMixin:
                     return parts[2]
         msg = _("version() saw unexpected output from dpkg!")
         LOG.error(msg)
-        raise exception.GuestError(msg)
 
     def pkg_remove(self, package_name, time_out):
         """Removes a package."""
@@ -238,9 +324,7 @@ class DebianPackagerMixin:
 class BasePackage(type):
 
     def __new__(meta, name, bases, dct):
-        if os.path.isfile("/etc/debian_version"):
+        if OS == REDHAT:
-            bases += (DebianPackagerMixin, )
-        elif os.path.isfile("/etc/redhat-release"):
             bases += (RedhatPackagerMixin, )
         else:
             # The default is debian

trove/tests/unittests/guestagent/test_dbaas.py

@@ -618,12 +618,12 @@ class MySqlAppInstallTest(MySqlAppTest):
     def setUp(self):
         super(MySqlAppInstallTest, self).setUp()
         self.orig_create_engine = sqlalchemy.create_engine
-        self.orig_pkg_version = dbaas.pkg.pkg_version
+        self.orig_pkg_version = dbaas.packager.pkg_version
 
     def tearDown(self):
         super(MySqlAppInstallTest, self).tearDown()
         sqlalchemy.create_engine = self.orig_create_engine
-        dbaas.pkg.pkg_version = self.orig_pkg_version
+        dbaas.packager.pkg_version = self.orig_pkg_version
 
     def test_install(self):
 
@@ -684,13 +684,13 @@ class MySqlAppInstallTest(MySqlAppTest):
 
     def test_is_installed(self):
 
-        dbaas.pkg.pkg_version = Mock(return_value=True)
+        dbaas.packager.pkg_version = Mock(return_value=True)
 
         self.assertTrue(self.mySqlApp.is_installed())
 
     def test_is_installed_not(self):
 
-        dbaas.pkg.pkg_version = Mock(return_value=None)
+        dbaas.packager.pkg_version = Mock(return_value=None)
 
         self.assertFalse(self.mySqlApp.is_installed())
 

trove/tests/unittests/guestagent/test_pkg.py

@@ -29,10 +29,10 @@ Unit tests for the classes and functions in pkg.py.
 """
 
 
-class PkgInstallTestCase(testtools.TestCase):
+class PkgDEBInstallTestCase(testtools.TestCase):
 
     def setUp(self):
-        super(PkgInstallTestCase, self).setUp()
+        super(PkgDEBInstallTestCase, self).setUp()
         self.utils_execute = utils.execute
         self.pexpect_spawn_init = pexpect.spawn.__init__
         self.pexpect_spawn_closed = pexpect.spawn.close
@@ -45,7 +45,7 @@ class PkgInstallTestCase(testtools.TestCase):
         self.pkgName = 'packageName'
 
     def tearDown(self):
-        super(PkgInstallTestCase, self).tearDown()
+        super(PkgDEBInstallTestCase, self).tearDown()
         utils.execute = self.utils_execute
         pexpect.spawn.__init__ = self.pexpect_spawn_init
         pexpect.spawn.close = self.pexpect_spawn_closed
@@ -107,10 +107,10 @@ class PkgInstallTestCase(testtools.TestCase):
                           self.pkgName, 5000)
 
 
-class PkgRemoveTestCase(testtools.TestCase):
+class PkgDEBRemoveTestCase(testtools.TestCase):
 
     def setUp(self):
-        super(PkgRemoveTestCase, self).setUp()
+        super(PkgDEBRemoveTestCase, self).setUp()
         self.utils_execute = utils.execute
         self.pexpect_spawn_init = pexpect.spawn.__init__
         self.pexpect_spawn_closed = pexpect.spawn.close
@@ -129,7 +129,7 @@ class PkgRemoveTestCase(testtools.TestCase):
         self.pkgName = 'packageName'
 
     def tearDown(self):
-        super(PkgRemoveTestCase, self).tearDown()
+        super(PkgDEBRemoveTestCase, self).tearDown()
         utils.execute = self.utils_execute
         pexpect.spawn.__init__ = self.pexpect_spawn_init
         pexpect.spawn.close = self.pexpect_spawn_closed
@@ -199,7 +199,7 @@ class PkgRemoveTestCase(testtools.TestCase):
                           self.pkgName, 5000)
 
 
-class PkgVersionTestCase(testtools.TestCase):
+class PkgDEBVersionTestCase(testtools.TestCase):
 
     @staticmethod
     def build_output(packageName, packageVersion, parts=None):
@@ -218,13 +218,13 @@ class PkgVersionTestCase(testtools.TestCase):
         return cmd_out
 
     def setUp(self):
-        super(PkgVersionTestCase, self).setUp()
+        super(PkgDEBVersionTestCase, self).setUp()
         self.pkgName = 'mysql-server-5.5'
         self.pkgVersion = '5.5.28-0'
         self.commands_output = commands.getstatusoutput
 
     def tearDown(self):
-        super(PkgVersionTestCase, self).tearDown()
+        super(PkgDEBVersionTestCase, self).tearDown()
         commands.getstatusoutput = self.commands_output
 
     def test_version_success(self):
@@ -242,15 +242,13 @@ class PkgVersionTestCase(testtools.TestCase):
     def test_version_no_output(self):
         cmd_out = self.build_output(self.pkgName, self.pkgVersion, "")
         commands.getstatusoutput = Mock(return_value=(0, cmd_out))
-        self.assertRaises(exception.GuestError,
+        self.assertIsNone(pkg.DebianPackagerMixin().pkg_version(self.pkgName))
-                          pkg.DebianPackagerMixin().pkg_version, self.pkgName)
 
     def test_version_unexpected_parts(self):
         unexp_parts = "ii   123"
         cmd_out = self.build_output(self.pkgName, self.pkgVersion, unexp_parts)
         commands.getstatusoutput = Mock(return_value=(0, cmd_out))
-        self.assertRaises(exception.GuestError,
+        self.assertIsNone(pkg.DebianPackagerMixin().pkg_version(self.pkgName))
-                          pkg.DebianPackagerMixin().pkg_version, self.pkgName)
 
     def test_version_wrong_package(self):
         invalid_pkg = "package_invalid_001"
@@ -269,3 +267,169 @@ class PkgVersionTestCase(testtools.TestCase):
         cmd_out = self.build_output(self.pkgName, '<none>')
         commands.getstatusoutput = Mock(return_value=(0, cmd_out))
         self.assertFalse(pkg.DebianPackagerMixin().pkg_version(self.pkgName))
+
+
+class PkgRPMVersionTestCase(testtools.TestCase):
+
+    def setUp(self):
+        super(PkgRPMVersionTestCase, self).setUp()
+        self.pkgName = 'python-requests'
+        self.pkgVersion = '0.14.2-1.el6'
+        self.commands_output = commands.getstatusoutput
+
+    def tearDown(self):
+        super(PkgRPMVersionTestCase, self).tearDown()
+        commands.getstatusoutput = self.commands_output
+
+    def test_version_no_output(self):
+        cmd_out = ''
+        commands.getstatusoutput = Mock(return_value=(0, cmd_out))
+        self.assertIsNone(pkg.RedhatPackagerMixin().pkg_version(self.pkgName))
+
+    def test_version_success(self):
+        cmd_out = self.pkgVersion
+        commands.getstatusoutput = Mock(return_value=(0, cmd_out))
+        version = pkg.RedhatPackagerMixin().pkg_version(self.pkgName)
+        self.assertTrue(version)
+        self.assertEqual(self.pkgVersion, version)
+
+
+class PkgRPMInstallTestCase(testtools.TestCase):
+
+    def setUp(self):
+        super(PkgRPMInstallTestCase, self).setUp()
+        self.utils_execute = utils.execute
+        self.pexpect_spawn_init = pexpect.spawn.__init__
+        self.pexpect_spawn_closed = pexpect.spawn.close
+        self.pkg = pkg.RedhatPackagerMixin()
+        utils.execute = Mock()
+        pexpect.spawn.__init__ = Mock(return_value=None)
+        pexpect.spawn.closed = Mock(return_value=None)
+        self.pkgName = 'packageName'
+
+    def tearDown(self):
+        super(PkgRPMInstallTestCase, self).tearDown()
+        utils.execute = self.utils_execute
+        pexpect.spawn.__init__ = self.pexpect_spawn_init
+        pexpect.spawn.close = self.pexpect_spawn_closed
+
+    def test_permission_error(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=0)
+        # test and verify
+        self.assertRaises(pkg.PkgPermissionError, self.pkg.pkg_install,
+                          self.pkgName, 5000)
+
+    def test_package_not_found(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=1)
+        # test and verify
+        self.assertRaises(pkg.PkgNotFoundError, self.pkg.pkg_install,
+                          self.pkgName, 5000)
+
+    def test_transaction_check_error(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=2)
+        # test and verify
+        self.assertRaises(pkg.PkgTransactionCheckError, self.pkg.pkg_install,
+                          self.pkgName, 5000)
+
+    def test_package_scriptlet_error(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=3)
+        # test and verify
+        self.assertRaises(pkg.PkgScriptletError, self.pkg.pkg_install,
+                          self.pkgName, 5000)
+
+    def test_package_http_error(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=4)
+        # test and verify
+        self.assertRaises(pkg.PkgDownloadError, self.pkg.pkg_install,
+                          self.pkgName, 5000)
+
+    def test_package_nomirrors_error(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=5)
+        # test and verify
+        self.assertRaises(pkg.PkgDownloadError, self.pkg.pkg_install,
+                          self.pkgName, 5000)
+
+    def test_package_already_installed(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=6)
+        # test and verify
+        self.assertTrue(self.pkg.pkg_install(self.pkgName, 5000) is None)
+
+    def test_package_success_updated(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=7)
+        # test and verify
+        self.assertTrue(self.pkg.pkg_install(self.pkgName, 5000) is None)
+
+    def test_package_success_installed(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=8)
+        # test and verify
+        self.assertTrue(self.pkg.pkg_install(self.pkgName, 5000) is None)
+
+    def test_timeout_error(self):
+        # test timeout error
+        pexpect.spawn.expect = Mock(side_effect=pexpect.
+                                    TIMEOUT('timeout error'))
+        # test and verify
+        self.assertRaises(pkg.PkgTimeout, self.pkg.pkg_install,
+                          self.pkgName, 5000)
+
+
+class PkgRPMRemoveTestCase(testtools.TestCase):
+
+    def setUp(self):
+        super(PkgRPMRemoveTestCase, self).setUp()
+        self.utils_execute = utils.execute
+        self.pexpect_spawn_init = pexpect.spawn.__init__
+        self.pexpect_spawn_closed = pexpect.spawn.close
+        self.pkg = pkg.RedhatPackagerMixin()
+        self.pkg_version = self.pkg.pkg_version
+        self.pkg_install = self.pkg._install
+        utils.execute = Mock()
+        pexpect.spawn.__init__ = Mock(return_value=None)
+        pexpect.spawn.closed = Mock(return_value=None)
+        self.pkg.pkg_version = Mock(return_value="OK")
+        self.pkg._install = Mock(return_value=None)
+        self.pkgName = 'packageName'
+
+    def tearDown(self):
+        super(PkgRPMRemoveTestCase, self).tearDown()
+        utils.execute = self.utils_execute
+        pexpect.spawn.__init__ = self.pexpect_spawn_init
+        pexpect.spawn.close = self.pexpect_spawn_closed
+        self.pkg.pkg_version = self.pkg_version
+        self.pkg._install = self.pkg_install
+
+    def test_permission_error(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=0)
+        # test and verify
+        self.assertRaises(pkg.PkgPermissionError, self.pkg.pkg_remove,
+                          self.pkgName, 5000)
+
+    def test_package_not_found(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=1)
+        # test and verify
+        self.assertRaises(pkg.PkgNotFoundError, self.pkg.pkg_remove,
+                          self.pkgName, 5000)
+
+    def test_success_remove(self):
+        # test
+        pexpect.spawn.expect = Mock(return_value=2)
+        self.assertTrue(self.pkg.pkg_remove(self.pkgName, 5000) is None)
+
+    def test_timeout_error(self):
+        # test timeout error
+        pexpect.spawn.expect = Mock(side_effect=pexpect.
+                                    TIMEOUT('timeout error'))
+        # test and verify
+        self.assertRaises(pkg.PkgTimeout, self.pkg.pkg_remove,
+                          self.pkgName, 5000)