From ddd7c0dd27fcddf94defc21502185520fd1d5303 Mon Sep 17 00:00:00 2001 From: Amrith Kumar Date: Fri, 5 Sep 2014 17:18:05 -0400 Subject: [PATCH] Fix issue with intermittent test failures in test_pkg.py The test failures were a result of randomized test order and the way in which these tests were setting up mocks for pexpect.spawn.expect and pexpect.spawn.match. The code now mocks pexpect.spawn and then sets return values and match in each test as appropriate. Code in setUp establishes a patch and a cleanUp function is added to reverse the patch. No code is required in tearDown as a result. execute() was being Mocked using code in setUp and tearDown and this was replaced that with a similar patch in setUp with a corresponding cleanUp function to reverse its effects. Since we're close to Juno, I've not gone down the path of converting this test to use a fixture which would further reduce duplicated code. Change-Id: I3edfd499823f2c2f4591e709384a24f1b55abe5a Closes-Bug: 1365531 --- trove/tests/unittests/guestagent/test_pkg.py | 212 +++++++++---------- 1 file changed, 105 insertions(+), 107 deletions(-) diff --git a/trove/tests/unittests/guestagent/test_pkg.py b/trove/tests/unittests/guestagent/test_pkg.py index d505818b20..94e4096a97 100644 --- a/trove/tests/unittests/guestagent/test_pkg.py +++ b/trove/tests/unittests/guestagent/test_pkg.py @@ -14,9 +14,8 @@ # under the License. import testtools -from mock import Mock, MagicMock +from mock import Mock, MagicMock, patch import pexpect -from trove.common import utils from trove.guestagent import pkg import commands import re @@ -30,75 +29,75 @@ class PkgDEBInstallTestCase(testtools.TestCase): def setUp(self): super(PkgDEBInstallTestCase, self).setUp() - self.utils_execute = utils.execute - self.pexpect_spawn_init = pexpect.spawn.__init__ - self.pexpect_spawn_closed = pexpect.spawn.close self.pkg = pkg.DebianPackagerMixin() self.pkg_fix = self.pkg._fix self.pkg_fix_package_selections = self.pkg._fix_package_selections - utils.execute = Mock() - pexpect.spawn.__init__ = Mock(return_value=None) - pexpect.spawn.closed = Mock(return_value=None) + + p0 = patch('pexpect.spawn') + p0.start() + self.addCleanup(p0.stop) + + p1 = patch('trove.common.utils.execute') + p1.start() + self.addCleanup(p1.stop) + self.pkg._fix = Mock(return_value=None) self.pkg._fix_package_selections = Mock(return_value=None) self.pkgName = 'packageName' def tearDown(self): super(PkgDEBInstallTestCase, self).tearDown() - utils.execute = self.utils_execute - pexpect.spawn.__init__ = self.pexpect_spawn_init - pexpect.spawn.close = self.pexpect_spawn_closed self.pkg._fix = self.pkg_fix self.pkg._fix_package_selections = self.pkg_fix_package_selections - def test_pkg_is_instaled_no_packages(self): + def test_pkg_is_installed_no_packages(self): packages = [] self.assertTrue(self.pkg.pkg_is_installed(packages)) - def test_pkg_is_instaled_yes(self): + def test_pkg_is_installed_yes(self): packages = ["package1=1.0", "package2"] self.pkg.pkg_version = MagicMock(side_effect=["1.0", "2.0"]) self.assertTrue(self.pkg.pkg_is_installed(packages)) - def test_pkg_is_instaled_no(self): + def test_pkg_is_installed_no(self): packages = ["package1=1.0", "package2", "package3=3.1"] self.pkg.pkg_version = MagicMock(side_effect=["1.0", "2.0", "3.0"]) self.assertFalse(self.pkg.pkg_is_installed(packages)) def test_success_install(self): # test - pexpect.spawn.expect = Mock(return_value=7) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 7 + pexpect.spawn.return_value.match = False self.assertTrue(self.pkg.pkg_install(self.pkgName, {}, 5000) is None) def test_permission_error(self): # test - pexpect.spawn.expect = Mock(return_value=0) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 0 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgPermissionError, self.pkg.pkg_install, self.pkgName, {}, 5000) def test_package_not_found_1(self): # test - pexpect.spawn.expect = Mock(return_value=1) - pexpect.spawn.match = re.match('(.*)', self.pkgName) + pexpect.spawn.return_value.expect.return_value = 1 + pexpect.spawn.return_value.match = re.match('(.*)', self.pkgName) # test and verify self.assertRaises(pkg.PkgNotFoundError, self.pkg.pkg_install, self.pkgName, {}, 5000) def test_package_not_found_2(self): # test - pexpect.spawn.expect = Mock(return_value=2) - pexpect.spawn.match = re.match('(.*)', self.pkgName) + pexpect.spawn.return_value.expect.return_value = 2 + pexpect.spawn.return_value.match = re.match('(.*)', self.pkgName) # test and verify self.assertRaises(pkg.PkgNotFoundError, self.pkg.pkg_install, self.pkgName, {}, 5000) def test_run_DPKG_bad_State(self): # test _fix method is called and PackageStateError is thrown - pexpect.spawn.expect = Mock(return_value=4) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 4 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgPackageStateError, self.pkg.pkg_install, self.pkgName, {}, 5000) @@ -106,23 +105,23 @@ class PkgDEBInstallTestCase(testtools.TestCase): def test_admin_lock_error(self): # test 'Unable to lock the administration directory' error - pexpect.spawn.expect = Mock(return_value=5) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 5 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgAdminLockError, self.pkg.pkg_install, self.pkgName, {}, 5000) def test_package_broken_error(self): - pexpect.spawn.expect = Mock(return_value=6) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 6 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgBrokenError, self.pkg.pkg_install, self.pkgName, {}, 5000) def test_timeout_error(self): # test timeout error - pexpect.spawn.expect = Mock(side_effect=pexpect. - TIMEOUT('timeout error')) + pexpect.spawn.return_value.expect.side_effect = ( + pexpect.TIMEOUT('timeout error')) # test and verify self.assertRaises(pkg.PkgTimeout, self.pkg.pkg_install, self.pkgName, {}, 5000) @@ -132,17 +131,19 @@ class PkgDEBRemoveTestCase(testtools.TestCase): def setUp(self): super(PkgDEBRemoveTestCase, self).setUp() - self.utils_execute = utils.execute - self.pexpect_spawn_init = pexpect.spawn.__init__ - self.pexpect_spawn_closed = pexpect.spawn.close self.pkg = pkg.DebianPackagerMixin() self.pkg_version = self.pkg.pkg_version self.pkg_install = self.pkg._install self.pkg_fix = self.pkg._fix - utils.execute = Mock() - pexpect.spawn.__init__ = Mock(return_value=None) - pexpect.spawn.closed = Mock(return_value=None) + p0 = patch('pexpect.spawn') + p0.start() + self.addCleanup(p0.stop) + + p1 = patch('trove.common.utils.execute') + p1.start() + self.addCleanup(p1.stop) + self.pkg.pkg_version = Mock(return_value="OK") self.pkg._install = Mock(return_value=None) self.pkg._fix = Mock(return_value=None) @@ -151,39 +152,36 @@ class PkgDEBRemoveTestCase(testtools.TestCase): def tearDown(self): super(PkgDEBRemoveTestCase, 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 self.pkg._fix = self.pkg_fix def test_success_remove(self): # test - pexpect.spawn.expect = Mock(return_value=6) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 6 + pexpect.spawn.return_value.match = False self.assertTrue(self.pkg.pkg_remove(self.pkgName, 5000) is None) def test_permission_error(self): # test - pexpect.spawn.expect = Mock(return_value=0) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 0 + pexpect.spawn.return_value.match = False # 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) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 1 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgNotFoundError, self.pkg.pkg_remove, self.pkgName, 5000) def test_package_reinstall_first_1(self): # test - pexpect.spawn.expect = Mock(return_value=2) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 2 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgPackageStateError, self.pkg.pkg_remove, self.pkgName, 5000) @@ -192,8 +190,8 @@ class PkgDEBRemoveTestCase(testtools.TestCase): def test_package_reinstall_first_2(self): # test - pexpect.spawn.expect = Mock(return_value=3) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 3 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgPackageStateError, self.pkg.pkg_remove, self.pkgName, 5000) @@ -202,8 +200,8 @@ class PkgDEBRemoveTestCase(testtools.TestCase): def test_package_DPKG_first(self): # test - pexpect.spawn.expect = Mock(return_value=4) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 4 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgPackageStateError, self.pkg.pkg_remove, self.pkgName, 5000) @@ -212,16 +210,16 @@ class PkgDEBRemoveTestCase(testtools.TestCase): def test_admin_lock_error(self): # test 'Unable to lock the administration directory' error - pexpect.spawn.expect = Mock(return_value=5) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 5 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgAdminLockError, self.pkg.pkg_remove, self.pkgName, 5000) def test_timeout_error(self): # test timeout error - pexpect.spawn.expect = Mock(side_effect=pexpect. - TIMEOUT('timeout error')) + pexpect.spawn.return_value.expect.side_effect = ( + pexpect.TIMEOUT('timeout error')) # test and verify self.assertRaises(pkg.PkgTimeout, self.pkg.pkg_remove, self.pkgName, 5000) @@ -286,32 +284,31 @@ 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' + p0 = patch('pexpect.spawn') + p0.start() + self.addCleanup(p0.stop) + + p1 = patch('trove.common.utils.execute') + p1.start() + self.addCleanup(p1.stop) + 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_pkg_is_instaled_no_packages(self): + def test_pkg_is_installed_no_packages(self): packages = [] self.assertTrue(self.pkg.pkg_is_installed(packages)) - def test_pkg_is_instaled_yes(self): + def test_pkg_is_installed_yes(self): packages = ["package1=1.0", "package2"] commands.getstatusoutput = MagicMock(return_value={1: "package1=1.0\n" "package2=2.0"}) self.assertTrue(self.pkg.pkg_is_installed(packages)) - def test_pkg_is_instaled_no(self): + def test_pkg_is_installed_no(self): packages = ["package1=1.0", "package2", "package3=3.0"] commands.getstatusoutput = MagicMock(return_value={1: "package1=1.0\n" "package2=2.0"}) @@ -319,24 +316,24 @@ class PkgRPMInstallTestCase(testtools.TestCase): def test_permission_error(self): # test - pexpect.spawn.expect = Mock(return_value=0) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 0 + pexpect.spawn.return_value.match = False # 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) - pexpect.spawn.match = re.match('(.*)', self.pkgName) + pexpect.spawn.return_value.expect.return_value = 1 + pexpect.spawn.return_value.match = re.match('(.*)', self.pkgName) # test and verify self.assertRaises(pkg.PkgNotFoundError, self.pkg.pkg_install, self.pkgName, {}, 5000) def test_package_conflict_remove(self): # test - pexpect.spawn.expect = Mock(return_value=2) - pexpect.spawn.match = re.match('(.*)', self.pkgName) + pexpect.spawn.return_value.expect.return_value = 2 + pexpect.spawn.return_value.match = re.match('(.*)', self.pkgName) self.pkg._rpm_remove_nodeps = Mock() # test and verify self.pkg._install(self.pkgName, 5000) @@ -344,62 +341,62 @@ class PkgRPMInstallTestCase(testtools.TestCase): def test_package_scriptlet_error(self): # test - pexpect.spawn.expect = Mock(return_value=5) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 5 + pexpect.spawn.return_value.match = False # 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=6) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 6 + pexpect.spawn.return_value.match = False # 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=7) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 7 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgDownloadError, self.pkg.pkg_install, self.pkgName, {}, 5000) def test_package_sign_error(self): # test - pexpect.spawn.expect = Mock(return_value=8) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 8 + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgSignError, self.pkg.pkg_install, self.pkgName, {}, 5000) def test_package_already_installed(self): # test - pexpect.spawn.expect = Mock(return_value=9) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 9 + pexpect.spawn.return_value.match = False # 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=10) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 10 + pexpect.spawn.return_value.match = False # 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=11) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 11 + pexpect.spawn.return_value.match = False # 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')) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.side_effect = ( + pexpect.TIMEOUT('timeout error')) + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgTimeout, self.pkg.pkg_install, self.pkgName, {}, 5000) @@ -409,53 +406,54 @@ 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) + + p0 = patch('pexpect.spawn') + p0.start() + self.addCleanup(p0.stop) + + p1 = patch('trove.common.utils.execute') + p1.start() + self.addCleanup(p1.stop) + 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) + pexpect.spawn.return_value.expect.return_value = 0 + pexpect.spawn.return_value.match = False # 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) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 1 + pexpect.spawn.return_value.match = False # 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) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.return_value = 2 + pexpect.spawn.return_value.match = False 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')) - pexpect.spawn.match = False + pexpect.spawn.return_value.expect.side_effect = ( + pexpect.TIMEOUT('timeout error')) + pexpect.spawn.return_value.match = False # test and verify self.assertRaises(pkg.PkgTimeout, self.pkg.pkg_remove, self.pkgName, 5000)