import os import sys import lxml.etree from mock import Mock, MagicMock, patch from Bcfg2.Server.Plugins.Properties import * from Bcfg2.Server.Plugin import PluginExecutionError # 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 * from TestPlugin import TestStructFile, TestFileBacked, TestConnector, \ TestPlugin, TestDirectoryBacked try: import json JSON = "json" except ImportError: JSON = "simplejson" class TestPropertyFile(Bcfg2TestCase): test_obj = PropertyFile 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) def test_write(self): pf = self.get_obj() pf.validate_data = Mock() pf._write = Mock() xstr = u("\n") pf.xdata = lxml.etree.XML(xstr) def reset(): pf.validate_data.reset_mock() pf._write.reset_mock() # test writes disabled Bcfg2.Options.setup.writes_enabled = False self.assertRaises(PluginExecutionError, pf.write) self.assertFalse(pf.validate_data.called) self.assertFalse(pf._write.called) # test successful write reset() 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() # test error from _write reset() pf._write.side_effect = IOError self.assertRaises(PluginExecutionError, pf.write) pf.validate_data.assert_called_with() pf._write.assert_called_with() # test error from validate_data reset() pf.validate_data.side_effect = PluginExecutionError self.assertRaises(PluginExecutionError, pf.write) pf.validate_data.assert_called_with() def test__write(self): pf = self.get_obj() self.assertRaises(NotImplementedError, pf._write) def test_validate_data(self): pf = self.get_obj() self.assertRaises(NotImplementedError, pf.validate_data) @patch("copy.copy") def test_get_additional_data(self, mock_copy): pf = self.get_obj() self.assertEqual(pf.get_additional_data(Mock()), mock_copy.return_value) mock_copy.assert_called_with(pf) class TestJSONPropertyFile(TestFileBacked, TestPropertyFile): test_obj = JSONPropertyFile @skipUnless(HAS_JSON, "JSON libraries not found, skipping") def setUp(self): TestFileBacked.setUp(self) TestPropertyFile.setUp(self) @patch("%s.loads" % JSON) def test_Index(self, mock_loads): pf = self.get_obj() pf.Index() mock_loads.assert_called_with(pf.data) self.assertEqual(pf.json, mock_loads.return_value) mock_loads.reset_mock() mock_loads.side_effect = ValueError self.assertRaises(PluginExecutionError, pf.Index) mock_loads.assert_called_with(pf.data) @patch("%s.dump" % JSON) @patch("%s.open" % builtins) def test__write(self, mock_open, mock_dump): pf = self.get_obj() self.assertTrue(pf._write()) mock_open.assert_called_with(pf.name, 'wb') mock_dump.assert_called_with(pf.json, mock_open.return_value) @patch("%s.dumps" % JSON) def test_validate_data(self, mock_dumps): pf = self.get_obj() pf.validate_data() mock_dumps.assert_called_with(pf.json) mock_dumps.reset_mock() mock_dumps.side_effect = ValueError self.assertRaises(PluginExecutionError, pf.validate_data) mock_dumps.assert_called_with(pf.json) class TestYAMLPropertyFile(TestFileBacked, TestPropertyFile): test_obj = YAMLPropertyFile @skipUnless(HAS_YAML, "YAML libraries not found, skipping") def setUp(self): TestFileBacked.setUp(self) TestPropertyFile.setUp(self) @patch("yaml.load") def test_Index(self, mock_load): pf = self.get_obj() pf.Index() mock_load.assert_called_with(pf.data) self.assertEqual(pf.yaml, mock_load.return_value) mock_load.reset_mock() mock_load.side_effect = yaml.YAMLError self.assertRaises(PluginExecutionError, pf.Index) mock_load.assert_called_with(pf.data) @patch("yaml.dump") @patch("%s.open" % builtins) def test__write(self, mock_open, mock_dump): pf = self.get_obj() self.assertTrue(pf._write()) mock_open.assert_called_with(pf.name, 'wb') mock_dump.assert_called_with(pf.yaml, mock_open.return_value) @patch("yaml.dump") def test_validate_data(self, mock_dump): pf = self.get_obj() pf.validate_data() mock_dump.assert_called_with(pf.yaml) mock_dump.reset_mock() mock_dump.side_effect = yaml.YAMLError self.assertRaises(PluginExecutionError, pf.validate_data) mock_dump.assert_called_with(pf.yaml) 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) @patch("%s.open" % builtins) def test__write(self, mock_open): pf = self.get_obj() pf.xdata = lxml.etree.Element("Test") self.assertTrue(pf._write()) mock_open.assert_called_with(pf.name, "wb") self.assertXMLEqual(pf.xdata, lxml.etree.XML(mock_open.return_value.write.call_args[0][0])) @patch("os.path.exists") @patch("lxml.etree.XMLSchema") def test_validate_data(self, mock_XMLSchema, mock_exists): pf = self.get_obj() pf.name = os.path.join(datastore, "Properties", "test.xml") schemafile = os.path.join(datastore, "Properties", "test.xsd") def reset(): mock_XMLSchema.reset_mock() mock_exists.reset_mock() # test no schema file mock_exists.return_value = False self.assertTrue(pf.validate_data()) mock_exists.assert_called_with(schemafile) # test schema file exists, valid data reset() mock_exists.return_value = True mock_XMLSchema.return_value = Mock() mock_XMLSchema.return_value.validate.return_value = True self.assertTrue(pf.validate_data()) mock_exists.assert_called_with(schemafile) mock_XMLSchema.assert_called_with(file=schemafile) mock_XMLSchema.return_value.validate.assert_called_with(pf.xdata) # test schema file exists, invalid data reset() mock_XMLSchema.return_value = Mock() mock_XMLSchema.return_value.validate.return_value = False self.assertRaises(PluginExecutionError, pf.validate_data) mock_exists.assert_called_with(schemafile) mock_XMLSchema.assert_called_with(file=schemafile) mock_XMLSchema.return_value.validate.assert_called_with(pf.xdata) # test invalid schema file reset() mock_XMLSchema.side_effect = lxml.etree.XMLSchemaParseError(pf.xdata) self.assertRaises(PluginExecutionError, pf.validate_data) mock_exists.assert_called_with(schemafile) mock_XMLSchema.assert_called_with(file=schemafile) @patch("copy.copy") def test_get_additional_data(self, mock_copy): pf = self.get_obj() pf.setup = Mock() pf.XMLMatch = Mock() metadata = Mock() def reset(): mock_copy.reset_mock() pf.XMLMatch.reset_mock() pf.setup.reset_mock() pf.xdata = lxml.etree.Element("Properties", automatch="true") for Bcfg2.Options.setup.automatch in [True, False]: reset() self.assertEqual(pf.get_additional_data(metadata), pf.XMLMatch.return_value) pf.XMLMatch.assert_called_with(metadata) self.assertFalse(mock_copy.called) pf.xdata = lxml.etree.Element("Properties", automatch="false") for Bcfg2.Options.setup.automatch in [True, False]: reset() self.assertEqual(pf.get_additional_data(metadata), mock_copy.return_value) mock_copy.assert_called_with(pf) self.assertFalse(pf.XMLMatch.called) pf.xdata = lxml.etree.Element("Properties") reset() 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) reset() Bcfg2.Options.setup.automatch = True self.assertEqual(pf.get_additional_data(metadata), pf.XMLMatch.return_value) pf.XMLMatch.assert_called_with(metadata) self.assertFalse(mock_copy.called) class TestProperties(TestPlugin, TestConnector, TestDirectoryBacked): test_obj = Properties testfiles = ['foo.xml', 'bar.baz.xml'] if HAS_JSON: testfiles.extend(["foo.json", "foo.xml.json"]) if HAS_YAML: testfiles.extend(["foo.yaml", "foo.yml", "foo.xml.yml"]) ignore = ['foo.xsd', 'bar.baz.xsd', 'quux.xml.xsd'] badevents = ['bogus.txt'] def get_obj(self, core=None): @patch("%s.%s.add_directory_monitor" % (self.test_obj.__module__, self.test_obj.__name__), Mock()) def inner(): return TestPlugin.get_obj(self, core=core) return inner() @patch("copy.copy") def test_get_additional_data(self, mock_copy): TestConnector.test_get_additional_data(self) p = self.get_obj() metadata = Mock() p.entries = {"foo.xml": Mock(), "foo.yml": Mock()} rv = p.get_additional_data(metadata) expected = dict() for name, entry in p.entries.items(): entry.get_additional_data.assert_called_with(metadata) expected[name] = entry.get_additional_data.return_value self.assertItemsEqual(rv, expected)