summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorChris St. Pierre <chris.a.st.pierre@gmail.com>2013-02-20 07:52:00 -0500
committerChris St. Pierre <chris.a.st.pierre@gmail.com>2013-02-20 07:52:07 -0500
commitbe0de88922a58504c655361970378375426b5acc (patch)
treeaf61cafdadf2634cbdad83ac0a5d1a9c5934a1ae
parent1f572cc7b6ce8462128fc1093458917af01f71ee (diff)
downloadbcfg2-be0de88922a58504c655361970378375426b5acc.tar.gz
bcfg2-be0de88922a58504c655361970378375426b5acc.tar.bz2
bcfg2-be0de88922a58504c655361970378375426b5acc.zip
wrote unit tests for base client Tool classes
-rw-r--r--src/lib/Bcfg2/Client/Frame.py31
-rw-r--r--src/lib/Bcfg2/Client/Tools/POSIX/Hardlink.py1
-rw-r--r--src/lib/Bcfg2/Client/Tools/__init__.py110
-rw-r--r--src/lib/Bcfg2/Client/__init__.py28
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestDevice.py56
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestDirectory.py111
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestFile.py300
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestHardlink.py62
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestNonexistent.py28
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestPermissions.py1
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/TestSymlink.py70
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/Test__init.py127
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIX/Testbase.py625
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIXUsers.py17
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/Test_init.py648
15 files changed, 1417 insertions, 798 deletions
diff --git a/src/lib/Bcfg2/Client/Frame.py b/src/lib/Bcfg2/Client/Frame.py
index a95c0a7a6..637a916d6 100644
--- a/src/lib/Bcfg2/Client/Frame.py
+++ b/src/lib/Bcfg2/Client/Frame.py
@@ -1,14 +1,12 @@
""" Frame is the Client Framework that verifies and installs entries,
and generates statistics. """
-import os
-import sys
import time
-import select
import fnmatch
import logging
import Bcfg2.Client.Tools
-from Bcfg2.Compat import input, any, all # pylint: disable=W0622
+from Bcfg2.Client import prompt
+from Bcfg2.Compat import any, all # pylint: disable=W0622
def cmpent(ent1, ent2):
@@ -154,7 +152,7 @@ class Frame(object):
for entry in multi:
self.logger.debug(entry)
- def promptFilter(self, prompt, entries):
+ def promptFilter(self, msg, entries):
"""Filter a supplied list based on user input."""
ret = []
entries.sort(cmpent)
@@ -165,20 +163,9 @@ class Frame(object):
if 'qtext' in entry.attrib:
iprompt = entry.get('qtext')
else:
- iprompt = prompt % (entry.tag, entry.get('name'))
- # flush input buffer
- while len(select.select([sys.stdin.fileno()], [], [], 0.0)[0]) > 0:
- os.read(sys.stdin.fileno(), 4096)
- try:
- ans = input(iprompt.encode(sys.stdout.encoding, 'replace'))
- if ans in ['y', 'Y']:
- ret.append(entry)
- except EOFError:
- # python 2.4.3 on CentOS doesn't like ^C for some reason
- break
- except:
- print("Error while reading input")
- continue
+ iprompt = msg % (entry.tag, entry.get('name'))
+ if prompt(iprompt):
+ ret.append(entry)
return ret
def __getattr__(self, name):
@@ -281,7 +268,7 @@ class Frame(object):
def Decide(self): # pylint: disable=R0912
"""Set self.whitelist based on user interaction."""
- prompt = "Install %s: %s? (y/N): "
+ iprompt = "Install %s: %s? (y/N): "
rprompt = "Remove %s: %s? (y/N): "
if self.setup['remove']:
if self.setup['remove'] == 'all':
@@ -354,7 +341,7 @@ class Frame(object):
(bmodified or a.get('when') == 'always'))]
# now we process all "always actions"
if self.setup['interactive']:
- self.promptFilter(prompt, actions)
+ self.promptFilter(iprompt, actions)
self.DispatchInstallCalls(actions)
# need to test to fail entries in whitelist
@@ -377,7 +364,7 @@ class Frame(object):
if b.get("name")))
if self.setup['interactive']:
- self.whitelist = self.promptFilter(prompt, self.whitelist)
+ self.whitelist = self.promptFilter(iprompt, self.whitelist)
self.removal = self.promptFilter(rprompt, self.removal)
for entry in candidates:
diff --git a/src/lib/Bcfg2/Client/Tools/POSIX/Hardlink.py b/src/lib/Bcfg2/Client/Tools/POSIX/Hardlink.py
index 896ca5f49..64a0b1e15 100644
--- a/src/lib/Bcfg2/Client/Tools/POSIX/Hardlink.py
+++ b/src/lib/Bcfg2/Client/Tools/POSIX/Hardlink.py
@@ -12,5 +12,4 @@ class POSIXHardlink(POSIXLinkTool):
return os.path.samefile(entry.get('name'), entry.get('to'))
def _link(self, entry):
- ## TODO: set permissions
return os.link(entry.get('to'), entry.get('name'))
diff --git a/src/lib/Bcfg2/Client/Tools/__init__.py b/src/lib/Bcfg2/Client/Tools/__init__.py
index c0dd60c1e..08dc09294 100644
--- a/src/lib/Bcfg2/Client/Tools/__init__.py
+++ b/src/lib/Bcfg2/Client/Tools/__init__.py
@@ -3,10 +3,10 @@
import os
import sys
import stat
-import select
from subprocess import Popen, PIPE
+import Bcfg2.Client
import Bcfg2.Client.XML
-from Bcfg2.Compat import input, walk_packages # pylint: disable=W0622
+from Bcfg2.Compat import walk_packages # pylint: disable=W0622
__all__ = [m[1] for m in walk_packages(path=__path__)]
@@ -155,25 +155,34 @@ class Tool(object):
#: A list of all entries handled by this tool
self.handled = []
- for struct in config:
+ self._analyze_config()
+ self._check_execs()
+
+ def _analyze_config(self):
+ """ Analyze the config at tool initialization-time for
+ important and handled entries """
+ for struct in self.config:
for entry in struct:
if (entry.tag == 'Path' and
entry.get('important', 'false').lower() == 'true'):
self.__important__.append(entry.get('name'))
- if self.handlesEntry(entry):
- self.handled.append(entry)
+ self.handled = self.getSupportedEntries()
+
+ def _check_execs(self):
+ """ Check all executables used by this tool to ensure that
+ they exist and are executable """
for filename in self.__execs__:
try:
mode = stat.S_IMODE(os.stat(filename)[stat.ST_MODE])
- if mode & stat.S_IEXEC != stat.S_IEXEC:
- raise ToolInstantiationError("%s: %s not executable" %
- (self.name, filename))
except OSError:
raise ToolInstantiationError(sys.exc_info()[1])
except:
raise ToolInstantiationError("%s: Failed to stat %s" %
- (self.name, filename),
- exc_info=1)
+ (self.name, filename))
+ if not mode & stat.S_IEXEC:
+ raise ToolInstantiationError("%s: %s not executable" %
+ (self.name, filename))
+
def BundleUpdated(self, bundle, states): # pylint: disable=W0613
""" Callback that is invoked when a bundle has been updated.
@@ -227,11 +236,13 @@ class Tool(object):
if self.canVerify(entry):
try:
func = getattr(self, "Verify%s" % entry.tag)
- states[entry] = func(entry, mods)
except AttributeError:
self.logger.error("%s: Cannot verify %s entries" %
(self.name, entry.tag))
- except:
+ continue
+ try:
+ states[entry] = func(entry, mods)
+ except: # pylint: disable=W0702
self.logger.error("%s: Unexpected failure verifying %s"
% (self.name,
self.primarykey(entry)),
@@ -255,14 +266,16 @@ class Tool(object):
:returns: None """
for entry in entries:
try:
- func = getattr(self, "Install%s" % (entry.tag))
- states[entry] = func(entry)
- if states[entry]:
- self.modified.append(entry)
+ func = getattr(self, "Install%s" % entry.tag)
except AttributeError:
self.logger.error("%s: Cannot install %s entries" %
(self.name, entry.tag))
- except:
+ continue
+ try:
+ states[entry] = func(entry)
+ if states[entry]:
+ self.modified.append(entry)
+ except: # pylint: disable=W0702
self.logger.error("%s: Unexpected failure installing %s" %
(self.name, self.primarykey(entry)),
exc_info=1)
@@ -451,6 +464,19 @@ class PkgTool(Tool):
"""
raise NotImplementedError
+ def _get_package_command(self, packages):
+ """ Get the command to install the given list of packages.
+
+ :param packages: The Package entries to install
+ :type packages: list of lxml.etree._Element
+ :returns: string - the command to run
+ """
+ pkgargs = " ".join(self.pkgtool[1][0] %
+ tuple(pkg.get(field)
+ for field in self.pkgtool[1][1])
+ for pkg in packages)
+ return self.pkgtool[0] % pkgargs
+
def Install(self, packages, states):
""" Run a one-pass install where all required packages are
installed with a single command, followed by single package
@@ -464,14 +490,10 @@ class PkgTool(Tool):
self.logger.info("Trying single pass package install for pkgtype %s" %
self.pkgtype)
- data = [tuple([pkg.get(field) for field in self.pkgtool[1][1]])
- for pkg in packages]
- pkgargs = " ".join([self.pkgtool[1][0] % datum for datum in data])
-
- self.logger.debug("Installing packages: %s" % pkgargs)
- self.logger.debug("Running command: %s" % (self.pkgtool[0] % pkgargs))
+ pkgcmd = self._get_package_command(packages)
+ self.logger.debug("Running command: %s" % pkgcmd)
- cmdrc = self.cmd.run(self.pkgtool[0] % pkgargs)[0]
+ cmdrc = self.cmd.run(pkgcmd)[0]
if cmdrc == 0:
self.logger.info("Single Pass Succeded")
# set all package states to true and flush workqueues
@@ -481,7 +503,7 @@ class PkgTool(Tool):
and entry.get('type') == self.pkgtype
and entry.get('name') in pkgnames):
self.logger.debug('Setting state to true for pkg %s' %
- (entry.get('name')))
+ entry.get('name'))
states[entry] = True
self.RefreshPackages()
else:
@@ -497,19 +519,13 @@ class PkgTool(Tool):
else:
self.logger.info("Installing pkg %s version %s" %
(pkg.get('name'), pkg.get('version')))
- cmdrc = self.cmd.run(
- self.pkgtool[0] %
- (self.pkgtool[1][0] %
- tuple([pkg.get(field)
- for field in self.pkgtool[1][1]])))
- if cmdrc[0] == 0:
+ if self.cmd.run(self._get_package_command([pkg]))[0] == 0:
states[pkg] = True
else:
self.logger.error("Failed to install package %s" %
- (pkg.get('name')))
+ pkg.get('name'))
self.RefreshPackages()
- for entry in [ent for ent in packages if states[ent]]:
- self.modified.append(entry)
+ self.modified.extend(entry for entry in packages if states[entry])
def RefreshPackages(self):
""" Refresh the internal representation of the package
@@ -603,11 +619,13 @@ class SvcTool(Tool):
if self.setup['servicemode'] == 'disabled':
return
- for entry in [ent for ent in bundle if self.handlesEntry(ent)]:
- restart = entry.get("restart", "true")
- if (restart.lower() == "false" or
- (restart.lower() == "interactive" and
- not self.setup['interactive'])):
+ for entry in bundle:
+ if not self.handlesEntry(entry):
+ continue
+
+ restart = entry.get("restart", "true").lower()
+ if (restart == "false" or
+ (restart == "interactive" and not self.setup['interactive'])):
continue
rv = None
@@ -616,14 +634,8 @@ class SvcTool(Tool):
rv = self.stop_service(entry)
elif entry.get('name') not in self.restarted:
if self.setup['interactive']:
- prompt = ('Restart service %s?: (y/N): ' %
- entry.get('name'))
- # flush input buffer
- while len(select.select([sys.stdin.fileno()], [], [],
- 0.0)[0]) > 0:
- os.read(sys.stdin.fileno(), 4096)
- ans = input(prompt)
- if ans not in ['y', 'Y']:
+ if not Bcfg2.Client.prompt('Restart service %s? (y/N) '
+ % entry.get('name')):
continue
rv = self.restart_service(entry)
if not rv:
@@ -639,8 +651,8 @@ class SvcTool(Tool):
install_entries = []
for entry in entries:
if entry.get('install', 'true').lower() == 'false':
- self.logger.info("Service %s installation is false. Skipping "
- "installation." % (entry.get('name')))
+ self.logger.info("Installation is false for %s:%s, skipping" %
+ (entry.tag, entry.get('name')))
else:
install_entries.append(entry)
return Tool.Install(self, install_entries, states)
diff --git a/src/lib/Bcfg2/Client/__init__.py b/src/lib/Bcfg2/Client/__init__.py
index c03021f14..dd5ae1e83 100644
--- a/src/lib/Bcfg2/Client/__init__.py
+++ b/src/lib/Bcfg2/Client/__init__.py
@@ -1,3 +1,31 @@
"""This contains all Bcfg2 Client modules"""
__all__ = ["Frame", "Tools", "XML", "Client"]
+
+import os
+import sys
+import select
+from Bcfg2.Compat import input # pylint: disable=W0622
+
+
+def prompt(msg):
+ """ Helper to give a yes/no prompt to the user. Flushes input
+ buffers, handles exceptions, etc. Returns True if the user
+ answers in the affirmative, False otherwise.
+
+ :param msg: The message to show to the user. The message is not
+ altered in any way for display; i.e., it should
+ contain "[y/N]" if desired, etc.
+ :type msg: string
+ :returns: bool - True if yes, False if no """
+ while len(select.select([sys.stdin.fileno()], [], [], 0.0)[0]) > 0:
+ os.read(sys.stdin.fileno(), 4096)
+ try:
+ ans = input(msg.encode(sys.stdout.encoding, 'replace'))
+ return ans in ['y', 'Y']
+ except EOFError:
+ # python 2.4.3 on CentOS doesn't like ^C for some reason
+ return False
+ except:
+ print("Error while reading input: %s" % sys.exc_info()[1])
+ return False
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())