summaryrefslogtreecommitdiffstats
path: root/testsuite
diff options
context:
space:
mode:
authorChris St. Pierre <chris.a.st.pierre@gmail.com>2013-08-08 15:35:23 -0400
committerChris St. Pierre <chris.a.st.pierre@gmail.com>2013-08-09 10:26:31 -0400
commit48dea2ea0ce93b80484abac6875da4fdf5a20ce7 (patch)
treeee90f7fb0c593420a6c87801cd5aa4083a6568a9 /testsuite
parentb228b295546aa82d2dfe588e3e52817e392ddb6a (diff)
downloadbcfg2-48dea2ea0ce93b80484abac6875da4fdf5a20ce7.tar.gz
bcfg2-48dea2ea0ce93b80484abac6875da4fdf5a20ce7.tar.bz2
bcfg2-48dea2ea0ce93b80484abac6875da4fdf5a20ce7.zip
testsuite: fixed unit tests for server plugins
Diffstat (limited to 'testsuite')
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIXUsers.py5
-rw-r--r--testsuite/Testsrc/Testlib/TestClient/TestTools/Test_init.py6
-rw-r--r--testsuite/Testsrc/Testlib/TestLogger.py4
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testbase.py6
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py60
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgAuthorizedKeysGenerator.py28
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgCheetahGenerator.py14
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgEncryptedGenerator.py2
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgEncryptedGenshiGenerator.py2
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgGenshiGenerator.py47
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgInfoXML.py4
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgPrivateKeyCreator.py129
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/Test_init.py100
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestMetadata.py29
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py164
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProperties.py49
-rw-r--r--testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestRules.py12
-rw-r--r--testsuite/common.py33
18 files changed, 322 insertions, 372 deletions
diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIXUsers.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIXUsers.py
index 3d2e762ff..d9d8a26c6 100644
--- a/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIXUsers.py
+++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/TestPOSIXUsers.py
@@ -24,11 +24,14 @@ from TestTools.Test_init import TestTool
class TestPOSIXUsers(TestTool):
test_obj = POSIXUsers
- def get_obj(self, config=None):
+ def setUp(self):
+ TestTool.setUp(self)
set_setup_default('uid_whitelist')
set_setup_default('uid_blacklist')
set_setup_default('gid_whitelist')
set_setup_default('gid_blacklist')
+
+ def get_obj(self, config=None):
return TestTool.get_obj(self, config)
@patch("pwd.getpwall")
diff --git a/testsuite/Testsrc/Testlib/TestClient/TestTools/Test_init.py b/testsuite/Testsrc/Testlib/TestClient/TestTools/Test_init.py
index cbf8e4911..1bf073f81 100644
--- a/testsuite/Testsrc/Testlib/TestClient/TestTools/Test_init.py
+++ b/testsuite/Testsrc/Testlib/TestClient/TestTools/Test_init.py
@@ -21,11 +21,13 @@ from common import *
class TestTool(Bcfg2TestCase):
test_obj = Tool
+ def setUp(self):
+ set_setup_default('command_timeout')
+ set_setup_default('interactive', False)
+
def get_obj(self, config=None):
if config is None:
config = lxml.etree.Element("Configuration")
- set_setup_default('command_timeout')
- set_setup_default('interactive', False)
execs = self.test_obj.__execs__
self.test_obj.__execs__ = []
diff --git a/testsuite/Testsrc/Testlib/TestLogger.py b/testsuite/Testsrc/Testlib/TestLogger.py
index 0da39deb2..1baea2f35 100644
--- a/testsuite/Testsrc/Testlib/TestLogger.py
+++ b/testsuite/Testsrc/Testlib/TestLogger.py
@@ -19,8 +19,10 @@ from common import *
class TestDebuggable(Bcfg2TestCase):
test_obj = Debuggable
- def get_obj(self):
+ def setUp(self):
set_setup_default('debug', False)
+
+ def get_obj(self):
return self.test_obj()
def test__init(self):
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testbase.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testbase.py
index db126b3a1..9b55b3ae8 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testbase.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testbase.py
@@ -19,10 +19,14 @@ from TestLogger import TestDebuggable
class TestPlugin(TestDebuggable):
test_obj = Plugin
+ def setUp(self):
+ TestDebuggable.setUp(self)
+ set_setup_default("filemonitor", MagicMock())
+
def get_obj(self, core=None):
if core is None:
core = Mock()
- set_setup_default("debug", False)
+
@patchIf(not isinstance(os.makedirs, Mock), "os.makedirs", Mock())
def inner():
return self.test_obj(core, datastore)
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py
index 3439d475f..7006e29e3 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugin/Testhelpers.py
@@ -51,22 +51,28 @@ class TestFunctions(Bcfg2TestCase):
class TestDatabaseBacked(TestPlugin):
test_obj = DatabaseBacked
+ def setUp(self):
+ TestPlugin.setUp(self)
+ set_setup_default("%s_db" % self.test_obj.__name__.lower(), False)
+
@skipUnless(HAS_DJANGO, "Django not found")
def test__use_db(self):
- db = self.get_obj()
+ core = Mock()
+ db = self.get_obj(core=core)
+ attr = "%s_db" % self.test_obj.__name__.lower()
- Bcfg2.Options.setup.databasebacked_db = True
+ db.core.database_available = True
+ setattr(Bcfg2.Options.setup, attr, True)
self.assertTrue(db._use_db)
- Bcfg2.Options.setup.databasebacked_db = False
+ setattr(Bcfg2.Options.setup, attr, False)
self.assertFalse(db._use_db)
- Bcfg2.Server.Plugin.helpers.HAS_DJANGO = False
+ db.core.database_available = False
self.assertFalse(db._use_db)
- Bcfg2.Options.setup.databasebacked_db = False
+ setattr(Bcfg2.Options.setup, attr, True)
self.assertFalse(db._use_db)
- Bcfg2.Server.Plugin.helpers.HAS_DJANGO = True
class TestPluginDatabaseModel(Bcfg2TestCase):
@@ -74,14 +80,17 @@ class TestPluginDatabaseModel(Bcfg2TestCase):
pass
-class TestFileBacked(Bcfg2TestCase):
+class TestFileBacked(TestDebuggable):
test_obj = FileBacked
path = os.path.join(datastore, "test")
+ def setUp(self):
+ TestDebuggable.setUp(self)
+ set_setup_default("filemonitor", MagicMock())
+
def get_obj(self, path=None):
if path is None:
path = self.path
- set_setup_default("filemonitor", MagicMock())
return self.test_obj(path)
@patch("%s.open" % builtins)
@@ -110,7 +119,7 @@ class TestFileBacked(Bcfg2TestCase):
self.assertFalse(fb.Index.called)
-class TestDirectoryBacked(Bcfg2TestCase):
+class TestDirectoryBacked(TestDebuggable):
test_obj = DirectoryBacked
testpaths = {1: '',
2: '/foo',
@@ -124,6 +133,10 @@ class TestDirectoryBacked(Bcfg2TestCase):
badevents = [] # DirectoryBacked handles all files, so there's no
# such thing as a bad event
+ def setUp(self):
+ TestDebuggable.setUp(self)
+ set_setup_default("filemonitor", MagicMock())
+
def test_child_interface(self):
""" ensure that the child object has the correct interface """
self.assertTrue(hasattr(self.test_obj.__child__, "HandleEvent"))
@@ -133,7 +146,6 @@ class TestDirectoryBacked(Bcfg2TestCase):
if fam is None:
fam = Mock()
- set_setup_default("filemonitor", MagicMock())
@patch("%s.%s.add_directory_monitor" % (self.test_obj.__module__,
self.test_obj.__name__),
Mock())
@@ -378,11 +390,13 @@ class TestXMLFileBacked(TestFileBacked):
should_monitor = None
path = os.path.join(datastore, "test", "test1.xml")
+ def setUp(self):
+ TestFileBacked.setUp(self)
+ set_setup_default("encoding", 'utf-8')
+
def get_obj(self, path=None, should_monitor=False):
if path is None:
path = self.path
- set_setup_default("filemonitor", MagicMock())
- set_setup_default("encoding", 'utf-8')
@patchIf(not isinstance(os.path.exists, Mock),
"os.path.exists", Mock())
@@ -1158,6 +1172,11 @@ class TestXMLDirectoryBacked(TestDirectoryBacked):
class TestPrioDir(TestPlugin, TestGenerator, TestXMLDirectoryBacked):
test_obj = PrioDir
+ def setUp(self):
+ TestPlugin.setUp(self)
+ TestGenerator.setUp(self)
+ TestXMLDirectoryBacked.setUp(self)
+
def get_obj(self, core=None):
if core is None:
core = Mock()
@@ -1331,10 +1350,14 @@ class TestSpecificity(Bcfg2TestCase):
self.assertGreaterEqual(specs[j], specs[i])
-class TestSpecificData(Bcfg2TestCase):
+class TestSpecificData(TestDebuggable):
test_obj = SpecificData
path = os.path.join(datastore, "test.txt")
+ def setUp(self):
+ TestDebuggable.setUp(self)
+ set_setup_default("encoding", "utf-8")
+
def get_obj(self, name=None, specific=None):
if name is None:
name = self.path
@@ -1382,7 +1405,8 @@ class TestEntrySet(TestDebuggable):
ignore = ["foo~", ".#foo", ".foo.swp", ".foo.swx",
"test.txt.genshi_include", "test.G_foo.genshi_include"]
- def get_obj(self, basename="test", path=datastore, entry_type=MagicMock()):
+ def setUp(self):
+ TestDebuggable.setUp(self)
set_setup_default("default_owner")
set_setup_default("default_group")
set_setup_default("default_mode")
@@ -1390,6 +1414,8 @@ class TestEntrySet(TestDebuggable):
set_setup_default("default_important", False)
set_setup_default("default_paranoid", False)
set_setup_default("default_sensitive", False)
+
+ def get_obj(self, basename="test", path=datastore, entry_type=MagicMock()):
return self.test_obj(basename, path, entry_type)
def test__init(self):
@@ -1742,10 +1768,14 @@ class TestEntrySet(TestDebuggable):
class TestGroupSpool(TestPlugin, TestGenerator):
test_obj = GroupSpool
+ def setUp(self):
+ TestPlugin.setUp(self)
+ TestGenerator.setUp(self)
+ set_setup_default("encoding", "utf-8")
+
def get_obj(self, core=None):
if core is None:
core = MagicMock()
- set_setup_default("encoding", "utf-8")
@patch("%s.%s.AddDirectoryMonitor" % (self.test_obj.__module__,
self.test_obj.__name__),
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgAuthorizedKeysGenerator.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgAuthorizedKeysGenerator.py
index b77d52033..e5cef8fa2 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgAuthorizedKeysGenerator.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgAuthorizedKeysGenerator.py
@@ -23,6 +23,10 @@ class TestCfgAuthorizedKeysGenerator(TestCfgGenerator, TestStructFile):
test_obj = CfgAuthorizedKeysGenerator
should_monitor = False
+ def setUp(self):
+ TestCfgGenerator.setUp(self)
+ TestStructFile.setUp(self)
+
def get_obj(self, name=None, core=None, fam=None):
if name is None:
name = self.path
@@ -40,31 +44,9 @@ class TestCfgAuthorizedKeysGenerator(TestCfgGenerator, TestStructFile):
mock_HandleEvent.assert_called_with(akg, evt)
mock_handle_event.assert_called_with(akg, evt)
- def test_category(self):
- akg = self.get_obj()
- akg.setup = Mock()
- akg.setup.cfp.has_section.return_value = False
- akg.setup.cfp.has_option.return_value = False
-
- self.assertIsNone(akg.category)
- akg.setup.cfp.has_section.assert_called_with("sshkeys")
-
- akg.setup.reset_mock()
- akg.setup.cfp.has_section.return_value = True
- self.assertIsNone(akg.category)
- akg.setup.cfp.has_section.assert_called_with("sshkeys")
- akg.setup.cfp.has_option.assert_called_with("sshkeys", "category")
-
- akg.setup.reset_mock()
- akg.setup.cfp.has_option.return_value = True
- self.assertEqual(akg.category, akg.setup.cfp.get.return_value)
- akg.setup.cfp.has_section.assert_called_with("sshkeys")
- akg.setup.cfp.has_option.assert_called_with("sshkeys", "category")
- akg.setup.cfp.get.assert_called_with("sshkeys", "category")
-
@patch("Bcfg2.Server.Plugins.Cfg.CfgAuthorizedKeysGenerator.ClientMetadata")
- @patch("Bcfg2.Server.Plugins.Cfg.CfgAuthorizedKeysGenerator.CfgAuthorizedKeysGenerator.category", "category")
def test_get_data(self, mock_ClientMetadata):
+ Bcfg2.Options.setup.sshkeys_category = "category"
akg = self.get_obj()
akg.XMLMatch = Mock()
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgCheetahGenerator.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgCheetahGenerator.py
index 31227329c..e1ffa7272 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgCheetahGenerator.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgCheetahGenerator.py
@@ -23,25 +23,25 @@ if HAS_CHEETAH or can_skip:
@skipUnless(HAS_CHEETAH, "Cheetah libraries not found, skipping")
def setUp(self):
- pass
+ TestCfgGenerator.setUp(self)
+ set_setup_default("repository", datastore)
@patch("Bcfg2.Server.Plugins.Cfg.CfgCheetahGenerator.Template")
def test_get_data(self, mock_Template):
- ccg = self.get_obj(encoding='UTF-8')
+ ccg = self.get_obj()
ccg.data = "data"
entry = lxml.etree.Element("Path", name="/test.txt")
metadata = Mock()
- ccg.setup = MagicMock()
self.assertEqual(ccg.get_data(entry, metadata),
mock_Template.return_value.respond.return_value)
- ccg.setup.__getitem__.assert_called_with("repo")
- mock_Template.assert_called_with("data".decode(ccg.encoding),
- compilerSettings=ccg.settings)
+ mock_Template.assert_called_with(
+ "data".decode(Bcfg2.Options.setup.encoding),
+ compilerSettings=ccg.settings)
tmpl = mock_Template.return_value
tmpl.respond.assert_called_with()
self.assertEqual(tmpl.metadata, metadata)
self.assertEqual(tmpl.name, entry.get("name"))
self.assertEqual(tmpl.path, entry.get("name"))
self.assertEqual(tmpl.source_path, ccg.name)
- self.assertEqual(tmpl.repo, ccg.setup.__getitem__.return_value)
+ self.assertEqual(tmpl.repo, datastore)
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgEncryptedGenerator.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgEncryptedGenerator.py
index f8e9b1990..5409cf863 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgEncryptedGenerator.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgEncryptedGenerator.py
@@ -24,7 +24,7 @@ if can_skip or HAS_CRYPTO:
@skipUnless(HAS_CRYPTO, "Encryption libraries not found, skipping")
def setUp(self):
- pass
+ TestCfgGenerator.setUp(self)
@patchIf(HAS_CRYPTO,
"Bcfg2.Server.Plugins.Cfg.CfgEncryptedGenerator.bruteforce_decrypt")
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgEncryptedGenshiGenerator.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgEncryptedGenshiGenerator.py
index 330779c99..25d2fb83b 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgEncryptedGenshiGenerator.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgEncryptedGenshiGenerator.py
@@ -24,4 +24,4 @@ if can_skip or HAS_CRYPTO:
@skipUnless(HAS_CRYPTO, "Encryption libraries not found, skipping")
def setUp(self):
- pass
+ TestCfgGenshiGenerator.setUp(self)
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgGenshiGenerator.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgGenshiGenerator.py
index b73670fb7..9b6e7fe88 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgGenshiGenerator.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgGenshiGenerator.py
@@ -22,6 +22,10 @@ from TestServer.TestPlugins.TestCfg.Test_init import TestCfgGenerator
class TestCfgGenshiGenerator(TestCfgGenerator):
test_obj = CfgGenshiGenerator
+ def setUp(self):
+ TestCfgGenerator.setUp(self)
+ set_setup_default("repository", datastore)
+
def test__init(self):
TestCfgGenerator.test__init(self)
cgg = self.get_obj()
@@ -30,7 +34,6 @@ class TestCfgGenshiGenerator(TestCfgGenerator):
def test_get_data(self):
cgg = self.get_obj()
cgg._handle_genshi_exception = Mock()
- cgg.setup = MagicMock()
cgg.template = Mock()
fltr = Mock()
cgg.template.generate.return_value = fltr
@@ -39,26 +42,25 @@ class TestCfgGenshiGenerator(TestCfgGenerator):
entry = lxml.etree.Element("Path", name="/test.txt")
metadata = Mock()
-
def reset():
cgg.template.reset_mock()
cgg._handle_genshi_exception.reset_mock()
- cgg.setup.reset_mock()
template_vars = dict(
name=entry.get("name"),
metadata=metadata,
path=cgg.name,
source_path=cgg.name,
- repo=cgg.setup.__getitem__.return_value)
+ repo=datastore)
self.assertEqual(cgg.get_data(entry, metadata),
stream.render.return_value)
cgg.template.generate.assert_called_with(**template_vars)
- cgg.setup.__getitem__.assert_called_with("repo")
fltr.filter.assert_called_with(removecomment)
- stream.render.assert_called_with("text", encoding=cgg.encoding,
- strip_whitespace=False)
+ stream.render.assert_called_with(
+ "text",
+ encoding=Bcfg2.Options.setup.encoding,
+ strip_whitespace=False)
reset()
def render(fmt, **kwargs):
@@ -68,21 +70,22 @@ class TestCfgGenshiGenerator(TestCfgGenerator):
self.assertEqual(cgg.get_data(entry, metadata),
stream.render.return_value)
cgg.template.generate.assert_called_with(**template_vars)
- cgg.setup.__getitem__.assert_called_with("repo")
fltr.filter.assert_called_with(removecomment)
self.assertEqual(stream.render.call_args_list,
- [call("text", encoding=cgg.encoding,
- strip_whitespace=False),
- call("text", encoding=cgg.encoding)])
+ [call("text",
+ encoding=Bcfg2.Options.setup.encoding,
+ strip_whitespace=False),
+ call("text",
+ encoding=Bcfg2.Options.setup.encoding)])
reset()
stream.render.side_effect = UndefinedError("test")
self.assertRaises(UndefinedError,
cgg.get_data, entry, metadata)
cgg.template.generate.assert_called_with(**template_vars)
- cgg.setup.__getitem__.assert_called_with("repo")
fltr.filter.assert_called_with(removecomment)
- stream.render.assert_called_with("text", encoding=cgg.encoding,
+ stream.render.assert_called_with("text",
+ encoding=Bcfg2.Options.setup.encoding,
strip_whitespace=False)
reset()
@@ -91,9 +94,9 @@ class TestCfgGenshiGenerator(TestCfgGenerator):
self.assertRaises(ValueError,
cgg.get_data, entry, metadata)
cgg.template.generate.assert_called_with(**template_vars)
- cgg.setup.__getitem__.assert_called_with("repo")
fltr.filter.assert_called_with(removecomment)
- stream.render.assert_called_with("text", encoding=cgg.encoding,
+ stream.render.assert_called_with("text",
+ encoding=Bcfg2.Options.setup.encoding,
strip_whitespace=False)
self.assertTrue(cgg._handle_genshi_exception.called)
@@ -102,14 +105,16 @@ class TestCfgGenshiGenerator(TestCfgGenerator):
cgg.loader = Mock()
event = Mock()
cgg.handle_event(event)
- cgg.loader.load.assert_called_with(cgg.name,
- cls=NewTextTemplate,
- encoding=cgg.encoding)
+ cgg.loader.load.assert_called_with(
+ cgg.name,
+ cls=NewTextTemplate,
+ encoding=Bcfg2.Options.setup.encoding)
cgg.loader.reset_mock()
cgg.loader.load.side_effect = OSError
self.assertRaises(PluginExecutionError,
cgg.handle_event, event)
- cgg.loader.load.assert_called_with(cgg.name,
- cls=NewTextTemplate,
- encoding=cgg.encoding)
+ cgg.loader.load.assert_called_with(
+ cgg.name,
+ cls=NewTextTemplate,
+ encoding=Bcfg2.Options.setup.encoding)
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgInfoXML.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgInfoXML.py
index 7e7cb5e3c..349da2213 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgInfoXML.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgInfoXML.py
@@ -21,6 +21,10 @@ from TestServer.TestPlugins.TestCfg.Test_init import TestCfgInfo
class TestCfgInfoXML(TestCfgInfo):
test_obj = CfgInfoXML
+ def setUp(self):
+ TestCfgInfo.setUp(self)
+ set_setup_default("filemonitor", MagicMock())
+
def test__init(self):
TestCfgInfo.test__init(self)
ci = self.get_obj()
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgPrivateKeyCreator.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgPrivateKeyCreator.py
index 48d5cdbfe..c4961db1c 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgPrivateKeyCreator.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/TestCfgPrivateKeyCreator.py
@@ -42,59 +42,6 @@ class TestCfgPrivateKeyCreator(TestCfgCreator, TestStructFile):
mock_HandleEvent.assert_called_with(pkc, evt)
mock_handle_event.assert_called_with(pkc, evt)
- def test_category(self):
- pkc = self.get_obj()
- pkc.setup = Mock()
- pkc.setup.cfp = Mock()
- pkc.setup.cfp.has_section.return_value = False
- pkc.setup.cfp.has_option.return_value = False
-
- self.assertIsNone(pkc.category)
- pkc.setup.cfp.has_section.assert_called_with("sshkeys")
-
- pkc.setup.reset_mock()
- pkc.setup.cfp.has_section.return_value = True
- self.assertIsNone(pkc.category)
- pkc.setup.cfp.has_section.assert_called_with("sshkeys")
- pkc.setup.cfp.has_option.assert_called_with("sshkeys", "category")
-
- pkc.setup.reset_mock()
- pkc.setup.cfp.has_option.return_value = True
- self.assertEqual(pkc.category, pkc.setup.cfp.get.return_value)
- pkc.setup.cfp.has_section.assert_called_with("sshkeys")
- pkc.setup.cfp.has_option.assert_called_with("sshkeys", "category")
- pkc.setup.cfp.get.assert_called_with("sshkeys", "category")
-
- @skipUnless(HAS_CRYPTO, "No crypto libraries found, skipping")
- @patchIf(HAS_CRYPTO, "Bcfg2.Server.Encryption.get_passphrases")
- def test_passphrase(self, mock_get_passphrases):
- pkc = self.get_obj()
- pkc.setup = Mock()
- pkc.setup.cfp = Mock()
- pkc.setup.cfp.has_section.return_value = False
- pkc.setup.cfp.has_option.return_value = False
-
- self.assertIsNone(pkc.passphrase)
- pkc.setup.cfp.has_section.assert_called_with("sshkeys")
-
- pkc.setup.reset_mock()
- pkc.setup.cfp.has_section.return_value = True
- self.assertIsNone(pkc.passphrase)
- pkc.setup.cfp.has_section.assert_called_with("sshkeys")
- pkc.setup.cfp.has_option.assert_called_with("sshkeys",
- "passphrase")
-
- pkc.setup.reset_mock()
- pkc.setup.cfp.get.return_value = "test"
- mock_get_passphrases.return_value = dict(test="foo", test2="bar")
- pkc.setup.cfp.has_option.return_value = True
- self.assertEqual(pkc.passphrase, "foo")
- pkc.setup.cfp.has_section.assert_called_with("sshkeys")
- pkc.setup.cfp.has_option.assert_called_with("sshkeys",
- "passphrase")
- pkc.setup.cfp.get.assert_called_with("sshkeys", "passphrase")
- mock_get_passphrases.assert_called_with()
-
@patch("shutil.rmtree")
@patch("tempfile.mkdtemp")
def test__gen_keypair(self, mock_mkdtemp, mock_rmtree):
@@ -153,52 +100,46 @@ class TestCfgPrivateKeyCreator(TestCfgCreator, TestStructFile):
pkc.XMLMatch.reset_mock()
metadata.group_in_category.reset_mock()
- category = "Bcfg2.Server.Plugins.Cfg.CfgPrivateKeyCreator.CfgPrivateKeyCreator.category"
- @patch(category, None)
- def inner():
- pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey")
- self.assertItemsEqual(pkc.get_specificity(metadata),
- dict(host=metadata.hostname))
- inner()
-
- @patch(category, "foo")
- def inner2():
- pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey")
- self.assertItemsEqual(pkc.get_specificity(metadata),
- dict(group=metadata.group_in_category.return_value,
- prio=50))
- metadata.group_in_category.assert_called_with("foo")
+ Bcfg2.Options.setup.sshkeys_category = None
+ pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey")
+ self.assertItemsEqual(pkc.get_specificity(metadata),
+ dict(host=metadata.hostname))
- reset()
- pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey",
- perhost="true")
- self.assertItemsEqual(pkc.get_specificity(metadata),
- dict(host=metadata.hostname))
+ Bcfg2.Options.setup.sshkeys_category = "foo"
+ pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey")
+ self.assertItemsEqual(pkc.get_specificity(metadata),
+ dict(group=metadata.group_in_category.return_value,
+ prio=50))
+ metadata.group_in_category.assert_called_with("foo")
- reset()
- pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey",
- category="bar")
- self.assertItemsEqual(pkc.get_specificity(metadata),
- dict(group=metadata.group_in_category.return_value,
- prio=50))
- metadata.group_in_category.assert_called_with("bar")
+ reset()
+ pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey",
+ perhost="true")
+ self.assertItemsEqual(pkc.get_specificity(metadata),
+ dict(host=metadata.hostname))
- reset()
- pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey",
- prio="10")
- self.assertItemsEqual(pkc.get_specificity(metadata),
- dict(group=metadata.group_in_category.return_value,
- prio=10))
- metadata.group_in_category.assert_called_with("foo")
+ reset()
+ pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey",
+ category="bar")
+ self.assertItemsEqual(pkc.get_specificity(metadata),
+ dict(group=metadata.group_in_category.return_value,
+ prio=50))
+ metadata.group_in_category.assert_called_with("bar")
- reset()
- pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey")
- metadata.group_in_category.return_value = ''
- self.assertItemsEqual(pkc.get_specificity(metadata),
- dict(host=metadata.hostname))
- metadata.group_in_category.assert_called_with("foo")
+ reset()
+ pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey",
+ prio="10")
+ self.assertItemsEqual(pkc.get_specificity(metadata),
+ dict(group=metadata.group_in_category.return_value,
+ prio=10))
+ metadata.group_in_category.assert_called_with("foo")
- inner2()
+ reset()
+ pkc.XMLMatch.return_value = lxml.etree.Element("PrivateKey")
+ metadata.group_in_category.return_value = ''
+ self.assertItemsEqual(pkc.get_specificity(metadata),
+ dict(host=metadata.hostname))
+ metadata.group_in_category.assert_called_with("foo")
@patch("shutil.rmtree")
@patch("%s.open" % builtins)
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/Test_init.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/Test_init.py
index 07a1b120e..72be50299 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/Test_init.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestCfg/Test_init.py
@@ -152,7 +152,7 @@ class TestCfgInfo(TestCfgBaseFileMatcher):
@patch("Bcfg2.Server.Plugins.Cfg.CfgBaseFileMatcher.__init__")
def test__init(self, mock__init):
ci = self.get_obj("test.txt")
- mock__init.assert_called_with(ci, "test.txt", None, None)
+ mock__init.assert_called_with(ci, "test.txt", None)
def test_bind_info_to_entry(self):
ci = self.get_obj()
@@ -173,6 +173,10 @@ class TestCfgCreator(TestCfgBaseFileMatcher):
test_obj = CfgCreator
path = "/foo/bar/test.txt"
+ def setUp(self):
+ TestCfgBaseFileMatcher.setUp(self)
+ set_setup_default("filemonitor", MagicMock())
+
def get_obj(self, name=None):
if name is None:
name = self.path
@@ -269,31 +273,24 @@ class TestCfgDefaultInfo(TestCfgInfo):
class TestCfgEntrySet(TestEntrySet):
test_obj = CfgEntrySet
+ def setUp(self):
+ TestEntrySet.setUp(self)
+ set_setup_default("cfg_validation", False)
+
def test__init(self):
pass
- def test_handlers(self):
- # this is really really difficult to mock out, so we just get
- # a list of handlers and make sure that it roughly matches
- # what's on the filesystem
- expected = []
- for submodule in walk_packages(path=Bcfg2.Server.Plugins.Cfg.__path__,
- prefix="Bcfg2.Server.Plugins.Cfg."):
- expected.append(submodule[1].rsplit('.', 1)[-1])
- self.assertItemsEqual(expected, [h.__name__ for h in handlers()])
-
- @patch("Bcfg2.Server.Plugins.Cfg.handlers")
- def test_handle_event(self, mock_handlers):
+ def test_handle_event(self):
eset = self.get_obj()
eset.entry_init = Mock()
- mock_handlers.return_value = [Mock(), Mock(), Mock()]
- for hdlr in mock_handlers.return_value:
+ eset._handlers = [Mock(), Mock(), Mock()]
+ for hdlr in eset._handlers:
hdlr.__name__ = "handler"
eset.entries = dict()
def reset():
eset.entry_init.reset_mock()
- for hdlr in mock_handlers.return_value:
+ for hdlr in eset._handlers:
hdlr.reset_mock()
# test that a bogus deleted event is discarded
@@ -303,7 +300,7 @@ class TestCfgEntrySet(TestEntrySet):
eset.handle_event(evt)
self.assertFalse(eset.entry_init.called)
self.assertItemsEqual(eset.entries, dict())
- for hdlr in mock_handlers.return_value:
+ for hdlr in eset._handlers:
self.assertFalse(hdlr.handles.called)
self.assertFalse(hdlr.ignore.called)
@@ -314,7 +311,7 @@ class TestCfgEntrySet(TestEntrySet):
evt.filename = os.path.join(datastore, "test.txt")
# test with no handler that handles
- for hdlr in mock_handlers.return_value:
+ for hdlr in eset._handlers:
hdlr.handles.return_value = False
hdlr.ignore.return_value = False
@@ -322,16 +319,16 @@ class TestCfgEntrySet(TestEntrySet):
eset.handle_event(evt)
self.assertFalse(eset.entry_init.called)
self.assertItemsEqual(eset.entries, dict())
- for hdlr in mock_handlers.return_value:
+ for hdlr in eset._handlers:
hdlr.handles.assert_called_with(evt, basename=eset.path)
hdlr.ignore.assert_called_with(evt, basename=eset.path)
# test with a handler that handles the entry
reset()
- mock_handlers.return_value[-1].handles.return_value = True
+ eset._handlers[-1].handles.return_value = True
eset.handle_event(evt)
- eset.entry_init.assert_called_with(evt, mock_handlers.return_value[-1])
- for hdlr in mock_handlers.return_value:
+ eset.entry_init.assert_called_with(evt, eset._handlers[-1])
+ for hdlr in eset._handlers:
hdlr.handles.assert_called_with(evt, basename=eset.path)
if not hdlr.return_value:
hdlr.ignore.assert_called_with(evt, basename=eset.path)
@@ -339,14 +336,14 @@ class TestCfgEntrySet(TestEntrySet):
# test with a handler that ignores the entry before one
# that handles it
reset()
- mock_handlers.return_value[0].ignore.return_value = True
+ eset._handlers[0].ignore.return_value = True
eset.handle_event(evt)
self.assertFalse(eset.entry_init.called)
- mock_handlers.return_value[0].handles.assert_called_with(evt,
+ eset._handlers[0].handles.assert_called_with(evt,
+ basename=eset.path)
+ eset._handlers[0].ignore.assert_called_with(evt,
basename=eset.path)
- mock_handlers.return_value[0].ignore.assert_called_with(evt,
- basename=eset.path)
- for hdlr in mock_handlers.return_value[1:]:
+ for hdlr in eset._handlers[1:]:
self.assertFalse(hdlr.handles.called)
self.assertFalse(hdlr.ignore.called)
@@ -358,7 +355,7 @@ class TestCfgEntrySet(TestEntrySet):
eset.entries[evt.filename] = Mock()
eset.handle_event(evt)
self.assertFalse(eset.entry_init.called)
- for hdlr in mock_handlers.return_value:
+ for hdlr in eset._handlers:
self.assertFalse(hdlr.handles.called)
self.assertFalse(hdlr.ignore.called)
eset.entries[evt.filename].handle_event.assert_called_with(evt)
@@ -368,7 +365,7 @@ class TestCfgEntrySet(TestEntrySet):
evt.code2str.return_value = "deleted"
eset.handle_event(evt)
self.assertFalse(eset.entry_init.called)
- for hdlr in mock_handlers.return_value:
+ for hdlr in eset._handlers:
self.assertFalse(hdlr.handles.called)
self.assertFalse(hdlr.ignore.called)
self.assertItemsEqual(eset.entries, dict())
@@ -422,6 +419,7 @@ class TestCfgEntrySet(TestEntrySet):
def test_bind_entry(self, mock_b64encode, mock_u_str):
mock_u_str.side_effect = lambda x: x
+ Bcfg2.Options.setup.cfg_validation = False
eset = self.get_obj()
eset.bind_info_to_entry = Mock()
eset._generate_data = Mock()
@@ -505,7 +503,7 @@ class TestCfgEntrySet(TestEntrySet):
# test successful validation
entry = reset()
- eset.setup['validate'] = True
+ Bcfg2.Options.setup.cfg_validation = True
bound = eset.bind_entry(entry, metadata)
eset.bind_info_to_entry.assert_called_with(entry, metadata)
eset._generate_data.assert_called_with(entry, metadata)
@@ -527,16 +525,16 @@ class TestCfgEntrySet(TestEntrySet):
def test_get_handlers(self):
eset = self.get_obj()
eset.entries['test1.txt'] = CfgInfo("test1.txt")
- eset.entries['test2.txt'] = CfgGenerator("test2.txt", Mock(), None)
+ eset.entries['test2.txt'] = CfgGenerator("test2.txt", Mock())
eset.entries['test2.txt'].specific.matches.return_value = True
eset.entries['test3.txt'] = CfgInfo("test3.txt")
- eset.entries['test4.txt'] = CfgGenerator("test4.txt", Mock(), None)
+ eset.entries['test4.txt'] = CfgGenerator("test4.txt", Mock())
eset.entries['test4.txt'].specific.matches.return_value = False
- eset.entries['test5.txt'] = CfgGenerator("test5.txt", Mock(), None)
+ eset.entries['test5.txt'] = CfgGenerator("test5.txt", Mock())
eset.entries['test5.txt'].specific.matches.return_value = True
- eset.entries['test6.txt'] = CfgVerifier("test6.txt", Mock(), None)
+ eset.entries['test6.txt'] = CfgVerifier("test6.txt", Mock())
eset.entries['test6.txt'].specific.matches.return_value = True
- eset.entries['test7.txt'] = CfgFilter("test7.txt", Mock(), None)
+ eset.entries['test7.txt'] = CfgFilter("test7.txt", Mock())
eset.entries['test7.txt'].specific.matches.return_value = False
def reset():
@@ -737,38 +735,6 @@ class TestCfg(TestGroupSpool, TestPullTarget):
core = Mock()
return TestGroupSpool.get_obj(self, core=core)
- @patch("Bcfg2.Options.get_option_parser")
- @patch("Bcfg2.Server.Plugin.GroupSpool.__init__")
- @patch("Bcfg2.Server.Plugin.PullTarget.__init__")
- def test__init(self, mock_pulltarget_init, mock_groupspool_init,
- mock_get_option_parser):
- setup = MagicMock()
- setup.__contains__.return_value = False
- mock_get_option_parser.return_value = setup
-
- def reset():
- core.reset_mock()
- setup.reset_mock()
- mock_pulltarget_init.reset_mock()
- mock_groupspool_init.reset_mock()
-
- core = Mock()
- cfg = self.test_obj(core, datastore)
- mock_pulltarget_init.assert_called_with(cfg)
- mock_groupspool_init.assert_called_with(cfg, core, datastore)
- setup.add_option.assert_called_with(
- "validate",
- Bcfg2.Options.CFG_VALIDATION)
- mock_get_option_parser.return_value.reparse.assert_called_with()
-
- reset()
- setup.__contains__.return_value = True
- cfg = self.test_obj(core, datastore)
- mock_pulltarget_init.assert_called_with(cfg)
- mock_groupspool_init.assert_called_with(cfg, core, datastore)
- self.assertFalse(setup.add_option.called)
- self.assertFalse(setup.reparse.called)
-
def test_has_generator(self):
cfg = self.get_obj()
cfg.entries = dict()
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestMetadata.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestMetadata.py
index 90f592eb2..274b5e302 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestMetadata.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestMetadata.py
@@ -6,7 +6,6 @@ import socket
import lxml.etree
import Bcfg2.Server
import Bcfg2.Server.Plugin
-from Bcfg2.Server.Plugins.Metadata import *
from mock import Mock, MagicMock, patch
# add all parent testsuite directories to sys.path to allow (most)
@@ -19,9 +18,13 @@ while path != "/":
break
path = os.path.dirname(path)
from common import *
+from Bcfg2.Server.Plugins.Metadata import load_django_models
from TestPlugin import TestXMLFileBacked, TestMetadata as _TestMetadata, \
TestClientRunHooks, TestDatabaseBacked
+load_django_models()
+from Bcfg2.Server.Plugins.Metadata import *
+
def get_clients_test_tree():
return lxml.etree.XML('''
@@ -88,13 +91,12 @@ def get_groups_test_tree():
</Groups>''').getroottree()
-def get_metadata_object(core=None, watch_clients=False, use_db=False):
+def get_metadata_object(core=None, watch_clients=False):
if core is None:
core = Mock()
- core.setup = MagicMock()
core.metadata_cache = MagicMock()
- core.setup.cfp.getboolean = Mock(return_value=use_db)
+ set_setup_default("password")
@patchIf(not isinstance(os.makedirs, Mock), "os.makedirs", Mock())
@patchIf(not isinstance(lxml.etree.Element, Mock),
"lxml.etree.Element", Mock())
@@ -119,6 +121,7 @@ if HAS_DJANGO or can_skip:
@skipUnless(HAS_DJANGO, "Django not found")
def setUp(self):
+ TestDatabaseBacked.setUp(self)
self.test_obj = ClientVersions
syncdb(TestMetadataDB)
for client, version in self.test_clients.items():
@@ -476,11 +479,16 @@ class TestClientMetadata(Bcfg2TestCase):
class TestMetadata(_TestMetadata, TestClientRunHooks, TestDatabaseBacked):
test_obj = Metadata
- use_db = False
+
+ def setUp(self):
+ _TestMetadata.setUp(self)
+ TestClientRunHooks.setUp(self)
+ TestDatabaseBacked.setUp(self)
+ Bcfg2.Options.setup.metadata_db = False
+ Bcfg2.Options.setup.authentication = "cert+password"
def get_obj(self, core=None, watch_clients=False):
- return get_metadata_object(core=core, watch_clients=watch_clients,
- use_db=self.use_db)
+ return get_metadata_object(core=core, watch_clients=watch_clients)
@skipUnless(HAS_DJANGO, "Django not found")
def test__use_db(self):
@@ -767,7 +775,7 @@ class TestMetadata(_TestMetadata, TestClientRunHooks, TestDatabaseBacked):
metadata.clients_xml.xdata = copy.deepcopy(get_clients_test_tree())
metadata._handle_clients_xml_event(Mock())
- if not self.use_db:
+ if not Bcfg2.Options.setup.metadata_db:
self.assertItemsEqual(metadata.clients,
dict([(c.get("name"), c.get("profile"))
for c in get_clients_test_tree().findall("//Client")]))
@@ -1259,10 +1267,13 @@ class TestMetadata(_TestMetadata, TestClientRunHooks, TestDatabaseBacked):
class TestMetadataBase(TestMetadata):
""" base test object for testing Metadata with database enabled """
__test__ = False
- use_db = True
@skipUnless(HAS_DJANGO, "Django not found")
def setUp(self):
+ _TestMetadata.setUp(self)
+ TestClientRunHooks.setUp(self)
+ TestDatabaseBacked.setUp(self)
+ Bcfg2.Options.setup.metadata_db = True
syncdb(TestMetadataDB)
def load_clients_data(self, metadata=None, xdata=None):
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py
index f44bc338c..b0e6e9142 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProbes.py
@@ -18,10 +18,19 @@ while path != "/":
break
path = os.path.dirname(path)
from common import *
-from Bcfg2.Server.Plugins.Probes import *
+from Bcfg2.Server.Plugins.Probes import load_django_models
from TestPlugin import TestEntrySet, TestProbing, TestConnector, \
TestDatabaseBacked
+load_django_models()
+from Bcfg2.Server.Plugins.Probes import *
+
+if HAS_JSON:
+ json = json
+
+if HAS_YAML:
+ yaml = yaml
+
# test data for JSON and YAML tests
test_data = dict(a=1, b=[1, 2, 3], c="test",
d=dict(a=1, b=dict(a=1), c=(1, "2", 3)))
@@ -74,7 +83,8 @@ class FakeList(list):
class TestProbesDB(DBModelTestCase):
if HAS_DJANGO:
- models = [ProbesGroupsModel, ProbesDataModel]
+ models = [ProbesGroupsModel,
+ ProbesDataModel]
class TestClientProbeDataSet(Bcfg2TestCase):
@@ -108,19 +118,22 @@ class TestProbeData(Bcfg2TestCase):
def test_xdata(self):
xdata = lxml.etree.Element("test")
lxml.etree.SubElement(xdata, "test2")
- data = ProbeData(lxml.etree.tostring(xdata,
- xml_declaration=False).decode('UTF-8'))
+ data = ProbeData(
+ lxml.etree.tostring(xdata,
+ xml_declaration=False).decode('UTF-8'))
self.assertIsNotNone(data.xdata)
self.assertIsNotNone(data.xdata.find("test2"))
- @skipUnless(HAS_JSON, "JSON libraries not found, skipping JSON tests")
+ @skipUnless(HAS_JSON,
+ "JSON libraries not found, skipping JSON tests")
def test_json(self):
jdata = json.dumps(test_data)
data = ProbeData(jdata)
self.assertIsNotNone(data.json)
self.assertItemsEqual(test_data, data.json)
- @skipUnless(HAS_YAML, "YAML libraries not found, skipping YAML tests")
+ @skipUnless(HAS_YAML,
+ "YAML libraries not found, skipping YAML tests")
def test_yaml(self):
jdata = yaml.dump(test_data)
data = ProbeData(jdata)
@@ -139,7 +152,7 @@ class TestProbeSet(TestEntrySet):
# get_obj() accepts the basename argument, accepted by the
# parent get_obj() method, and just throws it away, since
# ProbeSet uses a regex for the "basename"
- rv = self.test_obj(path, encoding, plugin_name)
+ rv = self.test_obj(path, plugin_name)
rv.entry_type = MagicMock()
return rv
@@ -243,30 +256,47 @@ group-specific"""
class TestProbes(TestProbing, TestConnector, TestDatabaseBacked):
test_obj = Probes
- def get_obj(self, core=None):
- return TestDatabaseBacked.get_obj(self, core=core)
+ def get_obj(self, core=None, load_data=None):
+ core = MagicMock()
+ if load_data is None:
+ load_data = MagicMock()
+
+ @patch("%s.%s.load_data" % (self.test_obj.__module__,
+ self.test_obj.__name__), new=load_data)
+ def inner():
+ return TestDatabaseBacked.get_obj(self, core=core)
+
+ return inner()
def get_test_probedata(self):
test_xdata = lxml.etree.Element("test")
lxml.etree.SubElement(test_xdata, "test", foo="foo")
rv = dict()
- rv["foo.example.com"] = ClientProbeDataSet(timestamp=time.time())
+ rv["foo.example.com"] = ClientProbeDataSet(
+ timestamp=time.time())
rv["foo.example.com"]["xml"] = \
- ProbeData(lxml.etree.tostring(test_xdata,
- xml_declaration=False).decode('UTF-8'))
- rv["foo.example.com"]["text"] = ProbeData("freeform text")
- rv["foo.example.com"]["multiline"] = ProbeData("""multiple
+ ProbeData(lxml.etree.tostring(
+ test_xdata,
+ xml_declaration=False).decode('UTF-8'))
+ rv["foo.example.com"]["text"] = \
+ ProbeData("freeform text")
+ rv["foo.example.com"]["multiline"] = \
+ ProbeData("""multiple
lines
of
freeform
text
""")
- rv["bar.example.com"] = ClientProbeDataSet(timestamp=time.time())
- rv["bar.example.com"]["empty"] = ProbeData("")
+ rv["bar.example.com"] = ClientProbeDataSet(
+ timestamp=time.time())
+ rv["bar.example.com"]["empty"] = \
+ ProbeData("")
if HAS_JSON:
- rv["bar.example.com"]["json"] = ProbeData(json.dumps(test_data))
+ rv["bar.example.com"]["json"] = \
+ ProbeData(json.dumps(test_data))
if HAS_YAML:
- rv["bar.example.com"]["yaml"] = ProbeData(yaml.dump(test_data))
+ rv["bar.example.com"]["yaml"] = \
+ ProbeData(yaml.dump(test_data))
return rv
def get_test_cgroups(self):
@@ -274,62 +304,36 @@ text
"group-with-dashes"],
"bar.example.com": []}
- def get_probes_object(self, use_db=False, load_data=None):
- core = MagicMock()
- core.setup.cfp.getboolean = Mock()
- core.setup.cfp.getboolean.return_value = use_db
- if load_data is None:
- load_data = MagicMock()
- # we have to patch load_data() in a funny way because
- # different versions of Mock have different scopes for
- # patching. in some versions, a patch applied to
- # get_probes_object() would only apply to that function, while
- # in others it would also apply to the calling function (e.g.,
- # test__init(), which relies on being able to check the calls
- # of load_data(), and thus on load_data() being consistently
- # mocked)
- @patch("%s.%s.load_data" % (self.test_obj.__module__,
- self.test_obj.__name__), new=load_data)
- def inner():
- return self.get_obj(core)
-
- return inner()
-
def test__init(self):
mock_load_data = Mock()
- probes = self.get_probes_object(load_data=mock_load_data)
+ probes = self.get_obj(load_data=mock_load_data)
mock_load_data.assert_any_call()
- self.assertEqual(probes.probedata, ClientProbeDataSet())
+ self.assertEqual(probes.probedata,
+ ClientProbeDataSet())
self.assertEqual(probes.cgroups, dict())
- @patch("Bcfg2.Server.Plugins.Probes.Probes.load_data", Mock())
- def test__use_db(self):
- probes = self.get_probes_object()
- self.assertFalse(probes._use_db)
- probes.core.setup.cfp.getboolean.assert_called_with("probes",
- "use_database",
- default=False)
-
- @skipUnless(HAS_DJANGO, "Django not found, skipping")
- @patch("Bcfg2.Server.Plugins.Probes.Probes._write_data_db", Mock())
- @patch("Bcfg2.Server.Plugins.Probes.Probes._write_data_xml", Mock())
def test_write_data_xml(self):
- probes = self.get_probes_object(use_db=False)
+ Bcfg2.Options.setup.probes_db = False
+ probes = self.get_obj()
+ probes._write_data_db = Mock()
+ probes._write_data_xml = Mock()
probes.write_data("test")
probes._write_data_xml.assert_called_with("test")
self.assertFalse(probes._write_data_db.called)
@skipUnless(HAS_DJANGO, "Django not found, skipping")
- @patch("Bcfg2.Server.Plugins.Probes.Probes._write_data_db", Mock())
- @patch("Bcfg2.Server.Plugins.Probes.Probes._write_data_xml", Mock())
def test_write_data_db(self):
- probes = self.get_probes_object(use_db=True)
+ Bcfg2.Options.setup.probes_db = True
+ probes = self.get_obj()
+ probes._write_data_db = Mock()
+ probes._write_data_xml = Mock()
probes.write_data("test")
probes._write_data_db.assert_called_with("test")
self.assertFalse(probes._write_data_xml.called)
def test__write_data_xml(self):
- probes = self.get_probes_object(use_db=False)
+ Bcfg2.Options.setup.probes_db = False
+ probes = self.get_obj()
probes.probedata = self.get_test_probedata()
probes.cgroups = self.get_test_cgroups()
@@ -397,7 +401,8 @@ text
@skipUnless(HAS_DJANGO, "Django not found, skipping")
def test__write_data_db(self):
syncdb(TestProbesDB)
- probes = self.get_probes_object(use_db=True)
+ Bcfg2.Options.setup.probes_db = True
+ probes = self.get_obj()
probes.probedata = self.get_test_probedata()
probes.cgroups = self.get_test_cgroups()
@@ -446,27 +451,29 @@ text
pgroups = ProbesGroupsModel.objects.filter(hostname=cname).all()
self.assertEqual(len(pgroups), len(probes.cgroups[cname]))
- @skipUnless(HAS_DJANGO, "Django not found, skipping")
- @patch("Bcfg2.Server.Plugins.Probes.Probes._load_data_db", Mock())
- @patch("Bcfg2.Server.Plugins.Probes.Probes._load_data_xml", Mock())
def test_load_data_xml(self):
- probes = self.get_probes_object(use_db=False)
+ Bcfg2.Options.setup.probes_db = False
+ probes = self.get_obj()
+ probes._load_data_db = Mock()
+ probes._load_data_xml = Mock()
probes.load_data()
probes._load_data_xml.assert_any_call()
self.assertFalse(probes._load_data_db.called)
@skipUnless(HAS_DJANGO, "Django not found, skipping")
- @patch("Bcfg2.Server.Plugins.Probes.Probes._load_data_db", Mock())
- @patch("Bcfg2.Server.Plugins.Probes.Probes._load_data_xml", Mock())
def test_load_data_db(self):
- probes = self.get_probes_object(use_db=True)
+ Bcfg2.Options.setup.probes_db = True
+ probes = self.get_obj()
+ probes._load_data_db = Mock()
+ probes._load_data_xml = Mock()
probes.load_data()
probes._load_data_db.assert_any_call(client=None)
self.assertFalse(probes._load_data_xml.called)
@patch("lxml.etree.parse")
def test__load_data_xml(self, mock_parse):
- probes = self.get_probes_object(use_db=False)
+ Bcfg2.Options.setup.probes_db = False
+ probes = self.get_obj()
probes.probedata = self.get_test_probedata()
probes.cgroups = self.get_test_cgroups()
@@ -496,7 +503,8 @@ text
@skipUnless(HAS_DJANGO, "Django not found, skipping")
def test__load_data_db(self):
syncdb(TestProbesDB)
- probes = self.get_probes_object(use_db=True)
+ Bcfg2.Options.setup.probes_db = True
+ probes = self.get_obj()
probes.probedata = self.get_test_probedata()
probes.cgroups = self.get_test_cgroups()
for cname in probes.probedata.keys():
@@ -521,19 +529,19 @@ text
@patch("Bcfg2.Server.Plugins.Probes.ProbeSet.get_probe_data")
def test_GetProbes(self, mock_get_probe_data):
- probes = self.get_probes_object()
+ probes = self.get_obj()
metadata = Mock()
probes.GetProbes(metadata)
mock_get_probe_data.assert_called_with(metadata)
- @patch("Bcfg2.Server.Plugins.Probes.Probes.write_data")
- @patch("Bcfg2.Server.Plugins.Probes.Probes.ReceiveDataItem")
- def test_ReceiveData(self, mock_ReceiveDataItem, mock_write_data):
+ def test_ReceiveData(self):
# we use a simple (read: bogus) datalist here to make this
# easy to test
datalist = ["a", "b", "c"]
- probes = self.get_probes_object()
+ probes = self.get_obj()
+ probes.write_data = Mock()
+ probes.ReceiveDataItem = Mock()
probes.core.metadata_cache_mode = 'off'
client = Mock()
client.hostname = "foo.example.com"
@@ -541,11 +549,11 @@ text
cgroups = []
cprobedata = ClientProbeDataSet()
- self.assertItemsEqual(mock_ReceiveDataItem.call_args_list,
+ self.assertItemsEqual(probes.ReceiveDataItem.call_args_list,
[call(client, "a", cgroups, cprobedata),
call(client, "b", cgroups, cprobedata),
call(client, "c", cgroups, cprobedata)])
- mock_write_data.assert_called_with(client)
+ probes.write_data.assert_called_with(client)
self.assertFalse(probes.core.metadata_cache.expire.called)
# change the datalist, ensure that the cache is cleared
@@ -553,11 +561,11 @@ text
probes.core.metadata_cache_mode = 'aggressive'
probes.ReceiveData(client, ['a', 'b', 'd'])
- mock_write_data.assert_called_with(client)
+ probes.write_data.assert_called_with(client)
probes.core.metadata_cache.expire.assert_called_with(client.hostname)
def test_ReceiveDataItem(self):
- probes = self.get_probes_object()
+ probes = self.get_obj()
for cname, cdata in self.get_test_probedata().items():
client = Mock()
client.hostname = cname
@@ -589,7 +597,7 @@ text
def test_get_additional_groups(self):
TestConnector.test_get_additional_groups(self)
- probes = self.get_probes_object()
+ probes = self.get_obj()
test_cgroups = self.get_test_cgroups()
probes.cgroups = self.get_test_cgroups()
for cname in test_cgroups.keys():
@@ -606,7 +614,7 @@ text
def test_get_additional_data(self):
TestConnector.test_get_additional_data(self)
- probes = self.get_probes_object()
+ probes = self.get_obj()
test_probedata = self.get_test_probedata()
probes.probedata = self.get_test_probedata()
for cname in test_probedata.keys():
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProperties.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProperties.py
index 92dc85fb1..9fa2cc4db 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProperties.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestProperties.py
@@ -30,6 +30,7 @@ class TestPropertyFile(Bcfg2TestCase):
path = os.path.join(datastore, "test")
def get_obj(self, path=None):
+ set_setup_default("writes_enabled", False)
if path is None:
path = self.path
return self.test_obj(path)
@@ -38,7 +39,6 @@ class TestPropertyFile(Bcfg2TestCase):
pf = self.get_obj()
pf.validate_data = Mock()
pf._write = Mock()
- pf.setup = Mock()
xstr = u("<Properties/>\n")
pf.xdata = lxml.etree.XML(xstr)
@@ -46,20 +46,16 @@ class TestPropertyFile(Bcfg2TestCase):
def reset():
pf.validate_data.reset_mock()
pf._write.reset_mock()
- pf.setup.reset_mock()
# test writes disabled
- pf.setup.cfp.getboolean.return_value = False
+ Bcfg2.Options.setup.writes_enabled = False
self.assertRaises(PluginExecutionError, pf.write)
self.assertFalse(pf.validate_data.called)
self.assertFalse(pf._write.called)
- pf.setup.cfp.getboolean.assert_called_with("properties",
- "writes_enabled",
- default=True)
# test successful write
reset()
- pf.setup.cfp.getboolean.return_value = True
+ Bcfg2.Options.setup.writes_enabled = True
self.assertEqual(pf.write(), pf._write.return_value)
pf.validate_data.assert_called_with()
pf._write.assert_called_with()
@@ -97,12 +93,10 @@ if can_skip or HAS_JSON:
class TestJSONPropertyFile(TestFileBacked, TestPropertyFile):
test_obj = JSONPropertyFile
- def get_obj(self, *args, **kwargs):
- return TestFileBacked.get_obj(self, *args, **kwargs)
-
@skipUnless(HAS_JSON, "JSON libraries not found, skipping")
def setUp(self):
- pass
+ TestFileBacked.setUp(self)
+ TestPropertyFile.setUp(self)
@patch("%s.loads" % JSON)
def test_Index(self, mock_loads):
@@ -140,12 +134,10 @@ if can_skip or HAS_YAML:
class TestYAMLPropertyFile(TestFileBacked, TestPropertyFile):
test_obj = YAMLPropertyFile
- def get_obj(self, *args, **kwargs):
- return TestFileBacked.get_obj(self, *args, **kwargs)
-
@skipUnless(HAS_YAML, "YAML libraries not found, skipping")
def setUp(self):
- pass
+ TestFileBacked.setUp(self)
+ TestPropertyFile.setUp(self)
@patch("yaml.load")
def test_Index(self, mock_load):
@@ -183,6 +175,11 @@ class TestXMLPropertyFile(TestPropertyFile, TestStructFile):
test_obj = XMLPropertyFile
path = TestStructFile.path
+ def setUp(self):
+ TestPropertyFile.setUp(self)
+ TestStructFile.setUp(self)
+ set_setup_default("automatch", False)
+
def get_obj(self, *args, **kwargs):
return TestStructFile.get_obj(self, *args, **kwargs)
@@ -250,48 +247,34 @@ class TestXMLPropertyFile(TestPropertyFile, TestStructFile):
pf.setup.reset_mock()
pf.xdata = lxml.etree.Element("Properties", automatch="true")
- for automatch in [True, False]:
+ for Bcfg2.Options.setup.automatch in [True, False]:
reset()
- pf.setup.cfp.getboolean.return_value = automatch
self.assertEqual(pf.get_additional_data(metadata),
pf.XMLMatch.return_value)
pf.XMLMatch.assert_called_with(metadata)
- pf.setup.cfp.getboolean.assert_called_with("properties",
- "automatch",
- default=False)
self.assertFalse(mock_copy.called)
pf.xdata = lxml.etree.Element("Properties", automatch="false")
- for automatch in [True, False]:
+ for Bcfg2.Options.setup.automatch in [True, False]:
reset()
- pf.setup.cfp.getboolean.return_value = automatch
self.assertEqual(pf.get_additional_data(metadata),
mock_copy.return_value)
mock_copy.assert_called_with(pf)
self.assertFalse(pf.XMLMatch.called)
- pf.setup.cfp.getboolean.assert_called_with("properties",
- "automatch",
- default=False)
pf.xdata = lxml.etree.Element("Properties")
reset()
- pf.setup.cfp.getboolean.return_value = False
+ Bcfg2.Options.setup.automatch = False
self.assertEqual(pf.get_additional_data(metadata),
mock_copy.return_value)
mock_copy.assert_called_with(pf)
self.assertFalse(pf.XMLMatch.called)
- pf.setup.cfp.getboolean.assert_called_with("properties",
- "automatch",
- default=False)
reset()
- pf.setup.cfp.getboolean.return_value = True
+ Bcfg2.Options.setup.automatch = True
self.assertEqual(pf.get_additional_data(metadata),
pf.XMLMatch.return_value)
pf.XMLMatch.assert_called_with(metadata)
- pf.setup.cfp.getboolean.assert_called_with("properties",
- "automatch",
- default=False)
self.assertFalse(mock_copy.called)
diff --git a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestRules.py b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestRules.py
index 7083fff06..0bd69b371 100644
--- a/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestRules.py
+++ b/testsuite/Testsrc/Testlib/TestServer/TestPlugins/TestRules.py
@@ -21,6 +21,10 @@ from TestPlugin import TestPrioDir
class TestRules(TestPrioDir):
test_obj = Rules
+ def setUp(self):
+ TestPrioDir.setUp(self)
+ set_setup_default("rules_regex", False)
+
def test_HandlesEntry(self):
r = self.get_obj()
r.Entries = dict(Path={"/etc/foo.conf": Mock(),
@@ -76,11 +80,3 @@ class TestRules(TestPrioDir):
self.assertTrue(r._matches(entry, metadata, candidate))
mock_matches.assert_called_with(r, entry, metadata, candidate)
self.assertIn("/etc/.*\.conf", r._regex_cache.keys())
-
- def test__regex_enabled(self):
- r = self.get_obj()
- r.core.setup = MagicMock()
- self.assertEqual(r._regex_enabled,
- r.core.setup.cfp.getboolean.return_value)
- r.core.setup.cfp.getboolean.assert_called_with("rules", "regex",
- default=False)
diff --git a/testsuite/common.py b/testsuite/common.py
index e9c4ea339..b375d3703 100644
--- a/testsuite/common.py
+++ b/testsuite/common.py
@@ -35,16 +35,34 @@ inPy3k = False
if sys.hexversion >= 0x03000000:
inPy3k = True
+
+#: A function to set a default config option if it's not already set
+def set_setup_default(option, value=None):
+ if not hasattr(Bcfg2.Options.setup, option):
+ setattr(Bcfg2.Options.setup, option, value)
+
try:
from django.core.management import setup_environ
has_django = True
os.environ['DJANGO_SETTINGS_MODULE'] = "Bcfg2.settings"
+ set_setup_default("db_engine", "sqlite3")
+ set_setup_default("db_name",
+ os.path.join(os.path.dirname(os.path.abspath(__file__)),
+ "test.sqlite"))
+ set_setup_default("db_user")
+ set_setup_default("db_password")
+ set_setup_default("db_host")
+ set_setup_default("db_port")
+ set_setup_default("db_opts", dict())
+ set_setup_default("db_schema")
+ set_setup_default("timezone")
+ set_setup_default("web_debug", False)
+ set_setup_default("web_prefix")
+
import Bcfg2.settings
- Bcfg2.settings.DATABASE_NAME = \
- os.path.join(os.path.dirname(os.path.abspath(__file__)), "test.sqlite")
- Bcfg2.settings.DATABASES['default']['NAME'] = Bcfg2.settings.DATABASE_NAME
+ Bcfg2.settings.read_config()
except ImportError:
has_django = False
@@ -303,7 +321,8 @@ class DBModelTestCase(Bcfg2TestCase):
import django.core.management
django.core.management.call_command("syncdb", interactive=False,
verbosity=0)
- self.assertTrue(os.path.exists(Bcfg2.settings.DATABASE_NAME))
+ self.assertTrue(
+ os.path.exists(Bcfg2.settings.DATABASES['default']['NAME']))
@skipUnless(has_django, "Django not found, skipping")
def test_cleandb(self):
@@ -404,9 +423,3 @@ try:
re_type = re._pattern_type
except AttributeError:
re_type = type(re.compile(""))
-
-
-#: A function to set a default config option if it's not already set
-def set_setup_default(option, value=None):
- if not hasattr(Bcfg2.Options.setup, option):
- setattr(Bcfg2.Options.setup, option, value)