diff options
| author | Chris St. Pierre <chris.a.st.pierre@gmail.com> | 2013-02-20 07:52:00 -0500 |
|---|---|---|
| committer | Chris St. Pierre <chris.a.st.pierre@gmail.com> | 2013-02-20 07:52:07 -0500 |
| commit | be0de88922a58504c655361970378375426b5acc (patch) | |
| tree | af61cafdadf2634cbdad83ac0a5d1a9c5934a1ae /testsuite | |
| parent | 1f572cc7b6ce8462128fc1093458917af01f71ee (diff) | |
| download | bcfg2-be0de88922a58504c655361970378375426b5acc.tar.gz bcfg2-be0de88922a58504c655361970378375426b5acc.tar.bz2 bcfg2-be0de88922a58504c655361970378375426b5acc.zip | |
wrote unit tests for base client Tool classes
Diffstat (limited to 'testsuite')
11 files changed, 1319 insertions, 726 deletions
diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestDevice.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestDevice.py index 9f396b0b6..2f5eeb166 100644 --- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestDevice.py +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestDevice.py @@ -14,10 +14,10 @@ while path != "/": if os.path.basename(path) == "testsuite": break path = os.path.dirname(path) -from Test__init import get_posix_object from Testbase import TestPOSIXTool from common import * + class TestPOSIXDevice(TestPOSIXTool): test_obj = POSIXDevice @@ -37,48 +37,48 @@ class TestPOSIXDevice(TestPOSIXTool): entry.set("major", "0") entry.set("minor", "0") self.assertTrue(ptool.fully_specified(entry)) - + @patch("os.major") @patch("os.minor") - @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool._exists") @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.verify") - def test_verify(self, mock_verify, mock_exists, mock_minor, mock_major): + def test_verify(self, mock_verify, mock_minor, mock_major): entry = lxml.etree.Element("Path", name="/test", type="device", mode='0644', owner='root', group='root', dev_type="block", major="0", minor="10") ptool = self.get_obj() + ptool._exists = Mock() def reset(): - mock_exists.reset_mock() + ptool._exists.reset_mock() mock_verify.reset_mock() mock_minor.reset_mock() mock_major.reset_mock() - mock_exists.return_value = False + ptool._exists.return_value = False self.assertFalse(ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) + ptool._exists.assert_called_with(entry) reset() - mock_exists.return_value = MagicMock() + ptool._exists.return_value = MagicMock() mock_major.return_value = 0 mock_minor.return_value = 10 mock_verify.return_value = True self.assertTrue(ptool.verify(entry, [])) mock_verify.assert_called_with(ptool, entry, []) - mock_exists.assert_called_with(entry) - mock_major.assert_called_with(mock_exists.return_value.st_rdev) - mock_minor.assert_called_with(mock_exists.return_value.st_rdev) + ptool._exists.assert_called_with(entry) + mock_major.assert_called_with(ptool._exists.return_value.st_rdev) + mock_minor.assert_called_with(ptool._exists.return_value.st_rdev) reset() - mock_exists.return_value = MagicMock() + ptool._exists.return_value = MagicMock() mock_major.return_value = 0 mock_minor.return_value = 10 mock_verify.return_value = False self.assertFalse(ptool.verify(entry, [])) mock_verify.assert_called_with(ptool, entry, []) - mock_exists.assert_called_with(entry) - mock_major.assert_called_with(mock_exists.return_value.st_rdev) - mock_minor.assert_called_with(mock_exists.return_value.st_rdev) + ptool._exists.assert_called_with(entry) + mock_major.assert_called_with(ptool._exists.return_value.st_rdev) + mock_minor.assert_called_with(ptool._exists.return_value.st_rdev) reset() mock_verify.return_value = True @@ -86,26 +86,26 @@ class TestPOSIXDevice(TestPOSIXTool): mode='0644', owner='root', group='root', dev_type="fifo") self.assertTrue(ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) + ptool._exists.assert_called_with(entry) mock_verify.assert_called_with(ptool, entry, []) self.assertFalse(mock_major.called) self.assertFalse(mock_minor.called) - + @patch("os.makedev") @patch("os.mknod") - @patch("Bcfg2.Client.Tools.POSIX.Device.%s._exists" % test_obj.__name__) @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.install") - def test_install(self, mock_install, mock_exists, mock_mknod, mock_makedev): + def test_install(self, mock_install, mock_mknod, mock_makedev): entry = lxml.etree.Element("Path", name="/test", type="device", mode='0644', owner='root', group='root', dev_type="block", major="0", minor="10") ptool = self.get_obj() + ptool._exists = Mock() - mock_exists.return_value = False + ptool._exists.return_value = False mock_makedev.return_value = Mock() mock_install.return_value = True self.assertTrue(ptool.install(entry)) - mock_exists.assert_called_with(entry, remove=True) + ptool._exists.assert_called_with(entry, remove=True) mock_makedev.assert_called_with(0, 10) mock_mknod.assert_called_with(entry.get("name"), # 0o644 device_map[entry.get("dev_type")] | 420, @@ -114,29 +114,29 @@ class TestPOSIXDevice(TestPOSIXTool): mock_makedev.reset_mock() mock_mknod.reset_mock() - mock_exists.reset_mock() + ptool._exists.reset_mock() mock_install.reset_mock() mock_makedev.side_effect = OSError self.assertFalse(ptool.install(entry)) mock_makedev.reset_mock() mock_mknod.reset_mock() - mock_exists.reset_mock() + ptool._exists.reset_mock() mock_install.reset_mock() mock_mknod.side_effect = OSError self.assertFalse(ptool.install(entry)) - + mock_makedev.reset_mock() mock_mknod.reset_mock() - mock_exists.reset_mock() - mock_install.reset_mock() + ptool._exists.reset_mock() + mock_install.reset_mock() mock_mknod.side_effect = None entry = lxml.etree.Element("Path", name="/test", type="device", mode='0644', owner='root', group='root', dev_type="fifo") - + self.assertTrue(ptool.install(entry)) - mock_exists.assert_called_with(entry, remove=True) + ptool._exists.assert_called_with(entry, remove=True) mock_mknod.assert_called_with(entry.get("name"), # 0o644 device_map[entry.get("dev_type")] | 420) mock_install.assert_called_with(ptool, entry) diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestDirectory.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestDirectory.py index 16490808e..d9431dc63 100644 --- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestDirectory.py +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestDirectory.py @@ -15,46 +15,47 @@ while path != "/": if os.path.basename(path) == "testsuite": break path = os.path.dirname(path) -from Test__init import get_posix_object from Testbase import TestPOSIXTool from common import * + class TestPOSIXDirectory(TestPOSIXTool): test_obj = POSIXDirectory @patch("os.listdir") @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.verify") - @patch("Bcfg2.Client.Tools.POSIX.Directory.%s._exists" % test_obj.__name__) - def test_verify(self, mock_exists, mock_verify, mock_listdir): + def test_verify(self, mock_verify, mock_listdir): + ptool = self.get_obj() + ptool._exists = Mock() entry = lxml.etree.Element("Path", name="/test", type="directory", mode='0644', owner='root', group='root') - mock_exists.return_value = False - self.assertFalse(self.ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) + ptool._exists.return_value = False + self.assertFalse(ptool.verify(entry, [])) + ptool._exists.assert_called_with(entry) - mock_exists.reset_mock() + ptool._exists.reset_mock() exists_rv = MagicMock() exists_rv.__getitem__.return_value = stat.S_IFREG | 420 # 0o644 - mock_exists.return_value = exists_rv - self.assertFalse(self.ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) + ptool._exists.return_value = exists_rv + self.assertFalse(ptool.verify(entry, [])) + ptool._exists.assert_called_with(entry) - mock_exists.reset_mock() + ptool._exists.reset_mock() mock_verify.return_value = False exists_rv.__getitem__.return_value = stat.S_IFDIR | 420 # 0o644 - self.assertFalse(self.ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) - mock_verify.assert_called_with(self.ptool, entry, []) + self.assertFalse(ptool.verify(entry, [])) + ptool._exists.assert_called_with(entry) + mock_verify.assert_called_with(ptool, entry, []) - mock_exists.reset_mock() + ptool._exists.reset_mock() mock_verify.reset_mock() mock_verify.return_value = True - self.assertTrue(self.ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) - mock_verify.assert_called_with(self.ptool, entry, []) + self.assertTrue(ptool.verify(entry, [])) + ptool._exists.assert_called_with(entry) + mock_verify.assert_called_with(ptool, entry, []) - mock_exists.reset_mock() + ptool._exists.reset_mock() mock_verify.reset_mock() entry.set("prune", "true") orig_entry = copy.deepcopy(entry) @@ -62,9 +63,9 @@ class TestPOSIXDirectory(TestPOSIXTool): entries = ["foo", "bar", "bar/baz"] mock_listdir.return_value = entries modlist = [os.path.join(entry.get("name"), entries[0])] - self.assertFalse(self.ptool.verify(entry, modlist)) - mock_exists.assert_called_with(entry) - mock_verify.assert_called_with(self.ptool, entry, modlist) + self.assertFalse(ptool.verify(entry, modlist)) + ptool._exists.assert_called_with(entry) + mock_verify.assert_called_with(ptool, entry, modlist) mock_listdir.assert_called_with(entry.get("name")) expected = [os.path.join(entry.get("name"), e) for e in entries @@ -73,67 +74,65 @@ class TestPOSIXDirectory(TestPOSIXTool): self.assertItemsEqual(expected, actual) mock_verify.reset_mock() - mock_exists.reset_mock() + ptool._exists.reset_mock() mock_listdir.reset_mock() entry = copy.deepcopy(orig_entry) modlist = [os.path.join(entry.get("name"), e) for e in entries] - self.assertTrue(self.ptool.verify(entry, modlist)) - mock_exists.assert_called_with(entry) - mock_verify.assert_called_with(self.ptool, entry, modlist) + self.assertTrue(ptool.verify(entry, modlist)) + ptool._exists.assert_called_with(entry) + mock_verify.assert_called_with(ptool, entry, modlist) mock_listdir.assert_called_with(entry.get("name")) self.assertEqual(len(entry.findall("Prune")), 0) @patch("os.unlink") @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.install") - @patch("Bcfg2.Client.Tools.POSIX.Directory.%s._exists" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.Directory.%s._makedirs" % - test_obj.__name__) - def test_install(self, mock_makedirs, mock_exists, mock_install, - mock_unlink): + def test_install(self, mock_install, mock_unlink): entry = lxml.etree.Element("Path", name="/test/foo/bar", type="directory", mode='0644', owner='root', group='root') - self.ptool._makedirs = Mock() - self.ptool._remove = Mock() + ptool = self.get_obj() + ptool._exists = Mock() + ptool._makedirs = Mock() + ptool._remove = Mock() def reset(): - mock_exists.reset_mock() + ptool._exists.reset_mock() mock_install.reset_mock() mock_unlink.reset_mock() - self.ptool._makedirs.reset_mock() - self.ptool._remove.reset_mock() + ptool._makedirs.reset_mock() + ptool._remove.reset_mock() - self.ptool._makedirs.return_value = True - mock_exists.return_value = False + ptool._makedirs.return_value = True + ptool._exists.return_value = False mock_install.return_value = True - self.assertTrue(self.ptool.install(entry)) - mock_exists.assert_called_with(entry) - mock_install.assert_called_with(self.ptool, entry) - self.ptool._makedirs.assert_called_with(entry) + self.assertTrue(ptool.install(entry)) + ptool._exists.assert_called_with(entry) + mock_install.assert_called_with(ptool, entry) + ptool._makedirs.assert_called_with(entry) reset() exists_rv = MagicMock() exists_rv.__getitem__.return_value = stat.S_IFREG | 420 # 0o644 - mock_exists.return_value = exists_rv - self.assertTrue(self.ptool.install(entry)) + ptool._exists.return_value = exists_rv + self.assertTrue(ptool.install(entry)) mock_unlink.assert_called_with(entry.get("name")) - mock_exists.assert_called_with(entry) - self.ptool._makedirs.assert_called_with(entry) - mock_install.assert_called_with(self.ptool, entry) + ptool._exists.assert_called_with(entry) + ptool._makedirs.assert_called_with(entry) + mock_install.assert_called_with(ptool, entry) reset() exists_rv.__getitem__.return_value = stat.S_IFDIR | 420 # 0o644 mock_install.return_value = True - self.assertTrue(self.ptool.install(entry)) - mock_exists.assert_called_with(entry) - mock_install.assert_called_with(self.ptool, entry) + self.assertTrue(ptool.install(entry)) + ptool._exists.assert_called_with(entry) + mock_install.assert_called_with(ptool, entry) reset() mock_install.return_value = False - self.assertFalse(self.ptool.install(entry)) - mock_install.assert_called_with(self.ptool, entry) + self.assertFalse(ptool.install(entry)) + mock_install.assert_called_with(ptool, entry) entry.set("prune", "true") prune = ["/test/foo/bar/prune1", "/test/foo/bar/prune2"] @@ -143,9 +142,9 @@ class TestPOSIXDirectory(TestPOSIXTool): reset() mock_install.return_value = True - self.assertTrue(self.ptool.install(entry)) - mock_exists.assert_called_with(entry) - mock_install.assert_called_with(self.ptool, entry) + self.assertTrue(ptool.install(entry)) + ptool._exists.assert_called_with(entry) + mock_install.assert_called_with(ptool, entry) self.assertItemsEqual([c[0][0].get("path") - for c in self.ptool._remove.call_args_list], + for c in ptool._remove.call_args_list], prune) diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestFile.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestFile.py index 69b021421..662e0e1b6 100644 --- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestFile.py +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestFile.py @@ -3,9 +3,8 @@ import os import sys import copy import difflib -import binascii import lxml.etree -from Bcfg2.Compat import b64encode, b64decode, u_str +from Bcfg2.Compat import b64encode, u_str from mock import Mock, MagicMock, patch from Bcfg2.Client.Tools.POSIX.File import * @@ -18,47 +17,46 @@ while path != "/": if os.path.basename(path) == "testsuite": break path = os.path.dirname(path) -from Test__init import get_posix_object -from Testbase import TestPOSIXTool +from TestPOSIX.Testbase import TestPOSIXTool from common import * -def get_file_object(posix=None): - if posix is None: - posix = get_posix_object() - return POSIXFile(posix.logger, posix.setup, posix.config) class TestPOSIXFile(TestPOSIXTool): test_obj = POSIXFile def test_fully_specified(self): + ptool = self.get_obj() + entry = lxml.etree.Element("Path", name="/test", type="file") - self.assertFalse(self.ptool.fully_specified(entry)) + self.assertFalse(ptool.fully_specified(entry)) entry.set("empty", "true") - self.assertTrue(self.ptool.fully_specified(entry)) + self.assertTrue(ptool.fully_specified(entry)) entry.set("empty", "false") entry.text = "text" - self.assertTrue(self.ptool.fully_specified(entry)) - + self.assertTrue(ptool.fully_specified(entry)) + def test_is_string(self): + ptool = self.get_obj() + for char in list(range(8)) + list(range(14, 32)): - self.assertFalse(self.ptool._is_string("foo" + chr(char) + "bar", + self.assertFalse(ptool._is_string("foo" + chr(char) + "bar", 'UTF-8')) for char in list(range(9, 14)) + list(range(33, 128)): - self.assertTrue(self.ptool._is_string("foo" + chr(char) + "bar", + self.assertTrue(ptool._is_string("foo" + chr(char) + "bar", 'UTF-8')) ustr = 'é' - self.assertTrue(self.ptool._is_string(ustr, 'UTF-8')) + self.assertTrue(ptool._is_string(ustr, 'UTF-8')) if not inPy3k: - self.assertFalse(self.ptool._is_string("foo" + chr(128) + "bar", + self.assertFalse(ptool._is_string("foo" + chr(128) + "bar", 'ascii')) - self.assertFalse(self.ptool._is_string(ustr, 'ascii')) + self.assertFalse(ptool._is_string(ustr, 'ascii')) def test_get_data(self): orig_entry = lxml.etree.Element("Path", name="/test", type="file") - setup = dict(encoding="ascii", ppath='/', max_copies=5) - ptool = self.get_obj(posix=get_posix_object(setup=setup)) + ptool = self.get_obj(setup=dict(encoding="ascii", ppath='/', + max_copies=5)) entry = copy.deepcopy(orig_entry) entry.text = b64encode("test") @@ -83,124 +81,122 @@ class TestPOSIXFile(TestPOSIXTool): @patch("%s.open" % builtins) @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.verify") - @patch("Bcfg2.Client.Tools.POSIX.File.%s._exists" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.File.%s._get_data" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.File.%s._get_diffs" % test_obj.__name__) - def test_verify(self, mock_get_diffs, mock_get_data, mock_exists, - mock_verify, mock_open): + def test_verify(self, mock_verify, mock_open): entry = lxml.etree.Element("Path", name="/test", type="file") - setup = dict(interactive=False, ppath='/', max_copies=5) - ptool = self.get_obj(posix=get_posix_object(setup=setup)) + ptool = self.get_obj(setup=dict(interactive=False, ppath='/', + max_copies=5)) + ptool._exists = Mock() + ptool._get_data = Mock() + ptool._get_diffs = Mock() def reset(): - mock_get_diffs.reset_mock() - mock_get_data.reset_mock() - mock_exists.reset_mock() + ptool._get_diffs.reset_mock() + ptool._get_data.reset_mock() + ptool._exists.reset_mock() mock_verify.reset_mock() mock_open.reset_mock() - mock_get_data.return_value = ("test", False) - mock_exists.return_value = False + ptool._get_data.return_value = ("test", False) + ptool._exists.return_value = False mock_verify.return_value = True self.assertFalse(ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) + ptool._exists.assert_called_with(entry) mock_verify.assert_called_with(ptool, entry, []) - mock_get_diffs.assert_called_with(entry, interactive=False, - sensitive=False, - is_binary=False, + ptool._get_diffs.assert_called_with(entry, interactive=False, + sensitive=False, is_binary=False, content="") reset() exists_rv = MagicMock() exists_rv.__getitem__.return_value = 5 - mock_exists.return_value = exists_rv - mock_get_data.return_value = ("test", True) + ptool._exists.return_value = exists_rv + ptool._get_data.return_value = ("test", True) self.assertFalse(ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) + ptool._exists.assert_called_with(entry) mock_verify.assert_called_with(ptool, entry, []) - mock_get_diffs.assert_called_with(entry, interactive=False, - sensitive=False, - is_binary=True, + ptool._get_diffs.assert_called_with(entry, interactive=False, + sensitive=False, is_binary=True, content=None) - + reset() - mock_get_data.return_value = ("test", False) + ptool._get_data.return_value = ("test", False) exists_rv.__getitem__.return_value = 4 entry.set("sensitive", "true") mock_open.return_value.read.return_value = "tart" self.assertFalse(ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) + ptool._exists.assert_called_with(entry) mock_verify.assert_called_with(ptool, entry, []) mock_open.assert_called_with(entry.get("name")) mock_open.return_value.read.assert_called_with() - mock_get_diffs.assert_called_with(entry, interactive=False, - sensitive=True, - is_binary=False, + ptool._get_diffs.assert_called_with(entry, interactive=False, + sensitive=True, is_binary=False, content="tart") reset() mock_open.return_value.read.return_value = "test" self.assertTrue(ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) + ptool._exists.assert_called_with(entry) mock_verify.assert_called_with(ptool, entry, []) mock_open.assert_called_with(entry.get("name")) mock_open.return_value.read.assert_called_with() - self.assertFalse(mock_get_diffs.called) + self.assertFalse(ptool._get_diffs.called) reset() mock_open.side_effect = IOError self.assertFalse(ptool.verify(entry, [])) - mock_exists.assert_called_with(entry) + ptool._exists.assert_called_with(entry) mock_open.assert_called_with(entry.get("name")) - + @patch("os.fdopen") @patch("tempfile.mkstemp") - @patch("Bcfg2.Client.Tools.POSIX.File.%s._get_data" % test_obj.__name__) - def test_write_tmpfile(self, mock_get_data, mock_mkstemp, mock_fdopen): + def test_write_tmpfile(self, mock_mkstemp, mock_fdopen): + ptool = self.get_obj() + ptool._get_data = Mock() entry = lxml.etree.Element("Path", name="/test", type="file", mode='0644', owner='root', group='root') newfile = "/foo/bar" def reset(): - mock_get_data.reset_mock() + ptool._get_data.reset_mock() mock_mkstemp.reset_mock() mock_fdopen.reset_mock() - mock_get_data.return_value = ("test", False) + ptool._get_data.return_value = ("test", False) mock_mkstemp.return_value = (5, newfile) - self.assertEqual(self.ptool._write_tmpfile(entry), newfile) - mock_get_data.assert_called_with(entry) + self.assertEqual(ptool._write_tmpfile(entry), newfile) + ptool._get_data.assert_called_with(entry) mock_mkstemp.assert_called_with(prefix='test', dir='/') mock_fdopen.assert_called_with(5, 'w') mock_fdopen.return_value.write.assert_called_with("test") reset() mock_mkstemp.side_effect = OSError - self.assertFalse(self.ptool._write_tmpfile(entry)) + self.assertFalse(ptool._write_tmpfile(entry)) mock_mkstemp.assert_called_with(prefix='test', dir='/') reset() mock_mkstemp.side_effect = None mock_fdopen.side_effect = OSError - self.assertFalse(self.ptool._write_tmpfile(entry)) + self.assertFalse(ptool._write_tmpfile(entry)) mock_mkstemp.assert_called_with(prefix='test', dir='/') - mock_get_data.assert_called_with(entry) + ptool._get_data.assert_called_with(entry) mock_fdopen.assert_called_with(5, 'w') - + @patch("os.rename") @patch("os.unlink") def test_rename_tmpfile(self, mock_unlink, mock_rename): + ptool = self.get_obj() entry = lxml.etree.Element("Path", name="/test", type="file", mode='0644', owner='root', group='root') newfile = "/foo/bar" - self.assertTrue(self.ptool._rename_tmpfile(newfile, entry)) + self.assertTrue(ptool._rename_tmpfile(newfile, entry)) mock_rename.assert_called_with(newfile, entry.get("name")) - + mock_rename.reset_mock() mock_unlink.reset_mock() mock_rename.side_effect = OSError - self.assertFalse(self.ptool._rename_tmpfile(newfile, entry)) + self.assertFalse(ptool._rename_tmpfile(newfile, entry)) mock_rename.assert_called_with(newfile, entry.get("name")) mock_unlink.assert_called_with(newfile) @@ -208,58 +204,57 @@ class TestPOSIXFile(TestPOSIXTool): mock_rename.reset_mock() mock_unlink.reset_mock() mock_unlink.side_effect = OSError - self.assertFalse(self.ptool._rename_tmpfile(newfile, entry)) + self.assertFalse(ptool._rename_tmpfile(newfile, entry)) mock_rename.assert_called_with(newfile, entry.get("name")) mock_unlink.assert_called_with(newfile) @patch("%s.open" % builtins) - @patch("Bcfg2.Client.Tools.POSIX.File.%s._diff" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.File.%s._get_data" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.File.%s._is_string" % test_obj.__name__) - def test__get_diffs(self, mock_is_string, mock_get_data, mock_diff, - mock_open): + def test__get_diffs(self, mock_open): orig_entry = lxml.etree.Element("Path", name="/test", type="file", mode='0644', owner='root', group='root') orig_entry.text = "test" ondisk = "test2" - setup = dict(encoding="utf-8", ppath='/', max_copies=5) - ptool = self.get_obj(posix=get_posix_object(setup=setup)) + ptool = self.get_obj(setup=dict(encoding="utf-8", ppath='/', + max_copies=5)) + ptool._get_data = Mock() + ptool._diff = Mock() + ptool._is_string = Mock() def reset(): - mock_is_string.reset_mock() - mock_get_data.reset_mock() - mock_diff.reset_mock() + ptool._is_string.reset_mock() + ptool._get_data.reset_mock() + ptool._diff.reset_mock() mock_open.reset_mock() return copy.deepcopy(orig_entry) - - mock_is_string.return_value = True - mock_get_data.return_value = (orig_entry.text, False) + + ptool._is_string.return_value = True + ptool._get_data.return_value = (orig_entry.text, False) mock_open.return_value.read.return_value = ondisk - mock_diff.return_value = ["-test2", "+test"] + ptool._diff.return_value = ["-test2", "+test"] # binary data in the entry entry = reset() ptool._get_diffs(entry, is_binary=True) mock_open.assert_called_with(entry.get("name")) mock_open.return_value.read.assert_any_call() - self.assertFalse(mock_diff.called) + self.assertFalse(ptool._diff.called) self.assertEqual(entry.get("current_bfile"), b64encode(ondisk)) # binary data on disk entry = reset() - mock_is_string.return_value = False + ptool._is_string.return_value = False ptool._get_diffs(entry, content=ondisk) self.assertFalse(mock_open.called) - self.assertFalse(mock_diff.called) + self.assertFalse(ptool._diff.called) self.assertEqual(entry.get("current_bfile"), b64encode(ondisk)) # sensitive, non-interactive -- do nothing entry = reset() - mock_is_string.return_value = True + ptool._is_string.return_value = True ptool._get_diffs(entry, sensitive=True, interactive=False) self.assertFalse(mock_open.called) - self.assertFalse(mock_diff.called) + self.assertFalse(ptool._diff.called) self.assertXMLEqual(entry, orig_entry) # sensitive, interactive @@ -267,8 +262,9 @@ class TestPOSIXFile(TestPOSIXTool): ptool._get_diffs(entry, sensitive=True, interactive=True) mock_open.assert_called_with(entry.get("name")) mock_open.return_value.read.assert_any_call() - mock_diff.assert_called_with(ondisk, entry.text, difflib.unified_diff, - filename=entry.get("name")) + ptool._diff.assert_called_with(ondisk, entry.text, + difflib.unified_diff, + filename=entry.get("name")) self.assertIsNotNone(entry.get("qtext")) del entry.attrib['qtext'] self.assertItemsEqual(orig_entry.attrib, entry.attrib) @@ -277,11 +273,11 @@ class TestPOSIXFile(TestPOSIXTool): entry = reset() ptool._get_diffs(entry, content=ondisk) self.assertFalse(mock_open.called) - mock_diff.assert_called_with(ondisk, entry.text, difflib.ndiff, + ptool._diff.assert_called_with(ondisk, entry.text, difflib.ndiff, filename=entry.get("name")) self.assertIsNone(entry.get("qtext")) self.assertEqual(entry.get("current_bdiff"), - b64encode("\n".join(mock_diff.return_value))) + b64encode("\n".join(ptool._diff.return_value))) del entry.attrib["current_bdiff"] self.assertItemsEqual(orig_entry.attrib, entry.attrib) @@ -292,7 +288,7 @@ class TestPOSIXFile(TestPOSIXTool): ptool._get_diffs(entry, interactive=True) mock_open.assert_called_with(entry.get("name")) mock_open.return_value.read.assert_any_call() - self.assertItemsEqual(mock_diff.call_args_list, + self.assertItemsEqual(ptool._diff.call_args_list, [call(ondisk, entry.text, difflib.unified_diff, filename=entry.get("name")), call(ondisk, entry.text, difflib.ndiff, @@ -300,7 +296,7 @@ class TestPOSIXFile(TestPOSIXTool): self.assertIsNotNone(entry.get("qtext")) self.assertTrue(entry.get("qtext").startswith("test\n")) self.assertEqual(entry.get("current_bdiff"), - b64encode("\n".join(mock_diff.return_value))) + b64encode("\n".join(ptool._diff.return_value))) del entry.attrib['qtext'] del entry.attrib["current_bdiff"] self.assertItemsEqual(orig_entry.attrib, entry.attrib) @@ -308,118 +304,118 @@ class TestPOSIXFile(TestPOSIXTool): # non-sensitive, interactive with unicode data entry = reset() entry.text = u("tëst") - encoded = entry.text.encode(setup['encoding']) - mock_diff.return_value = ["-test2", "+tëst"] - mock_get_data.return_value = (encoded, False) + encoded = entry.text.encode(ptool.setup['encoding']) + ptool._diff.return_value = ["-test2", "+tëst"] + ptool._get_data.return_value = (encoded, False) ptool._get_diffs(entry, interactive=True) mock_open.assert_called_with(entry.get("name")) mock_open.return_value.read.assert_any_call() - self.assertItemsEqual(mock_diff.call_args_list, + self.assertItemsEqual(ptool._diff.call_args_list, [call(ondisk, encoded, difflib.unified_diff, filename=entry.get("name")), call(ondisk, encoded, difflib.ndiff, filename=entry.get("name"))]) self.assertIsNotNone(entry.get("qtext")) self.assertEqual(entry.get("current_bdiff"), - b64encode("\n".join(mock_diff.return_value))) + b64encode("\n".join(ptool._diff.return_value))) del entry.attrib['qtext'] del entry.attrib["current_bdiff"] self.assertItemsEqual(orig_entry.attrib, entry.attrib) @patch("os.path.exists") @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.install") - @patch("Bcfg2.Client.Tools.POSIX.File.%s._makedirs" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.File.%s._set_perms" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.File.%s._write_tmpfile" % - test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.File.%s._rename_tmpfile" % - test_obj.__name__) - def test_install(self, mock_rename, mock_write, mock_set_perms, - mock_makedirs, mock_install, mock_exists): + def test_install(self, mock_install, mock_exists): + ptool = self.get_obj() + ptool._makedirs = Mock() + ptool._set_perms = Mock() + ptool._write_tmpfile = Mock() + ptool._rename_tmpfile = Mock() entry = lxml.etree.Element("Path", name="/test", type="file", mode='0644', owner='root', group='root') def reset(): - mock_rename.reset_mock() - mock_write.reset_mock() - mock_set_perms.reset_mock() - mock_makedirs.reset_mock() + ptool._rename_tmpfile.reset_mock() + ptool._write_tmpfile.reset_mock() + ptool._set_perms.reset_mock() + ptool._makedirs.reset_mock() mock_install.reset_mock() mock_exists.reset_mock() mock_exists.return_value = False - mock_makedirs.return_value = False - self.assertFalse(self.ptool.install(entry)) + ptool._makedirs.return_value = False + self.assertFalse(ptool.install(entry)) mock_exists.assert_called_with("/") - mock_makedirs.assert_called_with(entry, path="/") - + ptool._makedirs.assert_called_with(entry, path="/") + reset() - mock_makedirs.return_value = True - mock_write.return_value = False - self.assertFalse(self.ptool.install(entry)) + ptool._makedirs.return_value = True + ptool._write_tmpfile.return_value = False + self.assertFalse(ptool.install(entry)) mock_exists.assert_called_with("/") - mock_makedirs.assert_called_with(entry, path="/") - mock_write.assert_called_with(entry) + ptool._makedirs.assert_called_with(entry, path="/") + ptool._write_tmpfile.assert_called_with(entry) reset() newfile = '/test.X987yS' - mock_write.return_value = newfile - mock_set_perms.return_value = False - mock_rename.return_value = False - self.assertFalse(self.ptool.install(entry)) + ptool._write_tmpfile.return_value = newfile + ptool._set_perms.return_value = False + ptool._rename_tmpfile.return_value = False + self.assertFalse(ptool.install(entry)) mock_exists.assert_called_with("/") - mock_makedirs.assert_called_with(entry, path="/") - mock_write.assert_called_with(entry) - mock_set_perms.assert_called_with(entry, path=newfile) - mock_rename.assert_called_with(newfile, entry) + ptool._makedirs.assert_called_with(entry, path="/") + ptool._write_tmpfile.assert_called_with(entry) + ptool._set_perms.assert_called_with(entry, path=newfile) + ptool._rename_tmpfile.assert_called_with(newfile, entry) reset() - mock_rename.return_value = True + ptool._rename_tmpfile.return_value = True mock_install.return_value = False - self.assertFalse(self.ptool.install(entry)) + self.assertFalse(ptool.install(entry)) mock_exists.assert_called_with("/") - mock_makedirs.assert_called_with(entry, path="/") - mock_write.assert_called_with(entry) - mock_set_perms.assert_called_with(entry, path=newfile) - mock_rename.assert_called_with(newfile, entry) - mock_install.assert_called_with(self.ptool, entry) + ptool._makedirs.assert_called_with(entry, path="/") + ptool._write_tmpfile.assert_called_with(entry) + ptool._set_perms.assert_called_with(entry, path=newfile) + ptool._rename_tmpfile.assert_called_with(newfile, entry) + mock_install.assert_called_with(ptool, entry) reset() mock_install.return_value = True - self.assertFalse(self.ptool.install(entry)) + self.assertFalse(ptool.install(entry)) mock_exists.assert_called_with("/") - mock_makedirs.assert_called_with(entry, path="/") - mock_write.assert_called_with(entry) - mock_set_perms.assert_called_with(entry, path=newfile) - mock_rename.assert_called_with(newfile, entry) - mock_install.assert_called_with(self.ptool, entry) + ptool._makedirs.assert_called_with(entry, path="/") + ptool._write_tmpfile.assert_called_with(entry) + ptool._set_perms.assert_called_with(entry, path=newfile) + ptool._rename_tmpfile.assert_called_with(newfile, entry) + mock_install.assert_called_with(ptool, entry) reset() - mock_set_perms.return_value = True - self.assertTrue(self.ptool.install(entry)) + ptool._set_perms.return_value = True + self.assertTrue(ptool.install(entry)) mock_exists.assert_called_with("/") - mock_makedirs.assert_called_with(entry, path="/") - mock_write.assert_called_with(entry) - mock_set_perms.assert_called_with(entry, path=newfile) - mock_rename.assert_called_with(newfile, entry) - mock_install.assert_called_with(self.ptool, entry) + ptool._makedirs.assert_called_with(entry, path="/") + ptool._write_tmpfile.assert_called_with(entry) + ptool._set_perms.assert_called_with(entry, path=newfile) + ptool._rename_tmpfile.assert_called_with(newfile, entry) + mock_install.assert_called_with(ptool, entry) reset() mock_exists.return_value = True - self.assertTrue(self.ptool.install(entry)) + self.assertTrue(ptool.install(entry)) mock_exists.assert_called_with("/") - self.assertFalse(mock_makedirs.called) - mock_write.assert_called_with(entry) - mock_set_perms.assert_called_with(entry, path=newfile) - mock_rename.assert_called_with(newfile, entry) - mock_install.assert_called_with(self.ptool, entry) + self.assertFalse(ptool._makedirs.called) + ptool._write_tmpfile.assert_called_with(entry) + ptool._set_perms.assert_called_with(entry, path=newfile) + ptool._rename_tmpfile.assert_called_with(newfile, entry) + mock_install.assert_called_with(ptool, entry) @patch("time.time") def test_diff(self, mock_time): + ptool = self.get_obj() content1 = "line1\nline2" content2 = "line3" self.now = 1345640723 + def time_rv(): self.now += 1 return self.now @@ -428,7 +424,7 @@ class TestPOSIXFile(TestPOSIXTool): rv = ["line1", "line2", "line3"] func = Mock() func.return_value = rv - self.assertItemsEqual(self.ptool._diff(content1, content2, func), rv) + self.assertItemsEqual(ptool._diff(content1, content2, func), rv) func.assert_called_with(["line1", "line2"], ["line3"]) func.reset_mock() @@ -442,5 +438,5 @@ class TestPOSIXFile(TestPOSIXTool): for i in range(1, 10): yield "line%s" % i func.side_effect = slow_diff - self.assertFalse(self.ptool._diff(content1, content2, func), rv) + self.assertFalse(ptool._diff(content1, content2, func), rv) func.assert_called_with(["line1", "line2"], ["line3"]) diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestHardlink.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestHardlink.py index c38e86aeb..3159b69df 100644 --- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestHardlink.py +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestHardlink.py @@ -14,70 +14,26 @@ while path != "/": if os.path.basename(path) == "testsuite": break path = os.path.dirname(path) -from Test__init import get_posix_object -from Testbase import TestPOSIXTool +from Testbase import TestPOSIXLinkTool from common import * -class TestPOSIXHardlink(TestPOSIXTool): + +class TestPOSIXHardlink(TestPOSIXLinkTool): test_obj = POSIXHardlink @patch("os.path.samefile") - @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.verify") - def test_verify(self, mock_verify, mock_samefile): + def test__verify(self, mock_samefile): entry = lxml.etree.Element("Path", name="/test", type="hardlink", to="/dest") ptool = self.get_obj() - - mock_samefile.return_value = True - mock_verify.return_value = False - self.assertFalse(ptool.verify(entry, [])) - mock_samefile.assert_called_with(entry.get("name"), - entry.get("to")) - mock_verify.assert_called_with(ptool, entry, []) - - mock_samefile.reset_mock() - mock_verify.reset_mock() - mock_verify.return_value = True - self.assertTrue(ptool.verify(entry, [])) - mock_samefile.assert_called_with(entry.get("name"), - entry.get("to")) - mock_verify.assert_called_with(ptool, entry, []) - - mock_samefile.reset_mock() - mock_verify.reset_mock() - mock_samefile.return_value = False - self.assertFalse(ptool.verify(entry, [])) - mock_samefile.assert_called_with(entry.get("name"), - entry.get("to")) - mock_verify.assert_called_with(ptool, entry, []) - - mock_samefile.reset_mock() - mock_verify.reset_mock() - mock_samefile.side_effect = OSError - self.assertFalse(ptool.verify(entry, [])) - mock_samefile.assert_called_with(entry.get("name"), - entry.get("to")) + self.assertEqual(ptool._verify(entry), mock_samefile.return_value) + self.assertItemsEqual(mock_samefile.call_args[0], + [entry.get("name"), entry.get("to")]) @patch("os.link") - @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.install") - @patch("Bcfg2.Client.Tools.POSIX.Hardlink.%s._exists" % test_obj.__name__) - def test_install(self, mock_exists, mock_install, mock_link): + def test__link(self, mock_link): entry = lxml.etree.Element("Path", name="/test", type="hardlink", to="/dest") ptool = self.get_obj() - - mock_exists.return_value = False - mock_install.return_value = True - self.assertTrue(ptool.install(entry)) - mock_exists.assert_called_with(entry, remove=True) - mock_link.assert_called_with(entry.get("to"), entry.get("name")) - mock_install.assert_called_with(ptool, entry) - - mock_link.reset_mock() - mock_exists.reset_mock() - mock_install.reset_mock() - mock_link.side_effect = OSError - self.assertFalse(ptool.install(entry)) - mock_exists.assert_called_with(entry, remove=True) + self.assertEqual(ptool._link(entry), mock_link.return_value) mock_link.assert_called_with(entry.get("to"), entry.get("name")) - mock_install.assert_called_with(ptool, entry) diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestNonexistent.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestNonexistent.py index 583d17e32..3d1ddbf84 100644 --- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestNonexistent.py +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestNonexistent.py @@ -14,7 +14,7 @@ while path != "/": if os.path.basename(path) == "testsuite": break path = os.path.dirname(path) -from Test__init import get_config, get_posix_object +from Test__init import get_config from Testbase import TestPOSIXTool from common import * @@ -24,34 +24,36 @@ class TestPOSIXNonexistent(TestPOSIXTool): @patch("os.path.lexists") def test_verify(self, mock_lexists): + ptool = self.get_obj() entry = lxml.etree.Element("Path", name="/test", type="nonexistent") for val in [True, False]: mock_lexists.reset_mock() mock_lexists.return_value = val - self.assertEqual(self.ptool.verify(entry, []), not val) + self.assertEqual(ptool.verify(entry, []), not val) mock_lexists.assert_called_with(entry.get("name")) def test_install(self): entry = lxml.etree.Element("Path", name="/test", type="nonexistent") - self.ptool._remove = Mock() + ptool = self.get_obj() + ptool._remove = Mock() def reset(): - self.ptool._remove.reset_mock() + ptool._remove.reset_mock() - self.assertTrue(self.ptool.install(entry)) - self.ptool._remove.assert_called_with(entry, recursive=False) + self.assertTrue(ptool.install(entry)) + ptool._remove.assert_called_with(entry, recursive=False) reset() entry.set("recursive", "true") - self.assertTrue(self.ptool.install(entry)) - self.ptool._remove.assert_called_with(entry, recursive=True) + self.assertTrue(ptool.install(entry)) + ptool._remove.assert_called_with(entry, recursive=True) reset() child_entry = lxml.etree.Element("Path", name="/test/foo", type="nonexistent") - ptool = self.get_obj(posix=get_posix_object(config=get_config([child_entry]))) + ptool = self.get_obj(config=get_config([child_entry])) ptool._remove = Mock() self.assertTrue(ptool.install(entry)) ptool._remove.assert_called_with(entry, recursive=True) @@ -59,13 +61,13 @@ class TestPOSIXNonexistent(TestPOSIXTool): reset() child_entry = lxml.etree.Element("Path", name="/test/foo", type="file") - ptool = self.get_obj(posix=get_posix_object(config=get_config([child_entry]))) + ptool = self.get_obj(config=get_config([child_entry])) ptool._remove = Mock() self.assertFalse(ptool.install(entry)) self.assertFalse(ptool._remove.called) reset() entry.set("recursive", "false") - self.ptool._remove.side_effect = OSError - self.assertFalse(self.ptool.install(entry)) - self.ptool._remove.assert_called_with(entry, recursive=False) + ptool._remove.side_effect = OSError + self.assertFalse(ptool.install(entry)) + ptool._remove.assert_called_with(entry, recursive=False) diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestPermissions.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestPermissions.py index 565857437..10194dbbf 100644 --- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestPermissions.py +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestPermissions.py @@ -1,5 +1,6 @@ from Bcfg2.Client.Tools.POSIX.Permissions import * from Testbase import TestPOSIXTool + class TestPOSIXPermissions(TestPOSIXTool): test_obj = POSIXPermissions diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestSymlink.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestSymlink.py index 4c8ddfa3f..78bfd716f 100644 --- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestSymlink.py +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestSymlink.py @@ -1,6 +1,5 @@ import os import sys -import copy import lxml.etree from mock import Mock, MagicMock, patch from Bcfg2.Client.Tools.POSIX.Symlink import * @@ -14,84 +13,33 @@ while path != "/": if os.path.basename(path) == "testsuite": break path = os.path.dirname(path) -from Test__init import get_posix_object -from Testbase import TestPOSIXTool +from Testbase import TestPOSIXLinkTool from common import * -class TestPOSIXSymlink(TestPOSIXTool): + +class TestPOSIXSymlink(TestPOSIXLinkTool): test_obj = POSIXSymlink @patch("os.readlink") - @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.verify") - def test_verify(self, mock_verify, mock_readlink): + def test__verify(self, mock_readlink): entry = lxml.etree.Element("Path", name="/test", type="symlink", to="/dest") ptool = self.get_obj() mock_readlink.return_value = entry.get("to") - mock_verify.return_value = False - self.assertFalse(ptool.verify(entry, [])) + self.assertTrue(ptool._verify(entry)) mock_readlink.assert_called_with(entry.get("name")) - mock_verify.assert_called_with(ptool, entry, []) mock_readlink.reset_mock() - mock_verify.reset_mock() - mock_verify.return_value = True - self.assertTrue(ptool.verify(entry, [])) - mock_readlink.assert_called_with(entry.get("name")) - mock_verify.assert_called_with(ptool, entry, []) - - mock_readlink.reset_mock() - mock_verify.reset_mock() mock_readlink.return_value = "/bogus" - self.assertFalse(ptool.verify(entry, [])) - mock_readlink.assert_called_with(entry.get("name")) - mock_verify.assert_called_with(ptool, entry, []) - - # relative symlink - mock_readlink.reset_mock() - mock_verify.reset_mock() - entry = lxml.etree.Element("Path", name="/test", type="symlink", - to="dest") - mock_readlink.return_value = entry.get("to") - self.assertTrue(ptool.verify(entry, [])) - mock_readlink.assert_called_with(entry.get("name")) - mock_verify.assert_called_with(ptool, entry, []) - - mock_readlink.reset_mock() - mock_verify.reset_mock() - mock_readlink.side_effect = OSError - self.assertFalse(ptool.verify(entry, [])) + self.assertFalse(ptool._verify(entry)) mock_readlink.assert_called_with(entry.get("name")) @patch("os.symlink") - @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.install") - @patch("Bcfg2.Client.Tools.POSIX.Symlink.%s._exists" % test_obj.__name__) - def test_install(self, mock_exists, mock_install, mock_symlink): + def test__link(self, mock_symlink): entry = lxml.etree.Element("Path", name="/test", type="symlink", to="/dest") ptool = self.get_obj() - - mock_exists.return_value = False - mock_install.return_value = True - self.assertTrue(ptool.install(entry)) - mock_exists.assert_called_with(entry, remove=True) - mock_symlink.assert_called_with(entry.get("to"), entry.get("name")) - mock_install.assert_called_with(ptool, entry) - - # relative symlink - entry = lxml.etree.Element("Path", name="/test", type="symlink", - to="dest") - self.assertTrue(ptool.install(entry)) - mock_exists.assert_called_with(entry, remove=True) - mock_symlink.assert_called_with(entry.get("to"), entry.get("name")) - mock_install.assert_called_with(ptool, entry) - - mock_symlink.reset_mock() - mock_exists.reset_mock() - mock_install.reset_mock() - mock_symlink.side_effect = OSError - self.assertFalse(ptool.install(entry)) - mock_exists.assert_called_with(entry, remove=True) + self.assertEqual(ptool._link(entry), + mock_symlink.return_value) mock_symlink.assert_called_with(entry.get("to"), entry.get("name")) - mock_install.assert_called_with(ptool, entry) diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/Test__init.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/Test__init.py index 4048be7ca..f01082e86 100644 --- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/Test__init.py +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/Test__init.py @@ -3,7 +3,7 @@ import sys import lxml.etree from mock import Mock, MagicMock, patch import Bcfg2.Client.Tools -import Bcfg2.Client.Tools.POSIX +from Bcfg2.Client.Tools.POSIX import * # add all parent testsuite directories to sys.path to allow (most) # relative imports in python 2.4 @@ -15,6 +15,7 @@ while path != "/": break path = os.path.dirname(path) from common import * +from TestTools.Test_init import TestTool def get_config(entries): @@ -24,35 +25,14 @@ def get_config(entries): return config -def get_posix_object(logger=None, setup=None, config=None): - if config is None: - config = lxml.etree.Element("Configuration") - if not logger: - def print_msg(msg): - print(msg) - logger = Mock() - logger.error = Mock(side_effect=print_msg) - logger.warning = Mock(side_effect=print_msg) - logger.info = Mock(side_effect=print_msg) - logger.debug = Mock(side_effect=print_msg) - if not setup: - setup = MagicMock() - return Bcfg2.Client.Tools.POSIX.POSIX(logger, setup, config) - - -class TestPOSIX(Bcfg2TestCase): - def setUp(self): - self.posix = get_posix_object() - - def tearDown(self): - # just to guarantee that we start fresh each time - self.posix = None +class TestPOSIX(TestTool): + test_obj = POSIX def test__init(self): entries = [lxml.etree.Element("Path", name="test", type="file")] - posix = get_posix_object(config=get_config(entries)) + posix = self.get_obj(config=get_config(entries)) self.assertIsInstance(posix, Bcfg2.Client.Tools.Tool) - self.assertIsInstance(posix, Bcfg2.Client.Tools.POSIX.POSIX) + self.assertIsInstance(posix, POSIX) self.assertIn('Path', posix.__req__) self.assertGreater(len(posix.__req__['Path']), 0) self.assertGreater(len(posix.__handles__), 0) @@ -60,98 +40,102 @@ class TestPOSIX(Bcfg2TestCase): @patch("Bcfg2.Client.Tools.Tool.canVerify") def test_canVerify(self, mock_canVerify): + posix = self.get_obj() entry = lxml.etree.Element("Path", name="test", type="file") # first, test superclass canVerify failure mock_canVerify.return_value = False - self.assertFalse(self.posix.canVerify(entry)) - mock_canVerify.assert_called_with(self.posix, entry) + self.assertFalse(posix.canVerify(entry)) + mock_canVerify.assert_called_with(posix, entry) # next, test fully_specified failure - self.posix.logger.error.reset_mock() + posix.logger.error.reset_mock() mock_canVerify.reset_mock() mock_canVerify.return_value = True mock_fully_spec = Mock() mock_fully_spec.return_value = False - self.posix._handlers[entry.get("type")].fully_specified = \ + posix._handlers[entry.get("type")].fully_specified = \ mock_fully_spec - self.assertFalse(self.posix.canVerify(entry)) - mock_canVerify.assert_called_with(self.posix, entry) + self.assertFalse(posix.canVerify(entry)) + mock_canVerify.assert_called_with(posix, entry) mock_fully_spec.assert_called_with(entry) - self.assertTrue(self.posix.logger.error.called) + self.assertTrue(posix.logger.error.called) # finally, test success - self.posix.logger.error.reset_mock() + posix.logger.error.reset_mock() mock_canVerify.reset_mock() mock_fully_spec.reset_mock() mock_fully_spec.return_value = True - self.assertTrue(self.posix.canVerify(entry)) - mock_canVerify.assert_called_with(self.posix, entry) + self.assertTrue(posix.canVerify(entry)) + mock_canVerify.assert_called_with(posix, entry) mock_fully_spec.assert_called_with(entry) - self.assertFalse(self.posix.logger.error.called) + self.assertFalse(posix.logger.error.called) @patch("Bcfg2.Client.Tools.Tool.canInstall") def test_canInstall(self, mock_canInstall): + posix = self.get_obj() entry = lxml.etree.Element("Path", name="test", type="file") # first, test superclass canInstall failure mock_canInstall.return_value = False - self.assertFalse(self.posix.canInstall(entry)) - mock_canInstall.assert_called_with(self.posix, entry) + self.assertFalse(posix.canInstall(entry)) + mock_canInstall.assert_called_with(posix, entry) # next, test fully_specified failure - self.posix.logger.error.reset_mock() + posix.logger.error.reset_mock() mock_canInstall.reset_mock() mock_canInstall.return_value = True mock_fully_spec = Mock() mock_fully_spec.return_value = False - self.posix._handlers[entry.get("type")].fully_specified = \ + posix._handlers[entry.get("type")].fully_specified = \ mock_fully_spec - self.assertFalse(self.posix.canInstall(entry)) - mock_canInstall.assert_called_with(self.posix, entry) + self.assertFalse(posix.canInstall(entry)) + mock_canInstall.assert_called_with(posix, entry) mock_fully_spec.assert_called_with(entry) - self.assertTrue(self.posix.logger.error.called) + self.assertTrue(posix.logger.error.called) # finally, test success - self.posix.logger.error.reset_mock() + posix.logger.error.reset_mock() mock_canInstall.reset_mock() mock_fully_spec.reset_mock() mock_fully_spec.return_value = True - self.assertTrue(self.posix.canInstall(entry)) - mock_canInstall.assert_called_with(self.posix, entry) + self.assertTrue(posix.canInstall(entry)) + mock_canInstall.assert_called_with(posix, entry) mock_fully_spec.assert_called_with(entry) - self.assertFalse(self.posix.logger.error.called) + self.assertFalse(posix.logger.error.called) def test_InstallPath(self): + posix = self.get_obj() entry = lxml.etree.Element("Path", name="test", type="file") mock_install = Mock() mock_install.return_value = True - self.posix._handlers[entry.get("type")].install = mock_install - self.assertTrue(self.posix.InstallPath(entry)) + posix._handlers[entry.get("type")].install = mock_install + self.assertTrue(posix.InstallPath(entry)) mock_install.assert_called_with(entry) def test_VerifyPath(self): + posix = self.get_obj() entry = lxml.etree.Element("Path", name="test", type="file") modlist = [] mock_verify = Mock() mock_verify.return_value = True - self.posix._handlers[entry.get("type")].verify = mock_verify - self.assertTrue(self.posix.VerifyPath(entry, modlist)) + posix._handlers[entry.get("type")].verify = mock_verify + self.assertTrue(posix.VerifyPath(entry, modlist)) mock_verify.assert_called_with(entry, modlist) mock_verify.reset_mock() mock_verify.return_value = False - self.posix.setup.__getitem__.return_value = True - self.assertFalse(self.posix.VerifyPath(entry, modlist)) + posix.setup.__getitem__.return_value = True + self.assertFalse(posix.VerifyPath(entry, modlist)) self.assertIsNotNone(entry.get('qtext')) @patch('os.remove') def test_prune_old_backups(self, mock_remove): entry = lxml.etree.Element("Path", name="/etc/foo", type="file") setup = dict(ppath='/', max_copies=5, paranoid=True) - posix = get_posix_object(setup=setup) + posix = self.get_obj(setup=setup) remove = ["_etc_foo_2012-07-20T04:13:22.364989", "_etc_foo_2012-07-31T04:13:23.894958", @@ -200,48 +184,55 @@ class TestPOSIX(Bcfg2TestCase): @patch("shutil.copy") @patch("os.path.isdir") - @patch("Bcfg2.Client.Tools.POSIX.POSIX._prune_old_backups") - def test_paranoid_backup(self, mock_prune, mock_isdir, mock_copy): + def test_paranoid_backup(self, mock_isdir, mock_copy): entry = lxml.etree.Element("Path", name="/etc/foo", type="file") setup = dict(ppath='/', max_copies=5, paranoid=False) - posix = get_posix_object(setup=setup) + posix = self.get_obj(setup=setup) + posix._prune_old_backups = Mock() # paranoid false globally posix._paranoid_backup(entry) - self.assertFalse(mock_prune.called) + self.assertFalse(posix._prune_old_backups.called) self.assertFalse(mock_copy.called) # paranoid false on the entry - mock_prune.reset_mock() setup['paranoid'] = True - posix = get_posix_object(setup=setup) + posix = self.get_obj(setup=setup) + posix._prune_old_backups = Mock() + + def reset(): + mock_isdir.reset_mock() + mock_copy.reset_mock() + posix._prune_old_backups.reset_mock() + + reset() posix._paranoid_backup(entry) - self.assertFalse(mock_prune.called) + self.assertFalse(posix._prune_old_backups.called) self.assertFalse(mock_copy.called) # entry does not exist on filesystem - mock_prune.reset_mock() + reset() entry.set("paranoid", "true") entry.set("current_exists", "false") posix._paranoid_backup(entry) - self.assertFalse(mock_prune.called) + self.assertFalse(posix._prune_old_backups.called) self.assertFalse(mock_copy.called) # entry is a directory on the filesystem - mock_prune.reset_mock() + reset() entry.set("current_exists", "true") mock_isdir.return_value = True posix._paranoid_backup(entry) - self.assertFalse(mock_prune.called) + self.assertFalse(posix._prune_old_backups.called) self.assertFalse(mock_copy.called) mock_isdir.assert_called_with(entry.get("name")) # test the actual backup now - mock_prune.reset_mock() + reset() mock_isdir.return_value = False posix._paranoid_backup(entry) mock_isdir.assert_called_with(entry.get("name")) - mock_prune.assert_called_with(entry) + posix._prune_old_backups.assert_called_with(entry) # it's basically impossible to test the shutil.copy() call # exactly because the destination includes microseconds, so we # just test it good enough diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/Testbase.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/Testbase.py index b3599db83..49e9be2ba 100644 --- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/Testbase.py +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/Testbase.py @@ -16,7 +16,7 @@ while path != "/": if os.path.basename(path) == "testsuite": break path = os.path.dirname(path) -from Test__init import get_posix_object +from TestTools.Test_init import TestTool from common import * try: @@ -32,96 +32,86 @@ except ImportError: HAS_ACLS = False -class TestPOSIXTool(Bcfg2TestCase): +class TestPOSIXTool(TestTool): test_obj = POSIXTool - def get_obj(self, posix=None): - if posix is None: - posix = get_posix_object() - return self.test_obj(posix.logger, posix.setup, posix.config) - - def setUp(self): - self.ptool = self.get_obj() - - def tearDown(self): - # just to guarantee that we start fresh each time - self.ptool = None - def test_fully_specified(self): # fully_specified should do no checking on the abstract # POSIXTool object - self.assertTrue(self.ptool.fully_specified(Mock())) + ptool = self.get_obj() + self.assertTrue(ptool.fully_specified(Mock())) @patch('os.stat') @patch('os.walk') - @patch("Bcfg2.Client.Tools.POSIX.base.%s._verify_metadata" % - test_obj.__name__) - def test_verify(self, mock_verify, mock_walk, mock_stat): + def test_verify(self, mock_walk, mock_stat): + ptool = self.get_obj() + ptool._verify_metadata = Mock() entry = lxml.etree.Element("Path", name="/test", type="file") mock_stat.return_value = MagicMock() - mock_verify.return_value = False - self.assertFalse(self.ptool.verify(entry, [])) - mock_verify.assert_called_with(entry) + ptool._verify_metadata.return_value = False + self.assertFalse(ptool.verify(entry, [])) + ptool._verify_metadata.assert_called_with(entry) - mock_verify.reset_mock() - mock_verify.return_value = True - self.assertTrue(self.ptool.verify(entry, [])) - mock_verify.assert_called_with(entry) + ptool._verify_metadata.reset_mock() + ptool._verify_metadata.return_value = True + self.assertTrue(ptool.verify(entry, [])) + ptool._verify_metadata.assert_called_with(entry) - mock_verify.reset_mock() + ptool._verify_metadata.reset_mock() entry.set("recursive", "true") walk_rv = [("/", ["dir1", "dir2"], ["file1", "file2"]), ("/dir1", ["dir3"], []), ("/dir2", [], ["file3", "file4"])] mock_walk.return_value = walk_rv - self.assertTrue(self.ptool.verify(entry, [])) + self.assertTrue(ptool.verify(entry, [])) mock_walk.assert_called_with(entry.get("name")) all_verifies = [call(entry)] for root, dirs, files in walk_rv: all_verifies.extend([call(entry, path=os.path.join(root, p)) for p in dirs + files]) - self.assertItemsEqual(mock_verify.call_args_list, all_verifies) + self.assertItemsEqual(ptool._verify_metadata.call_args_list, all_verifies) @patch('os.walk') - @patch("Bcfg2.Client.Tools.POSIX.base.%s._set_perms" % test_obj.__name__) - def test_install(self, mock_set_perms, mock_walk): + def test_install(self, mock_walk): + ptool = self.get_obj() + ptool._set_perms = Mock() entry = lxml.etree.Element("Path", name="/test", type="file") - mock_set_perms.return_value = True - self.assertTrue(self.ptool.install(entry)) - mock_set_perms.assert_called_with(entry) + ptool._set_perms.return_value = True + self.assertTrue(ptool.install(entry)) + ptool._set_perms.assert_called_with(entry) - mock_set_perms.reset_mock() + ptool._set_perms.reset_mock() entry.set("recursive", "true") walk_rv = [("/", ["dir1", "dir2"], ["file1", "file2"]), ("/dir1", ["dir3"], []), ("/dir2", [], ["file3", "file4"])] mock_walk.return_value = walk_rv - mock_set_perms.return_value = True - self.assertTrue(self.ptool.install(entry)) + ptool._set_perms.return_value = True + self.assertTrue(ptool.install(entry)) mock_walk.assert_called_with(entry.get("name")) all_set_perms = [call(entry)] for root, dirs, files in walk_rv: all_set_perms.extend([call(entry, path=os.path.join(root, p)) for p in dirs + files]) - self.assertItemsEqual(mock_set_perms.call_args_list, + self.assertItemsEqual(ptool._set_perms.call_args_list, all_set_perms) mock_walk.reset_mock() - mock_set_perms.reset_mock() + ptool._set_perms.reset_mock() def set_perms_rv(entry, path=None): if path == '/dir2/file3': return False else: return True - mock_set_perms.side_effect = set_perms_rv + ptool._set_perms.side_effect = set_perms_rv - self.assertFalse(self.ptool.install(entry)) + self.assertFalse(ptool.install(entry)) mock_walk.assert_called_with(entry.get("name")) - self.assertItemsEqual(mock_set_perms.call_args_list, all_set_perms) + self.assertItemsEqual(ptool._set_perms.call_args_list, all_set_perms) @patch('os.rmdir') @patch('os.unlink') @@ -130,6 +120,7 @@ class TestPOSIXTool(Bcfg2TestCase): @patch('os.path.islink') def test_remove(self, mock_islink, mock_isdir, mock_rmtree, mock_unlink, mock_rmdir): + ptool = self.get_obj() entry = lxml.etree.Element("Path", name="/etc/foo") def reset(): @@ -141,7 +132,7 @@ class TestPOSIXTool(Bcfg2TestCase): mock_islink.return_value = True mock_isdir.return_value = False - self.ptool._remove(entry) + ptool._remove(entry) mock_unlink.assert_called_with(entry.get('name')) self.assertFalse(mock_rmtree.called) self.assertFalse(mock_rmdir.called) @@ -149,13 +140,13 @@ class TestPOSIXTool(Bcfg2TestCase): reset() mock_islink.return_value = False mock_isdir.return_value = True - self.ptool._remove(entry) + ptool._remove(entry) mock_rmtree.assert_called_with(entry.get('name')) self.assertFalse(mock_unlink.called) self.assertFalse(mock_rmdir.called) reset() - self.ptool._remove(entry, recursive=False) + ptool._remove(entry, recursive=False) mock_rmdir.assert_called_with(entry.get('name')) self.assertFalse(mock_unlink.called) self.assertFalse(mock_rmtree.called) @@ -163,7 +154,7 @@ class TestPOSIXTool(Bcfg2TestCase): reset() mock_islink.return_value = False mock_isdir.return_value = False - self.ptool._remove(entry, recursive=False) + ptool._remove(entry, recursive=False) mock_unlink.assert_called_with(entry.get('name')) self.assertFalse(mock_rmtree.called) self.assertFalse(mock_rmdir.called) @@ -172,103 +163,101 @@ class TestPOSIXTool(Bcfg2TestCase): def test_exists(self, mock_lstat): entry = lxml.etree.Element("Path", name="/etc/foo", type="file") - self.ptool._remove = Mock() + ptool = self.get_obj() + ptool._remove = Mock() def reset(): mock_lstat.reset_mock() - self.ptool._remove.reset_mock() + ptool._remove.reset_mock() mock_lstat.side_effect = OSError - self.assertFalse(self.ptool._exists(entry)) + self.assertFalse(ptool._exists(entry)) mock_lstat.assert_called_with(entry.get('name')) - self.assertFalse(self.ptool._remove.called) + self.assertFalse(ptool._remove.called) reset() rv = MagicMock() mock_lstat.return_value = rv mock_lstat.side_effect = None - self.assertEqual(self.ptool._exists(entry), rv) + self.assertEqual(ptool._exists(entry), rv) mock_lstat.assert_called_with(entry.get('name')) - self.assertFalse(self.ptool._remove.called) + self.assertFalse(ptool._remove.called) reset() - self.assertEqual(self.ptool._exists(entry, remove=True), None) + self.assertEqual(ptool._exists(entry, remove=True), None) mock_lstat.assert_called_with(entry.get('name')) - self.ptool._remove.assert_called_with(entry) + ptool._remove.assert_called_with(entry) reset() - self.ptool._remove.side_effect = OSError - self.assertEqual(self.ptool._exists(entry, remove=True), rv) + ptool._remove.side_effect = OSError + self.assertEqual(ptool._exists(entry, remove=True), rv) mock_lstat.assert_called_with(entry.get('name')) - self.ptool._remove.assert_called_with(entry) + ptool._remove.assert_called_with(entry) @patch("os.chown") @patch("os.chmod") @patch("os.utime") - @patch("Bcfg2.Client.Tools.POSIX.base.%s._norm_entry_uid" % - test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.base.%s._norm_entry_gid" % - test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.base.%s._set_acls" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.base.%s._set_secontext" % - test_obj.__name__) - def test_set_perms(self, mock_set_secontext, mock_set_acls, mock_norm_gid, - mock_norm_uid, mock_utime, mock_chmod, mock_chown): + def test_set_perms(self, mock_utime, mock_chmod, mock_chown): + ptool = self.get_obj() + ptool._norm_entry_uid = Mock() + ptool._norm_entry_gid = Mock() + ptool._set_acls = Mock() + ptool._set_secontext = Mock() def reset(): - mock_set_secontext.reset_mock() - mock_set_acls.reset_mock() - mock_norm_gid.reset_mock() - mock_norm_uid.reset_mock() + ptool._set_secontext.reset_mock() + ptool._set_acls.reset_mock() + ptool._norm_entry_gid.reset_mock() + ptool._norm_entry_uid.reset_mock() mock_chmod.reset_mock() mock_chown.reset_mock() mock_utime.reset_mock() entry = lxml.etree.Element("Path", name="/etc/foo", to="/etc/bar", type="symlink") - mock_set_acls.return_value = True - mock_set_secontext.return_value = True - self.assertTrue(self.ptool._set_perms(entry)) - mock_set_secontext.assert_called_with(entry, path=entry.get("name")) - mock_set_acls.assert_called_with(entry, path=entry.get("name")) + ptool._set_acls.return_value = True + ptool._set_secontext.return_value = True + self.assertTrue(ptool._set_perms(entry)) + ptool._set_secontext.assert_called_with(entry, path=entry.get("name")) + ptool._set_acls.assert_called_with(entry, path=entry.get("name")) entry = lxml.etree.Element("Path", name="/etc/foo", owner="owner", group="group", mode="644", type="file") - mock_norm_uid.return_value = 10 - mock_norm_gid.return_value = 100 + ptool._norm_entry_uid.return_value = 10 + ptool._norm_entry_gid.return_value = 100 reset() - self.assertTrue(self.ptool._set_perms(entry)) - mock_norm_uid.assert_called_with(entry) - mock_norm_gid.assert_called_with(entry) + self.assertTrue(ptool._set_perms(entry)) + ptool._norm_entry_uid.assert_called_with(entry) + ptool._norm_entry_gid.assert_called_with(entry) mock_chown.assert_called_with(entry.get("name"), 10, 100) mock_chmod.assert_called_with(entry.get("name"), int(entry.get("mode"), 8)) self.assertFalse(mock_utime.called) - mock_set_secontext.assert_called_with(entry, path=entry.get("name")) - mock_set_acls.assert_called_with(entry, path=entry.get("name")) + ptool._set_secontext.assert_called_with(entry, path=entry.get("name")) + ptool._set_acls.assert_called_with(entry, path=entry.get("name")) reset() mtime = 1344459042 entry.set("mtime", str(mtime)) - self.assertTrue(self.ptool._set_perms(entry)) - mock_norm_uid.assert_called_with(entry) - mock_norm_gid.assert_called_with(entry) + self.assertTrue(ptool._set_perms(entry)) + ptool._norm_entry_uid.assert_called_with(entry) + ptool._norm_entry_gid.assert_called_with(entry) mock_chown.assert_called_with(entry.get("name"), 10, 100) mock_chmod.assert_called_with(entry.get("name"), int(entry.get("mode"), 8)) mock_utime.assert_called_with(entry.get("name"), (mtime, mtime)) - mock_set_secontext.assert_called_with(entry, path=entry.get("name")) - mock_set_acls.assert_called_with(entry, path=entry.get("name")) + ptool._set_secontext.assert_called_with(entry, path=entry.get("name")) + ptool._set_acls.assert_called_with(entry, path=entry.get("name")) reset() - self.assertTrue(self.ptool._set_perms(entry, path='/etc/bar')) - mock_norm_uid.assert_called_with(entry) - mock_norm_gid.assert_called_with(entry) + self.assertTrue(ptool._set_perms(entry, path='/etc/bar')) + ptool._norm_entry_uid.assert_called_with(entry) + ptool._norm_entry_gid.assert_called_with(entry) mock_chown.assert_called_with('/etc/bar', 10, 100) mock_chmod.assert_called_with('/etc/bar', int(entry.get("mode"), 8)) mock_utime.assert_called_with(entry.get("name"), (mtime, mtime)) - mock_set_secontext.assert_called_with(entry, path='/etc/bar') - mock_set_acls.assert_called_with(entry, path='/etc/bar') + ptool._set_secontext.assert_called_with(entry, path='/etc/bar') + ptool._set_acls.assert_called_with(entry, path='/etc/bar') # test dev_type modification of perms, failure of chown reset() @@ -280,15 +269,15 @@ class TestPOSIXTool(Bcfg2TestCase): os.chown.side_effect = chown_rv entry.set("type", "device") entry.set("dev_type", list(device_map.keys())[0]) - self.assertFalse(self.ptool._set_perms(entry)) - mock_norm_uid.assert_called_with(entry) - mock_norm_gid.assert_called_with(entry) + self.assertFalse(ptool._set_perms(entry)) + ptool._norm_entry_uid.assert_called_with(entry) + ptool._norm_entry_gid.assert_called_with(entry) mock_chown.assert_called_with(entry.get("name"), 0, 0) mock_chmod.assert_called_with(entry.get("name"), int(entry.get("mode"), 8) | list(device_map.values())[0]) mock_utime.assert_called_with(entry.get("name"), (mtime, mtime)) - mock_set_secontext.assert_called_with(entry, path=entry.get("name")) - mock_set_acls.assert_called_with(entry, path=entry.get("name")) + ptool._set_secontext.assert_called_with(entry, path=entry.get("name")) + ptool._set_acls.assert_called_with(entry, path=entry.get("name")) # test failure of chmod reset() @@ -296,15 +285,15 @@ class TestPOSIXTool(Bcfg2TestCase): os.chmod.side_effect = OSError entry.set("type", "file") del entry.attrib["dev_type"] - self.assertFalse(self.ptool._set_perms(entry)) - mock_norm_uid.assert_called_with(entry) - mock_norm_gid.assert_called_with(entry) + self.assertFalse(ptool._set_perms(entry)) + ptool._norm_entry_uid.assert_called_with(entry) + ptool._norm_entry_gid.assert_called_with(entry) mock_chown.assert_called_with(entry.get("name"), 10, 100) mock_chmod.assert_called_with(entry.get("name"), int(entry.get("mode"), 8)) mock_utime.assert_called_with(entry.get("name"), (mtime, mtime)) - mock_set_secontext.assert_called_with(entry, path=entry.get("name")) - mock_set_acls.assert_called_with(entry, path=entry.get("name")) + ptool._set_secontext.assert_called_with(entry, path=entry.get("name")) + ptool._set_acls.assert_called_with(entry, path=entry.get("name")) # test that even when everything fails, we try to do it all. # e.g., when chmod fails, we still try to apply acls, set @@ -312,33 +301,32 @@ class TestPOSIXTool(Bcfg2TestCase): reset() os.chown.side_effect = OSError os.utime.side_effect = OSError - mock_set_acls.return_value = False - mock_set_secontext.return_value = False - self.assertFalse(self.ptool._set_perms(entry)) - mock_norm_uid.assert_called_with(entry) - mock_norm_gid.assert_called_with(entry) + ptool._set_acls.return_value = False + ptool._set_secontext.return_value = False + self.assertFalse(ptool._set_perms(entry)) + ptool._norm_entry_uid.assert_called_with(entry) + ptool._norm_entry_gid.assert_called_with(entry) mock_chown.assert_called_with(entry.get("name"), 10, 100) mock_chmod.assert_called_with(entry.get("name"), int(entry.get("mode"), 8)) mock_utime.assert_called_with(entry.get("name"), (mtime, mtime)) - mock_set_secontext.assert_called_with(entry, path=entry.get("name")) - mock_set_acls.assert_called_with(entry, path=entry.get("name")) + ptool._set_secontext.assert_called_with(entry, path=entry.get("name")) + ptool._set_acls.assert_called_with(entry, path=entry.get("name")) @skipUnless(HAS_ACLS, "ACLS not found, skipping") @patchIf(HAS_ACLS, "posix1e.ACL") @patchIf(HAS_ACLS, "posix1e.Entry") @patch("os.path.isdir") - @patch("Bcfg2.Client.Tools.POSIX.base.%s._norm_uid" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.base.%s._norm_gid" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.base.%s._list_entry_acls" % - test_obj.__name__) - def test_set_acls(self, mock_list_entry_acls, mock_norm_gid, mock_norm_uid, - mock_isdir, mock_Entry, mock_ACL): + def test_set_acls(self, mock_isdir, mock_Entry, mock_ACL): + ptool = self.get_obj() + ptool._list_entry_acls = Mock() + ptool._norm_uid = Mock() + ptool._norm_gid = Mock() entry = lxml.etree.Element("Path", name="/etc/foo", type="file") # disable acls for the initial test Bcfg2.Client.Tools.POSIX.base.HAS_ACLS = False - self.assertTrue(self.ptool._set_acls(entry)) + self.assertTrue(ptool._set_acls(entry)) Bcfg2.Client.Tools.POSIX.base.HAS_ACLS = True # build a set of file ACLs to return from posix1e.ACL(file=...) @@ -361,9 +349,9 @@ class TestPOSIXTool(Bcfg2TestCase): # _list_entry_acls() entry_acls = {("default", posix1e.ACL_USER, "user"): 7, ("access", posix1e.ACL_GROUP, "group"): 5} - mock_list_entry_acls.return_value = entry_acls - mock_norm_uid.return_value = 10 - mock_norm_gid.return_value = 100 + ptool._list_entry_acls.return_value = entry_acls + ptool._norm_uid.return_value = 10 + ptool._norm_gid.return_value = 100 # set up the unreasonably complex return value for # posix1e.ACL(), which has three separate uses @@ -403,17 +391,17 @@ class TestPOSIXTool(Bcfg2TestCase): # test fs mounted noacl mock_ACL.side_effect = IOError(95, "Operation not permitted") - self.assertFalse(self.ptool._set_acls(entry)) + self.assertFalse(ptool._set_acls(entry)) # test other error reset() mock_ACL.side_effect = IOError - self.assertFalse(self.ptool._set_acls(entry)) + self.assertFalse(ptool._set_acls(entry)) reset() mock_ACL.side_effect = mock_acl_rv mock_isdir.return_value = True - self.assertTrue(self.ptool._set_acls(entry)) + self.assertTrue(ptool._set_acls(entry)) self.assertItemsEqual(mock_ACL.call_args_list, [call(file=entry.get("name")), call(filedef=entry.get("name"))]) @@ -421,9 +409,9 @@ class TestPOSIXTool(Bcfg2TestCase): [call(a) for a in remove_acls]) self.assertItemsEqual(filedef_rv.delete_entry.call_args_list, [call(a) for a in remove_acls]) - mock_list_entry_acls.assert_called_with(entry) - mock_norm_uid.assert_called_with("user") - mock_norm_gid.assert_called_with("group") + ptool._list_entry_acls.assert_called_with(entry) + ptool._norm_uid.assert_called_with("user") + ptool._norm_gid.assert_called_with("group") fileacl_rv.calc_mask.assert_any_call() fileacl_rv.applyto.assert_called_with(entry.get("name"), posix1e.ACL_TYPE_ACCESS) @@ -432,7 +420,7 @@ class TestPOSIXTool(Bcfg2TestCase): posix1e.ACL_TYPE_DEFAULT) # build tuples of the Entry objects that were added to acl - # and defaacl so they're easier to compare for equality + # and defacl so they're easier to compare for equality added_acls = [] for acl in acl_entries: added_acls.append((acl.acl, acl.tag_type, acl.qualifier, @@ -446,17 +434,17 @@ class TestPOSIXTool(Bcfg2TestCase): # they've already been iterated over once fileacl_rv.__iter__.return_value = iter(file_acls) filedef_rv.__iter__.return_value = iter(file_acls) - mock_list_entry_acls.reset_mock() - mock_norm_uid.reset_mock() - mock_norm_gid.reset_mock() + ptool._list_entry_acls.reset_mock() + ptool._norm_uid.reset_mock() + ptool._norm_gid.reset_mock() mock_isdir.return_value = False acl_entries = [] - self.assertTrue(self.ptool._set_acls(entry, path="/bin/bar")) + self.assertTrue(ptool._set_acls(entry, path="/bin/bar")) mock_ACL.assert_called_with(file="/bin/bar") self.assertItemsEqual(fileacl_rv.delete_entry.call_args_list, [call(a) for a in remove_acls]) - mock_list_entry_acls.assert_called_with(entry) - mock_norm_gid.assert_called_with("group") + ptool._list_entry_acls.assert_called_with(entry) + ptool._norm_gid.assert_called_with("group") fileacl_rv.calc_mask.assert_any_call() fileacl_rv.applyto.assert_called_with("/bin/bar", posix1e.ACL_TYPE_ACCESS) @@ -472,22 +460,23 @@ class TestPOSIXTool(Bcfg2TestCase): @patchIf(HAS_SELINUX, "selinux.restorecon") @patchIf(HAS_SELINUX, "selinux.lsetfilecon") def test_set_secontext(self, mock_lsetfilecon, mock_restorecon): + ptool = self.get_obj() entry = lxml.etree.Element("Path", name="/etc/foo", type="file") # disable selinux for the initial test Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX = False - self.assertTrue(self.ptool._set_secontext(entry)) + self.assertTrue(ptool._set_secontext(entry)) Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX = True # no context given - self.assertTrue(self.ptool._set_secontext(entry)) + self.assertTrue(ptool._set_secontext(entry)) self.assertFalse(mock_restorecon.called) self.assertFalse(mock_lsetfilecon.called) mock_restorecon.reset_mock() mock_lsetfilecon.reset_mock() entry.set("secontext", "__default__") - self.assertTrue(self.ptool._set_secontext(entry)) + self.assertTrue(ptool._set_secontext(entry)) mock_restorecon.assert_called_with(entry.get("name")) self.assertFalse(mock_lsetfilecon.called) @@ -495,85 +484,91 @@ class TestPOSIXTool(Bcfg2TestCase): mock_lsetfilecon.reset_mock() mock_lsetfilecon.return_value = 0 entry.set("secontext", "foo_t") - self.assertTrue(self.ptool._set_secontext(entry)) + self.assertTrue(ptool._set_secontext(entry)) self.assertFalse(mock_restorecon.called) mock_lsetfilecon.assert_called_with(entry.get("name"), "foo_t") mock_restorecon.reset_mock() mock_lsetfilecon.reset_mock() mock_lsetfilecon.return_value = 1 - self.assertFalse(self.ptool._set_secontext(entry)) + self.assertFalse(ptool._set_secontext(entry)) self.assertFalse(mock_restorecon.called) mock_lsetfilecon.assert_called_with(entry.get("name"), "foo_t") @patch("grp.getgrnam") def test_norm_gid(self, mock_getgrnam): - self.assertEqual(5, self.ptool._norm_gid("5")) + ptool = self.get_obj() + self.assertEqual(5, ptool._norm_gid("5")) self.assertFalse(mock_getgrnam.called) mock_getgrnam.reset_mock() mock_getgrnam.return_value = ("group", "x", 5, []) - self.assertEqual(5, self.ptool._norm_gid("group")) + self.assertEqual(5, ptool._norm_gid("group")) mock_getgrnam.assert_called_with("group") - @patch("Bcfg2.Client.Tools.POSIX.base.%s._norm_gid" % test_obj.__name__) - def test_norm_entry_gid(self, mock_norm_gid): + def test_norm_entry_gid(self): + ptool = self.get_obj() + ptool._norm_gid = Mock() entry = lxml.etree.Element("Path", name="/test", type="file", group="group", owner="user") - mock_norm_gid.return_value = 10 - self.assertEqual(10, self.ptool._norm_entry_gid(entry)) - mock_norm_gid.assert_called_with(entry.get("group")) + self.assertEqual(ptool._norm_entry_gid(entry), + ptool._norm_gid.return_value) + ptool._norm_gid.assert_called_with(entry.get("group")) - mock_norm_gid.reset_mock() - mock_norm_gid.side_effect = KeyError - self.assertEqual(0, self.ptool._norm_entry_gid(entry)) - mock_norm_gid.assert_called_with(entry.get("group")) + ptool._norm_gid.reset_mock() + ptool._norm_gid.side_effect = KeyError + self.assertEqual(ptool._norm_entry_gid(entry), 0) + ptool._norm_gid.assert_called_with(entry.get("group")) @patch("pwd.getpwnam") def test_norm_uid(self, mock_getpwnam): - self.assertEqual(5, self.ptool._norm_uid("5")) + ptool = self.get_obj() + self.assertEqual(5, ptool._norm_uid("5")) self.assertFalse(mock_getpwnam.called) mock_getpwnam.reset_mock() mock_getpwnam.return_value = ("user", "x", 5, 5, "User", "/home/user", "/bin/zsh") - self.assertEqual(5, self.ptool._norm_uid("user")) + self.assertEqual(5, ptool._norm_uid("user")) mock_getpwnam.assert_called_with("user") - @patch("Bcfg2.Client.Tools.POSIX.base.%s._norm_uid" % test_obj.__name__) - def test_norm_entry_uid(self, mock_norm_uid): + def test_norm_entry_uid(self): + ptool = self.get_obj() + ptool._norm_uid = Mock() entry = lxml.etree.Element("Path", name="/test", type="file", group="group", owner="user") - mock_norm_uid.return_value = 10 - self.assertEqual(10, self.ptool._norm_entry_uid(entry)) - mock_norm_uid.assert_called_with(entry.get("owner")) + self.assertEqual(ptool._norm_entry_uid(entry), + ptool._norm_uid.return_value) + ptool._norm_uid.assert_called_with(entry.get("owner")) - mock_norm_uid.reset_mock() - mock_norm_uid.side_effect = KeyError - self.assertEqual(0, self.ptool._norm_entry_uid(entry)) - mock_norm_uid.assert_called_with(entry.get("owner")) + ptool._norm_uid.reset_mock() + ptool._norm_uid.side_effect = KeyError + self.assertEqual(ptool._norm_entry_uid(entry), 0) + ptool._norm_uid.assert_called_with(entry.get("owner")) def test_norm_acl_perms(self): - # there's basically no reasonably way to test the Permset + # there's basically no reasonable way to test the Permset # object parsing feature without writing our own Mock object # that re-implements Permset.test(). silly pylibacl won't let # us create standalone Entry or Permset objects. - self.assertEqual(5, self.ptool._norm_acl_perms("5")) - self.assertEqual(0, self.ptool._norm_acl_perms("55")) - self.assertEqual(5, self.ptool._norm_acl_perms("rx")) - self.assertEqual(5, self.ptool._norm_acl_perms("r-x")) - self.assertEqual(6, self.ptool._norm_acl_perms("wr-")) - self.assertEqual(0, self.ptool._norm_acl_perms("rwrw")) - self.assertEqual(0, self.ptool._norm_acl_perms("-")) - self.assertEqual(0, self.ptool._norm_acl_perms("a")) - self.assertEqual(6, self.ptool._norm_acl_perms("rwa")) - self.assertEqual(4, self.ptool._norm_acl_perms("rr")) + ptool = self.get_obj() + self.assertEqual(5, ptool._norm_acl_perms("5")) + self.assertEqual(0, ptool._norm_acl_perms("55")) + self.assertEqual(5, ptool._norm_acl_perms("rx")) + self.assertEqual(5, ptool._norm_acl_perms("r-x")) + self.assertEqual(6, ptool._norm_acl_perms("wr-")) + self.assertEqual(0, ptool._norm_acl_perms("rwrw")) + self.assertEqual(0, ptool._norm_acl_perms("-")) + self.assertEqual(0, ptool._norm_acl_perms("a")) + self.assertEqual(6, ptool._norm_acl_perms("rwa")) + self.assertEqual(4, ptool._norm_acl_perms("rr")) @patch('os.stat') def test__gather_data(self, mock_stat): + ptool = self.get_obj() path = '/test' mock_stat.side_effect = OSError - self.assertFalse(self.ptool._gather_data(path)[0]) + self.assertFalse(ptool._gather_data(path)[0]) mock_stat.assert_called_with(path) mock_stat.reset_mock() @@ -598,7 +593,7 @@ class TestPOSIXTool(Bcfg2TestCase): Bcfg2.Client.Tools.POSIX.base.HAS_ACLS) Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX = False Bcfg2.Client.Tools.POSIX.base.HAS_ACLS = False - self.assertEqual(self.ptool._gather_data(path), + self.assertEqual(ptool._gather_data(path), (stat_rv, '0', '10', '0660', None, None)) Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX, \ Bcfg2.Client.Tools.POSIX.base.HAS_ACLS = states @@ -606,6 +601,7 @@ class TestPOSIXTool(Bcfg2TestCase): @skipUnless(HAS_SELINUX, "SELinux not found, skipping") def test__gather_data_selinux(self): + ptool = self.get_obj() context = 'system_u:object_r:root_t:s0' path = '/test' @@ -620,7 +616,7 @@ class TestPOSIXTool(Bcfg2TestCase): Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX) Bcfg2.Client.Tools.POSIX.base.HAS_ACLS = False Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX = True - self.assertEqual(self.ptool._gather_data(path)[4], 'root_t') + self.assertEqual(ptool._gather_data(path)[4], 'root_t') Bcfg2.Client.Tools.POSIX.base.HAS_ACLS, \ Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX = state mock_getfilecon.assert_called_with(path) @@ -629,12 +625,12 @@ class TestPOSIXTool(Bcfg2TestCase): @skipUnless(HAS_ACLS, "ACLS not found, skipping") @patch('os.stat') - @patch("Bcfg2.Client.Tools.POSIX.base.%s._list_file_acls" % - test_obj.__name__) - def test__gather_data_acls(self, mock_list_file_acls, mock_stat): + def test__gather_data_acls(self, mock_stat): + ptool = self.get_obj() + ptool._list_file_acls = Mock() acls = {("default", posix1e.ACL_USER, "testuser"): "rwx", ("access", posix1e.ACL_GROUP, "testgroup"): "rx"} - mock_list_file_acls.return_value = acls + ptool._list_file_acls.return_value = acls path = '/test' mock_stat.return_value = MagicMock() mock_stat.return_value.__getitem__.return_value = MagicMock() @@ -643,21 +639,18 @@ class TestPOSIXTool(Bcfg2TestCase): Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX) Bcfg2.Client.Tools.POSIX.base.HAS_ACLS = True Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX = False - self.assertItemsEqual(self.ptool._gather_data(path)[5], acls) + self.assertItemsEqual(ptool._gather_data(path)[5], acls) Bcfg2.Client.Tools.POSIX.base.HAS_ACLS, \ Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX = state - mock_list_file_acls.assert_called_with(path) + ptool._list_file_acls.assert_called_with(path) @patchIf(HAS_SELINUX, "selinux.matchpathcon") - @patch("Bcfg2.Client.Tools.POSIX.base.%s._verify_acls" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.base.%s._gather_data" % test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.base.%s._norm_entry_uid" % - test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.base.%s._norm_entry_gid" % - test_obj.__name__) - def test_verify_metadata(self, mock_norm_gid, mock_norm_uid, - mock_gather_data, mock_verify_acls, - mock_matchpathcon): + def test_verify_metadata(self, mock_matchpathcon): + ptool = self.get_obj() + ptool._norm_entry_uid = Mock() + ptool._norm_entry_gid = Mock() + ptool._verify_acls = Mock() + ptool._gather_data = Mock() entry = lxml.etree.Element("Path", name="/test", type="file", group="group", owner="user", mode="664", secontext='etc_t') @@ -666,34 +659,34 @@ class TestPOSIXTool(Bcfg2TestCase): orig_entry = copy.deepcopy(entry) def reset(): - mock_gather_data.reset_mock() - mock_verify_acls.reset_mock() - mock_norm_uid.reset_mock() - mock_norm_gid.reset_mock() + ptool._gather_data.reset_mock() + ptool._verify_acls.reset_mock() + ptool._norm_entry_uid.reset_mock() + ptool._norm_entry_gid.reset_mock() return copy.deepcopy(orig_entry) # test nonexistent file - mock_gather_data.return_value = (False, None, None, None, None, None) - self.assertFalse(self.ptool._verify_metadata(entry)) + ptool._gather_data.return_value = (False, None, None, None, None, None) + self.assertFalse(ptool._verify_metadata(entry)) self.assertEqual(entry.get("current_exists", "").lower(), "false") - mock_gather_data.assert_called_with(entry.get("name")) + ptool._gather_data.assert_called_with(entry.get("name")) # expected data. tuple of attr, return value index, value expected = [('current_owner', 1, '0'), ('current_group', 2, '10'), ('current_mode', 3, '0664'), ('current_secontext', 4, 'etc_t')] - mock_norm_uid.return_value = 0 - mock_norm_gid.return_value = 10 + ptool._norm_entry_uid.return_value = 0 + ptool._norm_entry_gid.return_value = 10 gather_data_rv = [MagicMock(), None, None, None, None, []] for attr, idx, val in expected: gather_data_rv[idx] = val entry = reset() - mock_gather_data.return_value = tuple(gather_data_rv) - self.assertTrue(self.ptool._verify_metadata(entry)) - mock_gather_data.assert_called_with(entry.get("name")) - mock_verify_acls.assert_called_with(entry, path=entry.get("name")) + ptool._gather_data.return_value = tuple(gather_data_rv) + self.assertTrue(ptool._verify_metadata(entry)) + ptool._gather_data.assert_called_with(entry.get("name")) + ptool._verify_acls.assert_called_with(entry, path=entry.get("name")) self.assertEqual(entry.get("current_exists", 'true'), 'true') for attr, idx, val in expected: self.assertEqual(entry.get(attr), val) @@ -703,10 +696,10 @@ class TestPOSIXTool(Bcfg2TestCase): gather_data_rv[4] = None sestate = Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX Bcfg2.Client.Tools.POSIX.base.HAS_SELINUX = False - mock_gather_data.return_value = tuple(gather_data_rv) - self.assertTrue(self.ptool._verify_metadata(entry)) - mock_gather_data.assert_called_with(entry.get("name")) - mock_verify_acls.assert_called_with(entry, path=entry.get("name")) + ptool._gather_data.return_value = tuple(gather_data_rv) + self.assertTrue(ptool._verify_metadata(entry)) + ptool._gather_data.assert_called_with(entry.get("name")) + ptool._verify_acls.assert_called_with(entry, path=entry.get("name")) self.assertEqual(entry.get("current_exists", 'true'), 'true') for attr, idx, val in expected: if attr != 'current_secontext': @@ -716,7 +709,7 @@ class TestPOSIXTool(Bcfg2TestCase): gather_data_rv = [MagicMock(), None, None, None, None, []] for attr, idx, val in expected: gather_data_rv[idx] = val - mock_gather_data.return_value = tuple(gather_data_rv) + ptool._gather_data.return_value = tuple(gather_data_rv) mtime = 1344430414 entry = reset() @@ -724,10 +717,10 @@ class TestPOSIXTool(Bcfg2TestCase): stat_rv = MagicMock() stat_rv.__getitem__.return_value = mtime gather_data_rv[0] = stat_rv - mock_gather_data.return_value = tuple(gather_data_rv) - self.assertTrue(self.ptool._verify_metadata(entry)) - mock_gather_data.assert_called_with(entry.get("name")) - mock_verify_acls.assert_called_with(entry, path=entry.get("name")) + ptool._gather_data.return_value = tuple(gather_data_rv) + self.assertTrue(ptool._verify_metadata(entry)) + ptool._gather_data.assert_called_with(entry.get("name")) + ptool._verify_acls.assert_called_with(entry, path=entry.get("name")) self.assertEqual(entry.get("current_exists", 'true'), 'true') for attr, idx, val in expected: self.assertEqual(entry.get(attr), val) @@ -748,10 +741,10 @@ class TestPOSIXTool(Bcfg2TestCase): for attr, idx, val in expected: gather_data_rv[idx] = val gather_data_rv[fail_idx] = fail_val - mock_gather_data.return_value = tuple(gather_data_rv) - self.assertFalse(self.ptool._verify_metadata(entry)) - mock_gather_data.assert_called_with(entry.get("name")) - mock_verify_acls.assert_called_with(entry, path=entry.get("name")) + ptool._gather_data.return_value = tuple(gather_data_rv) + self.assertFalse(ptool._verify_metadata(entry)) + ptool._gather_data.assert_called_with(entry.get("name")) + ptool._verify_acls.assert_called_with(entry, path=entry.get("name")) self.assertEqual(entry.get("current_exists", 'true'), 'true') self.assertEqual(entry.get(fail_attr), fail_val) for attr, idx, val in expected: @@ -768,10 +761,10 @@ class TestPOSIXTool(Bcfg2TestCase): gather_data_rv = [fail_stat_rv, None, None, None, None, []] for attr, idx, val in expected: gather_data_rv[idx] = val - mock_gather_data.return_value = tuple(gather_data_rv) - self.assertFalse(self.ptool._verify_metadata(entry)) - mock_gather_data.assert_called_with(entry.get("name")) - mock_verify_acls.assert_called_with(entry, path=entry.get("name")) + ptool._gather_data.return_value = tuple(gather_data_rv) + self.assertFalse(ptool._verify_metadata(entry)) + ptool._gather_data.assert_called_with(entry.get("name")) + ptool._verify_acls.assert_called_with(entry, path=entry.get("name")) self.assertEqual(entry.get("current_exists", 'true'), 'true') for attr, idx, val in expected: self.assertEqual(entry.get(attr), val) @@ -790,10 +783,10 @@ class TestPOSIXTool(Bcfg2TestCase): gather_data_rv = [stat_rv, None, None, None, None, []] for attr, idx, val in expected: gather_data_rv[idx] = val - mock_gather_data.return_value = tuple(gather_data_rv) - self.assertTrue(self.ptool._verify_metadata(entry)) - mock_gather_data.assert_called_with(entry.get("name")) - mock_verify_acls.assert_called_with(entry, + ptool._gather_data.return_value = tuple(gather_data_rv) + self.assertTrue(ptool._verify_metadata(entry)) + ptool._gather_data.assert_called_with(entry.get("name")) + ptool._verify_acls.assert_called_with(entry, path=entry.get("name")) mock_matchpathcon.assert_called_with(entry.get("name"), 0) self.assertEqual(entry.get("current_exists", 'true'), 'true') @@ -806,9 +799,9 @@ class TestPOSIXTool(Bcfg2TestCase): entry.set("secontext", "__default__") mock_matchpathcon.return_value = [1 + len(context2), context2] - self.assertFalse(self.ptool._verify_metadata(entry)) - mock_gather_data.assert_called_with(entry.get("name")) - mock_verify_acls.assert_called_with(entry, + self.assertFalse(ptool._verify_metadata(entry)) + ptool._gather_data.assert_called_with(entry.get("name")) + ptool._verify_acls.assert_called_with(entry, path=entry.get("name")) mock_matchpathcon.assert_called_with(entry.get("name"), 0) self.assertEqual(entry.get("current_exists", 'true'), 'true') @@ -818,21 +811,24 @@ class TestPOSIXTool(Bcfg2TestCase): @skipUnless(HAS_ACLS, "ACLS not found, skipping") def test_list_entry_acls(self): + ptool = self.get_obj() entry = lxml.etree.Element("Path", name="/test", type="file") lxml.etree.SubElement(entry, "ACL", scope="user", type="default", user="user", perms="rwx") lxml.etree.SubElement(entry, "ACL", scope="group", type="access", group="group", perms="5") - self.assertItemsEqual(self.ptool._list_entry_acls(entry), + self.assertItemsEqual(ptool._list_entry_acls(entry), {("default", posix1e.ACL_USER, "user"): 7, ("access", posix1e.ACL_GROUP, "group"): 5}) @skipUnless(HAS_ACLS, "ACLS not found, skipping") + @patchIf(HAS_ACLS, "posix1e.ACL") @patch("pwd.getpwuid") @patch("grp.getgrgid") @patch("os.path.isdir") def test_list_file_acls(self, mock_isdir, mock_getgrgid, mock_getpwuid, mock_ACL): + ptool = self.get_obj() path = '/test' # build a set of file ACLs to return from posix1e.ACL(file=...) @@ -881,15 +877,15 @@ class TestPOSIXTool(Bcfg2TestCase): mock_ACL.reset_mock() mock_ACL.side_effect = IOError(95, "Operation not supported") - self.assertItemsEqual(self.ptool._list_file_acls(path), dict()) + self.assertItemsEqual(ptool._list_file_acls(path), dict()) reset() mock_ACL.side_effect = IOError - self.assertItemsEqual(self.ptool._list_file_acls(path), dict()) + self.assertItemsEqual(ptool._list_file_acls(path), dict()) reset() mock_ACL.side_effect = mock_acl_rv - self.assertItemsEqual(self.ptool._list_file_acls(path), acls) + self.assertItemsEqual(ptool._list_file_acls(path), acls) mock_isdir.assert_called_with(path) mock_getgrgid.assert_called_with(100) mock_getpwuid.assert_called_with(10) @@ -903,7 +899,7 @@ class TestPOSIXTool(Bcfg2TestCase): defacls = acls for akey, perms in acls.items(): defacls[('default', akey[1], akey[2])] = perms - self.assertItemsEqual(self.ptool._list_file_acls(path), defacls) + self.assertItemsEqual(ptool._list_file_acls(path), defacls) mock_isdir.assert_called_with(path) self.assertItemsEqual(mock_getgrgid.call_args_list, [call(100), call(100)]) @@ -912,20 +908,11 @@ class TestPOSIXTool(Bcfg2TestCase): self.assertItemsEqual(mock_ACL.call_args_list, [call(file=path), call(filedef=path)]) - if HAS_ACLS: - # python 2.6 applies decorators at compile-time, not at - # run-time, so we can't do these as decorators because - # pylibacl might not be installed. (If it's not, this test - # will be skipped, so as long as this is done at run-time - # we're safe.) - test_list_file_acls = patch("posix1e.ACL")(test_list_file_acls) - @skipUnless(HAS_ACLS, "ACLS not found, skipping") - @patch("Bcfg2.Client.Tools.POSIX.base.%s._list_file_acls" % - test_obj.__name__) - @patch("Bcfg2.Client.Tools.POSIX.base.%s._list_entry_acls" % - test_obj.__name__) - def test_verify_acls(self, mock_list_entry_acls, mock_list_file_acls): + def test_verify_acls(self): + ptool = self.get_obj() + ptool._list_file_acls = Mock() + ptool._list_entry_acls = Mock() entry = lxml.etree.Element("Path", name="/test", type="file") # we can't test to make sure that errors get properly sorted # into (missing, extra, wrong) without refactoring the @@ -938,44 +925,45 @@ class TestPOSIXTool(Bcfg2TestCase): extra_acls = copy.deepcopy(acls) extra_acls[("access", posix1e.ACL_USER, "user2")] = 4 - mock_list_entry_acls.return_value = acls - mock_list_file_acls.return_value = acls - self.assertTrue(self.ptool._verify_acls(entry)) - mock_list_entry_acls.assert_called_with(entry) - mock_list_file_acls.assert_called_with(entry.get("name")) + ptool._list_entry_acls.return_value = acls + ptool._list_file_acls.return_value = acls + self.assertTrue(ptool._verify_acls(entry)) + ptool._list_entry_acls.assert_called_with(entry) + ptool._list_file_acls.assert_called_with(entry.get("name")) # test missing - mock_list_entry_acls.reset_mock() - mock_list_file_acls.reset_mock() - mock_list_file_acls.return_value = extra_acls - self.assertFalse(self.ptool._verify_acls(entry)) - mock_list_entry_acls.assert_called_with(entry) - mock_list_file_acls.assert_called_with(entry.get("name")) + ptool._list_entry_acls.reset_mock() + ptool._list_file_acls.reset_mock() + ptool._list_file_acls.return_value = extra_acls + self.assertFalse(ptool._verify_acls(entry)) + ptool._list_entry_acls.assert_called_with(entry) + ptool._list_file_acls.assert_called_with(entry.get("name")) # test extra - mock_list_entry_acls.reset_mock() - mock_list_file_acls.reset_mock() - mock_list_entry_acls.return_value = extra_acls - mock_list_file_acls.return_value = acls - self.assertFalse(self.ptool._verify_acls(entry)) - mock_list_entry_acls.assert_called_with(entry) - mock_list_file_acls.assert_called_with(entry.get("name")) + ptool._list_entry_acls.reset_mock() + ptool._list_file_acls.reset_mock() + ptool._list_entry_acls.return_value = extra_acls + ptool._list_file_acls.return_value = acls + self.assertFalse(ptool._verify_acls(entry)) + ptool._list_entry_acls.assert_called_with(entry) + ptool._list_file_acls.assert_called_with(entry.get("name")) # test wrong wrong_acls = copy.deepcopy(extra_acls) wrong_acls[("access", posix1e.ACL_USER, "user2")] = 5 - mock_list_entry_acls.reset_mock() - mock_list_file_acls.reset_mock() - mock_list_entry_acls.return_value = extra_acls - mock_list_file_acls.return_value = wrong_acls - self.assertFalse(self.ptool._verify_acls(entry)) - mock_list_entry_acls.assert_called_with(entry) - mock_list_file_acls.assert_called_with(entry.get("name")) + ptool._list_entry_acls.reset_mock() + ptool._list_file_acls.reset_mock() + ptool._list_entry_acls.return_value = extra_acls + ptool._list_file_acls.return_value = wrong_acls + self.assertFalse(ptool._verify_acls(entry)) + ptool._list_entry_acls.assert_called_with(entry) + ptool._list_file_acls.assert_called_with(entry.get("name")) @patch("os.makedirs") @patch("os.path.exists") - @patch("Bcfg2.Client.Tools.POSIX.base.%s._set_perms" % test_obj.__name__) - def test_makedirs(self, mock_set_perms, mock_exists, mock_makedirs): + def test_makedirs(self, mock_exists, mock_makedirs): + ptool = self.get_obj() + ptool._set_perms = Mock() entry = lxml.etree.Element("Path", name="/test/foo/bar", type="directory", mode="0644") parent_entry = lxml.etree.Element("Path", name="/test/foo/bar", @@ -983,33 +971,33 @@ class TestPOSIXTool(Bcfg2TestCase): def reset(): mock_exists.reset_mock() - mock_set_perms.reset_mock() + ptool._set_perms.reset_mock() mock_makedirs.reset_mock() - mock_set_perms.return_value = True + ptool._set_perms.return_value = True def path_exists_rv(path): if path == "/test": return True else: return False mock_exists.side_effect = path_exists_rv - self.assertTrue(self.ptool._makedirs(entry)) + self.assertTrue(ptool._makedirs(entry)) self.assertItemsEqual(mock_exists.call_args_list, [call("/test"), call("/test/foo"), call("/test/foo/bar")]) - for args in mock_set_perms.call_args_list: + for args in ptool._set_perms.call_args_list: self.assertXMLEqual(args[0][0], parent_entry) - self.assertItemsEqual([a[1] for a in mock_set_perms.call_args_list], + self.assertItemsEqual([a[1] for a in ptool._set_perms.call_args_list], [dict(path="/test/foo"), dict(path="/test/foo/bar")]) mock_makedirs.assert_called_with(entry.get("name")) reset() mock_makedirs.side_effect = OSError - self.assertFalse(self.ptool._makedirs(entry)) - for args in mock_set_perms.call_args_list: + self.assertFalse(ptool._makedirs(entry)) + for args in ptool._set_perms.call_args_list: self.assertXMLEqual(args[0][0], parent_entry) - self.assertItemsEqual([a[1] for a in mock_set_perms.call_args_list], + self.assertItemsEqual([a[1] for a in ptool._set_perms.call_args_list], [dict(path="/test/foo"), dict(path="/test/foo/bar")]) @@ -1020,14 +1008,89 @@ class TestPOSIXTool(Bcfg2TestCase): return False else: return True - mock_set_perms.side_effect = set_perms_rv - self.assertFalse(self.ptool._makedirs(entry)) + ptool._set_perms.side_effect = set_perms_rv + self.assertFalse(ptool._makedirs(entry)) self.assertItemsEqual(mock_exists.call_args_list, [call("/test"), call("/test/foo"), call("/test/foo/bar")]) - for args in mock_set_perms.call_args_list: + for args in ptool._set_perms.call_args_list: self.assertXMLEqual(args[0][0], parent_entry) - self.assertItemsEqual([a[1] for a in mock_set_perms.call_args_list], + self.assertItemsEqual([a[1] for a in ptool._set_perms.call_args_list], [dict(path="/test/foo"), dict(path="/test/foo/bar")]) mock_makedirs.assert_called_with(entry.get("name")) + + +class TestPOSIXLinkTool(TestPOSIXTool): + test_obj = POSIXLinkTool + + @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.verify") + def test_verify(self, mock_verify): + entry = lxml.etree.Element("Path", name="/test", type="testlink", + to="/dest") + ptool = self.get_obj() + linktype = ptool.__linktype__ + ptool.__linktype__ = "test" + ptool._verify = Mock() + + ptool._verify.return_value = True + mock_verify.return_value = False + self.assertFalse(ptool.verify(entry, [])) + ptool._verify.assert_called_with(entry) + mock_verify.assert_called_with(ptool, entry, []) + + ptool._verify.reset_mock() + mock_verify.reset_mock() + mock_verify.return_value = True + self.assertTrue(ptool.verify(entry, [])) + ptool._verify.assert_called_with(entry) + mock_verify.assert_called_with(ptool, entry, []) + + ptool._verify.reset_mock() + mock_verify.reset_mock() + ptool._verify.return_value = False + self.assertFalse(ptool.verify(entry, [])) + ptool._verify.assert_called_with(entry) + mock_verify.assert_called_with(ptool, entry, []) + + ptool._verify.reset_mock() + mock_verify.reset_mock() + ptool._verify.side_effect = OSError + self.assertFalse(ptool.verify(entry, [])) + ptool._verify.assert_called_with(entry) + ptool.__linktype__ = linktype + + def test__verify(self): + ptool = self.get_obj() + self.assertRaises(NotImplementedError, ptool._verify, Mock()) + + @patch("Bcfg2.Client.Tools.POSIX.base.POSIXTool.install") + def test_install(self, mock_install): + entry = lxml.etree.Element("Path", name="/test", type="symlink", + to="/dest") + ptool = self.get_obj() + linktype = ptool.__linktype__ + ptool.__linktype__ = "test" + ptool._exists = Mock() + ptool._link = Mock() + + ptool._exists.return_value = False + mock_install.return_value = True + self.assertTrue(ptool.install(entry)) + ptool._exists.assert_called_with(entry, remove=True) + ptool._link.assert_called_with(entry) + mock_install.assert_called_with(ptool, entry) + + ptool._link.reset_mock() + ptool._exists.reset_mock() + mock_install.reset_mock() + ptool._link.side_effect = OSError + self.assertFalse(ptool.install(entry)) + ptool._exists.assert_called_with(entry, remove=True) + ptool._link.assert_called_with(entry) + mock_install.assert_called_with(ptool, entry) + ptool.__linktype__ = linktype + + def test__link(self): + ptool = self.get_obj() + self.assertRaises(NotImplementedError, ptool._link, Mock()) diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIXUsers.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIXUsers.py index bcf6cf133..b42dc57d8 100644 --- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIXUsers.py +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIXUsers.py @@ -17,6 +17,7 @@ while path != "/": break path = os.path.dirname(path) from common import * +from TestTools.Test_init import TestTool class TestIDRangeSet(Bcfg2TestCase): @@ -118,26 +119,14 @@ class TestExecutor(Bcfg2TestCase): exc.retval) -class TestPOSIXUsers(Bcfg2TestCase): +class TestPOSIXUsers(TestTool): test_obj = POSIXUsers def get_obj(self, logger=None, setup=None, config=None): - if config is None: - config = lxml.etree.Element("Configuration") - - if logger is None: - def print_msg(msg): - print(msg) - logger = Mock() - logger.error = Mock(side_effect=print_msg) - logger.warning = Mock(side_effect=print_msg) - logger.info = Mock(side_effect=print_msg) - logger.debug = Mock(side_effect=print_msg) - if setup is None: setup = MagicMock() setup.__getitem__.return_value = [] - return self.test_obj(logger, setup, config) + return TestTool.get_obj(self, logger, setup, config) @patch("pwd.getpwall") @patch("grp.getgrall") diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/Test_init.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/Test_init.py new file mode 100644 index 000000000..355fec494 --- /dev/null +++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/Test_init.py @@ -0,0 +1,648 @@ +import os +import sys +import copy +import lxml.etree +import subprocess +from mock import Mock, MagicMock, patch +from Bcfg2.Client.Tools import Tool, SvcTool, PkgTool, \ + ToolInstantiationError + +# add all parent testsuite directories to sys.path to allow (most) +# relative imports in python 2.4 +path = os.path.dirname(__file__) +while path != "/": + if os.path.basename(path).lower().startswith("test"): + sys.path.append(path) + if os.path.basename(path) == "testsuite": + break + path = os.path.dirname(path) +from common import * + + +class TestTool(Bcfg2TestCase): + test_obj = Tool + + def get_obj(self, logger=None, setup=None, config=None): + if config is None: + config = lxml.etree.Element("Configuration") + if not logger: + def print_msg(msg): + print(msg) + logger = Mock() + logger.error = Mock(side_effect=print_msg) + logger.warning = Mock(side_effect=print_msg) + logger.info = Mock(side_effect=print_msg) + logger.debug = Mock(side_effect=print_msg) + if not setup: + setup = MagicMock() + if 'command_timeout' not in setup: + setup['command_timeout'] = None + execs = self.test_obj.__execs__ + self.test_obj.__execs__ = [] + rv = self.test_obj(logger, setup, config) + self.test_obj.__execs__ = execs + return rv + + def test__init(self): + @patch("%s.%s._check_execs" % (self.test_obj.__module__, + self.test_obj.__name__)) + @patch("%s.%s._analyze_config" % (self.test_obj.__module__, + self.test_obj.__name__)) + def inner(mock_analyze_config, mock_check_execs): + t = self.get_obj() + mock_analyze_config.assert_called_with() + mock_check_execs.assert_called_with() + + def test__analyze_config(self): + t = self.get_obj() + t.getSupportedEntries = Mock() + + t.__important__ = ["/test"] + important = [] + t.config = lxml.etree.Element("Config") + bundle1 = lxml.etree.SubElement(t.config, "Bundle") + important.append(lxml.etree.SubElement(bundle1, "Path", + name="/foo", important="true")) + lxml.etree.SubElement(bundle1, "Package", name="bar", important="true") + lxml.etree.SubElement(bundle1, "Path", name="/bar") + bundle2 = lxml.etree.SubElement(t.config, "Bundle") + important.append(lxml.etree.SubElement(bundle2, "Path", name="/quux", + important="true")) + lxml.etree.SubElement(bundle2, "Path", name="/baz", important="false") + + t._analyze_config() + self.assertItemsEqual(t.__important__, + ["/test"] + [e.get("name") for e in important]) + t.getSupportedEntries.assert_called_with() + + def test__check_execs(self): + t = self.get_obj() + if t.__execs__ == []: + t.__execs__.append("/bin/true") + + @patch("os.stat") + def inner(mock_stat): + mock_stat.return_value = (33261, 2245040, 64770L, 1, 0, 0, 25552, + 1360831382, 1352194410, 1354626626) + t._check_execs() + self.assertItemsEqual(mock_stat.call_args_list, + [call(e) for e in t.__execs__]) + + # not executable + mock_stat.reset_mock() + mock_stat.return_value = (33188, 2245040, 64770L, 1, 0, 0, 25552, + 1360831382, 1352194410, 1354626626) + self.assertRaises(ToolInstantiationError, t._check_execs) + + # non-existant + mock_stat.reset_mock() + mock_stat.side_effect = OSError + self.assertRaises(ToolInstantiationError, t._check_execs) + + inner() + + def test_BundleUpdated(self): + pass + + def test_BundleNotUpdated(self): + pass + + def test_Inventory(self): + t = self.get_obj() + t.canVerify = Mock() + t.canVerify.side_effect = lambda e: e.get("verify") != "false" + t.buildModlist = Mock() + t.FindExtra = Mock() + t.VerifyPath = Mock() + t.VerifyPackage = Mock() + t.VerifyService = Mock() + + def reset(): + t.canVerify.reset_mock() + t.buildModlist.reset_mock() + t.FindExtra.reset_mock() + t.VerifyPath.reset_mock() + t.VerifyPackage.reset_mock() + t.VerifyService.reset_mock() + + paths = [] + packages = [] + services = [] + config = lxml.etree.Element("Configuration") + bundle1 = lxml.etree.SubElement(config, "Bundle") + paths.append(lxml.etree.SubElement(bundle1, "Path", name="/foo")) + lxml.etree.SubElement(bundle1, "Package", name="foo", verify="false") + packages.append(lxml.etree.SubElement(bundle1, "Package", name="bar")) + lxml.etree.SubElement(bundle1, "Bogus") + + bundle2 = lxml.etree.SubElement(config, "Bundle") + paths.append(lxml.etree.SubElement(bundle2, "Path", name="/bar")) + services.append(lxml.etree.SubElement(bundle2, "Service", name="bar")) + lxml.etree.SubElement(bundle2, "Path", name="/baz", verify="false") + + expected_states = dict([(e, t.VerifyPath.return_value) + for e in paths]) + expected_states.update(dict([(e, t.VerifyPackage.return_value) + for e in packages])) + expected_states.update(dict([(e, t.VerifyService.return_value) + for e in services])) + + def perform_assertions(states): + t.buildModlist.assert_called_with() + t.FindExtra.assert_called_with() + self.assertItemsEqual(t.canVerify.call_args_list, + [call(e) for e in bundle1.getchildren()] + \ + [call(e) for e in bundle2.getchildren()]) + self.assertItemsEqual(t.VerifyPath.call_args_list, + [call(e, t.buildModlist.return_value) + for e in paths]) + self.assertItemsEqual(t.VerifyPackage.call_args_list, + [call(e, t.buildModlist.return_value) + for e in packages]) + self.assertItemsEqual(t.VerifyService.call_args_list, + [call(e, t.buildModlist.return_value) + for e in services]) + self.assertItemsEqual(states, expected_states) + self.assertEqual(t.extra, t.FindExtra.return_value) + + actual_states = dict() + t.Inventory(actual_states, structures=[bundle1, bundle2]) + perform_assertions(actual_states) + + reset() + actual_states = dict() + t.config = config + t.Inventory(actual_states) + perform_assertions(actual_states) + + def test_Install(self): + t = self.get_obj() + t.InstallPath = Mock() + t.InstallPackage = Mock() + t.InstallService = Mock() + + t.InstallPath.side_effect = lambda e: e.get("modified") == "true" + t.InstallPackage.side_effect = lambda e: e.get("modified") == "true" + t.InstallService.side_effect = lambda e: e.get("modified") == "true" + + entries = [lxml.etree.Element("Path", name="/foo", modified="true"), + lxml.etree.Element("Package", name="bar", modified="true"), + lxml.etree.Element("Bogus"), + lxml.etree.Element("Path", name="/bar", modified="true"), + lxml.etree.Element("Service", name="bar")] + + expected_states = dict([(e, t.InstallPath.return_value) + for e in entries if e.tag == "Path"]) + expected_states.update(dict([(e, t.InstallPackage.return_value) + for e in entries if e.tag == "Package"])) + expected_states.update(dict([(e, t.InstallService.return_value) + for e in entries if e.tag == "Service"])) + + actual_states = dict() + t.modified = [] + t.Install(entries, actual_states) + self.assertItemsEqual(t.InstallPath.call_args_list, + [call(e) for e in entries if e.tag == "Path"]) + self.assertItemsEqual(t.InstallPackage.call_args_list, + [call(e) for e in entries if e.tag == "Package"]) + self.assertItemsEqual(t.InstallService.call_args_list, + [call(e) for e in entries if e.tag == "Service"]) + self.assertItemsEqual(actual_states, expected_states) + self.assertItemsEqual(t.modified, + [e for e in entries + if e.get("modified") == "true"]) + + def rest_Remove(self): + pass + + def test_getSupportedEntries(self): + t = self.get_obj() + + def handlesEntry(entry): + return entry.get("handled") == "true" + t.handlesEntry = Mock() + t.handlesEntry.side_effect = handlesEntry + + handled = [] + t.config = lxml.etree.Element("Config") + bundle1 = lxml.etree.SubElement(t.config, "Bundle") + lxml.etree.SubElement(bundle1, "Path", name="/foo") + handled.append(lxml.etree.SubElement(bundle1, "Path", name="/bar", + handled="true")) + bundle2 = lxml.etree.SubElement(t.config, "Bundle") + handled.append(lxml.etree.SubElement(bundle2, "Path", name="/quux", + handled="true")) + lxml.etree.SubElement(bundle2, "Path", name="/baz") + + self.assertItemsEqual(handled, + t.getSupportedEntries()) + + def test_handlesEntry(self): + t = self.get_obj() + handles = t.__handles__ + t.__handles__ = [("Path", "file"), + ("Package", "yum")] + self.assertTrue(t.handlesEntry(lxml.etree.Element("Path", type="file", + name="/foo"))) + self.assertFalse(t.handlesEntry(lxml.etree.Element("Path", + type="permissions", + name="/bar"))) + self.assertFalse(t.handlesEntry(lxml.etree.Element("Bogus", + type="file", + name="/baz"))) + self.assertTrue(t.handlesEntry(lxml.etree.Element("Package", + type="yum", + name="quux"))) + t.__handles__ = handles + + def test_buildModlist(self): + t = self.get_obj() + paths = [] + + t.config = lxml.etree.Element("Config") + bundle1 = lxml.etree.SubElement(t.config, "Bundle") + paths.append(lxml.etree.SubElement(bundle1, "Path", name="/foo")) + lxml.etree.SubElement(bundle1, "Package", name="bar") + paths.append(lxml.etree.SubElement(bundle1, "Path", name="/bar")) + bundle2 = lxml.etree.SubElement(t.config, "Bundle") + paths.append(lxml.etree.SubElement(bundle2, "Path", name="/quux")) + lxml.etree.SubElement(bundle2, "Service", name="baz") + + self.assertItemsEqual([p.get("name") for p in paths], + t.buildModlist()) + + def test_missing_attrs(self): + t = self.get_obj() + req = t.__req__ + t.__req__ = dict(Path=dict(file=["name"], + permissions=["name", "owner", "group"]), + Package=["name"]) + # tuples of <entry>, <return value> + cases = [ + (lxml.etree.Element("Path", name="/foo"), ["type"]), + (lxml.etree.Element("Path", type="file"), ["name"]), + (lxml.etree.Element("Path", type="file", name="/foo"), []), + (lxml.etree.Element("Path", type="permissions", name="/foo"), + ["owner", "group"]), + (lxml.etree.Element("Path", type="permissions", name="/foo", + owner="root", group="root", mode="0644"), []), + (lxml.etree.Element("Package", type="yum"), ["name"]), + (lxml.etree.Element("Package", type="yum", name="/bar"), []), + (lxml.etree.Element("Package", type="apt", name="/bar"), [])] + for entry, expected in cases: + self.assertItemsEqual(t.missing_attrs(entry), expected) + + t.__req__ = req + + def test_canVerify(self): + t = self.get_obj() + entry = Mock() + t._entry_is_complete = Mock() + self.assertEqual(t.canVerify(entry), + t._entry_is_complete.return_value) + t._entry_is_complete.assert_called_with(entry, action="verify") + + def test_FindExtra(self): + t = self.get_obj() + self.assertItemsEqual(t.FindExtra(), []) + + def test_canInstall(self): + t = self.get_obj() + entry = Mock() + t._entry_is_complete = Mock() + self.assertEqual(t.canInstall(entry), + t._entry_is_complete.return_value) + t._entry_is_complete.assert_called_with(entry, action="install") + + def test__entry_is_complete(self): + t = self.get_obj() + t.handlesEntry = Mock() + t.missing_attrs = Mock() + + def reset(): + t.handlesEntry.reset_mock() + t.missing_attrs.reset_mock() + + entry = lxml.etree.Element("Path", name="/test") + + t.handlesEntry.return_value = False + t.missing_attrs.return_value = [] + self.assertFalse(t._entry_is_complete(entry)) + + reset() + t.handlesEntry.return_value = True + t.missing_attrs.return_value = ["type"] + self.assertFalse(t._entry_is_complete(entry)) + + reset() + t.missing_attrs.return_value = [] + self.assertTrue(t._entry_is_complete(entry)) + + reset() + entry.set("failure", "failure") + self.assertFalse(t._entry_is_complete(entry)) + + +class TestPkgTool(TestTool): + test_obj = PkgTool + + def get_obj(self, **kwargs): + @patch("%s.%s.RefreshPackages" % (self.test_obj.__module__, + self.test_obj.__name__), Mock()) + def inner(): + return TestTool.get_obj(self, **kwargs) + + return inner() + + def test_VerifyPackage(self): + pt = self.get_obj() + self.assertRaises(NotImplementedError, + pt.VerifyPackage, Mock(), Mock()) + + def test_Install(self): + pt = self.get_obj() + pt.cmd = Mock() + pt.RefreshPackages = Mock() + pt.VerifyPackage = Mock() + pt._get_package_command = Mock() + pt._get_package_command.side_effect = lambda pkgs: \ + [p.get("name") for p in pkgs] + packages = [lxml.etree.Element("Package", type="echo", name="foo", + version="1.2.3"), + lxml.etree.Element("Package", type="echo", name="bar", + version="any"), + lxml.etree.Element("Package", type="echo", name="baz", + version="2.3.4")] + + def reset(): + pt.cmd.reset_mock() + pt.RefreshPackages.reset_mock() + pt.VerifyPackage.reset_mock() + pt._get_package_command.reset_mock() + pt.modified = [] + + # test single-pass install success + reset() + pt.cmd.run.return_value = (0, '') + states = dict([(p, False) for p in packages]) + pt.Install(packages, states) + pt._get_package_command.assert_called_with(packages) + pt.cmd.run.assert_called_with([p.get("name") for p in packages]) + self.assertItemsEqual(states, + dict([(p, True) for p in packages])) + self.assertItemsEqual(pt.modified, packages) + + # test failed single-pass install + reset() + + def run(cmd): + if "foo" in cmd: + # fail when installing all packages, and when installing foo + return (1, '') + # succeed otherwise + return (0, '') + + pt.VerifyPackage.side_effect = lambda p, m: p.get("name") == "bar" + + pt.cmd.run.side_effect = run + states = dict([(p, False) for p in packages]) + pt.Install(packages, states) + pt._get_package_command.assert_any_call(packages) + for pkg in packages: + pt.VerifyPackage.assert_any_call(pkg, []) + if pkg.get("name") != "bar": + pt._get_package_command.assert_any_call([pkg]) + # pt.cmd.run is called once for all packages, and then once + # for each package that does not verify. "bar" verifies, so + # it's run for foo and baz + self.assertItemsEqual(pt.cmd.run.call_args_list, + [call([p.get("name") for p in packages]), + call(["foo"]), + call(["baz"])]) + pt.RefreshPackages.assert_called_with() + self.assertItemsEqual(states, + dict([(p, p.get("name") != "bar") + for p in packages])) + # bar is modified, because it verifies successfully; baz is + # modified, because it is installed successfully. foo is not + # installed successfully, so is not modified. + self.assertItemsEqual(pt.modified, + [p for p in packages if p.get("name") != "foo"]) + + def test__get_package_command(self): + packages = [lxml.etree.Element("Package", type="test", name="foo", + version="1.2.3"), + lxml.etree.Element("Package", type="test", name="bar", + version="any"), + lxml.etree.Element("Package", type="test", name="baz", + version="2.3.4")] + pt = self.get_obj() + pkgtool = pt.pkgtool + pt.pkgtool = ("install %s", ("%s-%s", ["name", "version"])) + self.assertEqual(pt._get_package_command([ + lxml.etree.Element("Package", type="test", name="foo", + version="1.2.3")]), + "install foo-1.2.3") + self.assertItemsEqual(pt._get_package_command(packages).split(), + ["install", "foo-1.2.3", "bar-any", "baz-2.3.4"]) + + def test_RefreshPackages(self): + pt = self.get_obj() + self.assertRaises(NotImplementedError, pt.RefreshPackages) + + def test_FindExtra(self): + pt = self.get_obj() + pt.getSupportedEntries = Mock() + pt.getSupportedEntries.return_value = [ + lxml.etree.Element("Package", name="foo"), + lxml.etree.Element("Package", name="bar"), + lxml.etree.Element("Package", name="baz")] + pt.installed = dict(foo="1.2.3", + bar="2.3.4", + quux="3.4.5", + xyzzy="4.5.6") + extra = pt.FindExtra() + self.assertEqual(len(extra), 2) + self.assertItemsEqual([e.get("name") for e in extra], + ["quux", "xyzzy"]) + for el in extra: + self.assertEqual(el.tag, "Package") + self.assertEqual(el.get("type"), pt.pkgtype) + + +class TestSvcTool(TestTool): + test_obj = SvcTool + + def test_start_service(self): + st = self.get_obj() + st.get_svc_command = Mock() + st.cmd = MagicMock() + service = lxml.etree.Element("Service", name="foo", type="test") + self.assertEqual(st.start_service(service), + st.cmd.run.return_value[0]) + st.get_svc_command.assert_called_with(service, "start") + st.cmd.run.assert_called_with(st.get_svc_command.return_value) + + def test_stop_service(self): + st = self.get_obj() + st.get_svc_command = Mock() + st.cmd = MagicMock() + service = lxml.etree.Element("Service", name="foo", type="test") + self.assertEqual(st.stop_service(service), + st.cmd.run.return_value[0]) + st.get_svc_command.assert_called_with(service, "stop") + st.cmd.run.assert_called_with(st.get_svc_command.return_value) + + def test_restart_service(self): + st = self.get_obj() + st.get_svc_command = Mock() + st.cmd = MagicMock() + + def reset(): + st.get_svc_command.reset_mock() + st.cmd.reset_mock() + + service = lxml.etree.Element("Service", name="foo", type="test") + self.assertEqual(st.restart_service(service), + st.cmd.run.return_value[0]) + st.get_svc_command.assert_called_with(service, "restart") + st.cmd.run.assert_called_with(st.get_svc_command.return_value) + + reset() + service.set('target', 'reload') + self.assertEqual(st.restart_service(service), + st.cmd.run.return_value[0]) + st.get_svc_command.assert_called_with(service, "reload") + st.cmd.run.assert_called_with(st.get_svc_command.return_value) + + def test_check_service(self): + st = self.get_obj() + st.get_svc_command = Mock() + st.cmd = MagicMock() + service = lxml.etree.Element("Service", name="foo", type="test") + + def reset(): + st.get_svc_command.reset_mock() + st.cmd.reset_mock() + + st.cmd.run.return_value = (0, '') + self.assertEqual(st.check_service(service), True) + st.get_svc_command.assert_called_with(service, "status") + st.cmd.run.assert_called_with(st.get_svc_command.return_value) + + reset() + st.cmd.run.return_value = (11, '') + self.assertEqual(st.check_service(service), False) + st.get_svc_command.assert_called_with(service, "status") + st.cmd.run.assert_called_with(st.get_svc_command.return_value) + + def test_Remove(self): + st = self.get_obj() + st.InstallService = Mock() + services = [lxml.etree.Element("Service", type="test", name="foo"), + lxml.etree.Element("Service", type="test", name="bar", + status="on")] + st.Remove(services) + self.assertItemsEqual(st.InstallService.call_args_list, + [call(e) for e in services]) + for entry in services: + self.assertEqual(entry.get("status"), "off") + + @patch("Bcfg2.Client.prompt") + def test_BundleUpdated(self, mock_prompt): + st = self.get_obj(setup=dict(interactive=False, + servicemode='default')) + st.handlesEntry = Mock() + st.handlesEntry.side_effect = lambda e: e.tag == "Service" + st.stop_service = Mock() + st.stop_service.return_value = 0 + st.restart_service = Mock() + st.restart_service.side_effect = lambda e: \ + int(e.get("name") == "failed") + + def reset(): + st.handlesEntry.reset_mock() + st.stop_service.reset_mock() + st.restart_service.reset_mock() + mock_prompt.reset_mock() + st.restarted = [] + + norestart = lxml.etree.Element("Service", type="test", + name="norestart", restart="false") + interactive = lxml.etree.Element("Service", type="test", + name="interactive", status="on", + restart="interactive") + interactive2 = lxml.etree.Element("Service", type="test", + name="interactive2", status="on", + restart="interactive") + stop = lxml.etree.Element("Service", type="test", name="stop", + status="off") + restart = lxml.etree.Element("Service", type="test", name="restart", + status="on") + duplicate = lxml.etree.Element("Service", type="test", name="restart", + status="on") + failed = lxml.etree.Element("Service", type="test", name="failed", + status="on") + unhandled = lxml.etree.Element("Path", type="file", name="/unhandled") + services = [norestart, interactive, interactive2, stop, restart, + duplicate, failed] + entries = services + [unhandled] + bundle = lxml.etree.Element("Bundle") + bundle.extend(entries) + + # test in non-interactive mode + reset() + states = dict() + st.BundleUpdated(bundle, states) + self.assertItemsEqual(st.handlesEntry.call_args_list, + [call(e) for e in entries]) + st.stop_service.assert_called_with(stop) + self.assertItemsEqual(st.restart_service.call_args_list, + [call(restart), call(failed)]) + self.assertItemsEqual(st.restarted, [restart.get("name")]) + self.assertFalse(mock_prompt.called) + + # test in interactive mode + reset() + mock_prompt.side_effect = lambda p: "interactive2" not in p + st.setup['interactive'] = True + states = dict() + st.BundleUpdated(bundle, states) + self.assertItemsEqual(st.handlesEntry.call_args_list, + [call(e) for e in entries]) + st.stop_service.assert_called_with(stop) + self.assertItemsEqual(st.restart_service.call_args_list, + [call(restart), call(failed), call(interactive)]) + self.assertItemsEqual(st.restarted, [restart.get("name"), + interactive.get("name")]) + self.assertEqual(len(mock_prompt.call_args_list), 4) + + # test in build mode + reset() + st.setup['interactive'] = False + st.setup['servicemode'] = 'build' + states = dict() + st.BundleUpdated(bundle, states) + self.assertItemsEqual(st.handlesEntry.call_args_list, + [call(e) for e in entries]) + self.assertItemsEqual(st.stop_service.call_args_list, + [call(restart), call(duplicate), call(failed), + call(stop)]) + self.assertFalse(mock_prompt.called) + self.assertFalse(st.restart_service.called) + self.assertItemsEqual(st.restarted, []) + + @patch("Bcfg2.Client.Tools.Tool.Install") + def test_Install(self, mock_Install): + install = [lxml.etree.Element("Service", type="test", name="foo")] + services = install + [lxml.etree.Element("Service", type="test", + name="bar", install="false")] + st = self.get_obj() + states = Mock() + self.assertEqual(st.Install(services, states), + mock_Install.return_value) + mock_Install.assert_called_with(st, install, states) + + def test_InstallService(self): + st = self.get_obj() + self.assertRaises(NotImplementedError, st.InstallService, Mock()) |