# Copyright (C) 2010 Internet Systems Consortium. # # Permission to use, copy, modify, and distribute this software for any # purpose with or without fee is hereby granted, provided that the above # copyright notice and this permission notice appear in all copies. # # THE SOFTWARE IS PROVIDED "AS IS" AND INTERNET SYSTEMS CONSORTIUM # DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL # INTERNET SYSTEMS CONSORTIUM BE LIABLE FOR ANY SPECIAL, DIRECT, # INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING # FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, # NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION # WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. # $Id$ # # Tests for the configuration manager module # import unittest import os from isc.config.cfgmgr import * from isc.config import config_data from unittest_fakesession import FakeModuleCCSession class TestConfigManagerData(unittest.TestCase): def setUp(self): self.data_path = os.environ['CONFIG_TESTDATA_PATH'] self.writable_data_path = os.environ['CONFIG_WR_TESTDATA_PATH'] self.config_manager_data = ConfigManagerData(self.writable_data_path) self.assert_(self.config_manager_data) def test_init(self): self.assertEqual(self.config_manager_data.data['version'], config_data.BIND10_CONFIG_DATA_VERSION) self.assertEqual(self.config_manager_data.data_path, self.writable_data_path) self.assertEqual(self.config_manager_data.db_filename, self.writable_data_path + os.sep + "b10-config.db") def test_read_from_file(self): ConfigManagerData.read_from_file(self.writable_data_path) self.assertRaises(ConfigManagerDataEmpty, ConfigManagerData.read_from_file, "doesnotexist") self.assertRaises(ConfigManagerDataReadError, ConfigManagerData.read_from_file, self.data_path, "b10-config-bad1.db") self.assertRaises(ConfigManagerDataReadError, ConfigManagerData.read_from_file, self.data_path, "b10-config-bad2.db") self.assertRaises(ConfigManagerDataReadError, ConfigManagerData.read_from_file, self.data_path, "b10-config-bad3.db") self.assertRaises(ConfigManagerDataReadError, ConfigManagerData.read_from_file, self.data_path, "b10-config-bad4.db") def test_write_to_file(self): output_file_name = "b10-config-write-test" self.config_manager_data.write_to_file(output_file_name) new_config = ConfigManagerData(self.data_path, output_file_name) self.assertEqual(self.config_manager_data, new_config) os.remove(output_file_name) def test_equality(self): # tests the __eq__ function. Equality is only defined # by equality of the .data element. If data_path or db_filename # are different, but the contents are the same, it's still # considered equal cfd1 = ConfigManagerData(self.data_path) cfd2 = ConfigManagerData(self.data_path) self.assertEqual(cfd1, cfd2) cfd2.data_path = "some/unknown/path" self.assertEqual(cfd1, cfd2) cfd2.db_filename = "bad_file.name" self.assertEqual(cfd1, cfd2) cfd2.data['test'] = { 'a': [ 1, 2, 3]} self.assertNotEqual(cfd1, cfd2) class TestConfigManager(unittest.TestCase): def setUp(self): self.data_path = os.environ['CONFIG_TESTDATA_PATH'] self.writable_data_path = os.environ['CONFIG_WR_TESTDATA_PATH'] self.fake_session = FakeModuleCCSession() self.cm = ConfigManager(self.writable_data_path, self.fake_session) self.name = "TestModule" self.spec = isc.config.module_spec_from_file(self.data_path + os.sep + "/spec2.spec") def test_init(self): self.assert_(self.cm.module_specs == {}) self.assert_(self.cm.data_path == self.writable_data_path) self.assert_(self.cm.config != None) self.assert_(self.fake_session.has_subscription("ConfigManager")) self.assert_(self.fake_session.has_subscription("Boss", "ConfigManager")) self.assertFalse(self.cm.running) def test_notify_boss(self): self.cm.notify_boss() msg = self.fake_session.get_message("Boss", None) self.assert_(msg) # this one is actually wrong, but 'current status quo' self.assertEqual(msg, {"running": "configmanager"}) def test_set_module_spec(self): module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec") self.assert_(module_spec.get_module_name() not in self.cm.module_specs) self.cm.set_module_spec(module_spec) self.assert_(module_spec.get_module_name() in self.cm.module_specs) def test_remove_module_spec(self): module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec") self.assert_(module_spec.get_module_name() not in self.cm.module_specs) self.cm.set_module_spec(module_spec) self.assert_(module_spec.get_module_name() in self.cm.module_specs) self.cm.remove_module_spec(module_spec.get_module_name()) self.assert_(module_spec.get_module_name() not in self.cm.module_specs) def test_get_module_spec(self): module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec") self.assert_(module_spec.get_module_name() not in self.cm.module_specs) self.cm.set_module_spec(module_spec) self.assert_(module_spec.get_module_name() in self.cm.module_specs) module_spec2 = self.cm.get_module_spec(module_spec.get_module_name()) self.assertEqual(module_spec, module_spec2) def test_get_config_spec(self): config_spec = self.cm.get_config_spec() self.assertEqual(config_spec, {}) module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec") self.assert_(module_spec.get_module_name() not in self.cm.module_specs) self.cm.set_module_spec(module_spec) self.assert_(module_spec.get_module_name() in self.cm.module_specs) config_spec = self.cm.get_config_spec() self.assertEqual(config_spec, { 'Spec1': None }) self.cm.remove_module_spec('Spec1') module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec2.spec") self.assert_(module_spec.get_module_name() not in self.cm.module_specs) self.cm.set_module_spec(module_spec) self.assert_(module_spec.get_module_name() in self.cm.module_specs) config_spec = self.cm.get_config_spec() self.assertEqual(config_spec['Spec2'], module_spec.get_config_spec()) config_spec = self.cm.get_config_spec('Spec2') self.assertEqual(config_spec['Spec2'], module_spec.get_config_spec()) def test_get_commands_spec(self): commands_spec = self.cm.get_commands_spec() self.assertEqual(commands_spec, {}) module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec") self.assert_(module_spec.get_module_name() not in self.cm.module_specs) self.cm.set_module_spec(module_spec) self.assert_(module_spec.get_module_name() in self.cm.module_specs) commands_spec = self.cm.get_commands_spec() self.assertEqual(commands_spec, { 'Spec1': None }) self.cm.remove_module_spec('Spec1') module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec2.spec") self.assert_(module_spec.get_module_name() not in self.cm.module_specs) self.cm.set_module_spec(module_spec) self.assert_(module_spec.get_module_name() in self.cm.module_specs) commands_spec = self.cm.get_commands_spec() self.assertEqual(commands_spec['Spec2'], module_spec.get_commands_spec()) commands_spec = self.cm.get_commands_spec('Spec2') self.assertEqual(commands_spec['Spec2'], module_spec.get_commands_spec()) def test_read_config(self): self.assertEqual(self.cm.config.data, {'version': config_data.BIND10_CONFIG_DATA_VERSION}) self.cm.read_config() # due to what get written, the value here is what the last set_config command in test_handle_msg does self.assertEqual(self.cm.config.data, {'TestModule': {'test': 125}, 'version': config_data.BIND10_CONFIG_DATA_VERSION}) self.cm.data_path = "/no_such_path" self.cm.read_config() self.assertEqual(self.cm.config.data, {'version': config_data.BIND10_CONFIG_DATA_VERSION}) def test_write_config(self): # tested in ConfigManagerData tests pass def _handle_msg_helper(self, msg, expected_answer): answer = self.cm.handle_msg(msg) self.assertEqual(expected_answer, answer) def test_handle_msg(self): self._handle_msg_helper({}, { 'result': [ 1, 'Unknown message format: {}']}) self._handle_msg_helper("", { 'result': [ 1, 'Unknown message format: ']}) self._handle_msg_helper({ "command": [ "badcommand" ] }, { 'result': [ 1, "Unknown command: badcommand"]}) self._handle_msg_helper({ "command": [ "get_commands_spec" ] }, { 'result': [ 0, {} ]}) self._handle_msg_helper({ "command": [ "get_module_spec" ] }, { 'result': [ 0, {} ]}) self._handle_msg_helper({ "command": [ "get_module_spec", { "module_name": "Spec2" } ] }, { 'result': [ 0, {} ]}) #self._handle_msg_helper({ "command": [ "get_module_spec", { "module_name": "nosuchmodule" } ] }, # {'result': [1, 'No specification for module nosuchmodule']}) self._handle_msg_helper({ "command": [ "get_module_spec", 1 ] }, {'result': [1, 'Bad get_module_spec command, argument not a dict']}) self._handle_msg_helper({ "command": [ "get_module_spec", { } ] }, {'result': [1, 'Bad module_name in get_module_spec command']}) self._handle_msg_helper({ "command": [ "get_config" ] }, { 'result': [ 0, { 'version': config_data.BIND10_CONFIG_DATA_VERSION } ]}) self._handle_msg_helper({ "command": [ "get_config", { "module_name": "nosuchmodule" } ] }, {'result': [0, { 'version': config_data.BIND10_CONFIG_DATA_VERSION }]}) self._handle_msg_helper({ "command": [ "get_config", 1 ] }, {'result': [1, 'Bad get_config command, argument not a dict']}) self._handle_msg_helper({ "command": [ "get_config", { } ] }, {'result': [1, 'Bad module_name in get_config command']}) self._handle_msg_helper({ "command": [ "set_config" ] }, {'result': [1, 'Wrong number of arguments']}) self._handle_msg_helper({ "command": [ "set_config", [{}]] }, {'result': [0]}) self.assertEqual(len(self.fake_session.message_queue), 0) # the targets of some of these tests expect specific answers, put # those in our fake msgq first. my_ok_answer = { 'result': [ 0 ] } # Send the 'ok' that cfgmgr expects back to the fake queue first self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager") # then send the command self._handle_msg_helper({ "command": [ "set_config", [self.name, { "test": 123 }] ] }, my_ok_answer) # The cfgmgr should have eaten the ok message, and sent out an update again self.assertEqual(len(self.fake_session.message_queue), 1) self.assertEqual({'command': [ 'config_update', {'test': 123}]}, self.fake_session.get_message(self.name, None)) # and the queue should now be empty again self.assertEqual(len(self.fake_session.message_queue), 0) # below are variations of the theme above self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager") self._handle_msg_helper({ "command": [ "set_config", [self.name, { "test": 124 }] ] }, my_ok_answer) self.assertEqual(len(self.fake_session.message_queue), 1) self.assertEqual({'command': [ 'config_update', {'test': 124}]}, self.fake_session.get_message(self.name, None)) self.assertEqual(len(self.fake_session.message_queue), 0) # This is the last 'succes' one, the value set here is what test_read_config expects self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager") self._handle_msg_helper({ "command": [ "set_config", [ { self.name: { "test": 125 } }] ] }, my_ok_answer ) self.assertEqual(len(self.fake_session.message_queue), 1) self.assertEqual({'command': [ 'config_update', {'test': 125}]}, self.fake_session.get_message(self.name, None)) self.assertEqual(len(self.fake_session.message_queue), 0) my_bad_answer = { 'result': [1, "bad_answer"] } self.fake_session.group_sendmsg(my_bad_answer, "ConfigManager") self._handle_msg_helper({ "command": [ "set_config", [ self.name, { "test": 125 }] ] }, my_bad_answer ) self.assertEqual(len(self.fake_session.message_queue), 1) self.assertEqual({'command': [ 'config_update', {'test': 125}]}, self.fake_session.get_message(self.name, None)) self.assertEqual(len(self.fake_session.message_queue), 0) self._handle_msg_helper({ "command": [ "set_config", [ ] ] }, {'result': [1, 'Wrong number of arguments']} ) self._handle_msg_helper({ "command": [ "set_config", [ self.name, { "test": 125 }] ] }, { 'result': [1, 'No answer message from TestModule']} ) #self.assertEqual(len(self.fake_session.message_queue), 1) #self.assertEqual({'config_update': {'test': 124}}, # self.fake_session.get_message(self.name, None)) #self.assertEqual({'version': 1, 'TestModule': {'test': 124}}, self.cm.config.data) # self._handle_msg_helper({ "command": ["module_spec", self.spec.get_full_spec()] }, {'result': [0]}) self._handle_msg_helper({ "command": [ "module_spec", { 'foo': 1 } ] }, {'result': [1, 'Error in data definition: no module_name in module_spec']}) self._handle_msg_helper({ "command": [ "get_module_spec" ] }, { 'result': [ 0, { self.spec.get_module_name(): self.spec.get_full_spec() } ]}) self._handle_msg_helper({ "command": [ "get_commands_spec" ] }, { 'result': [ 0, { self.spec.get_module_name(): self.spec.get_commands_spec() } ]}) # re-add this once we have new way to propagate spec changes (1 instead of the current 2 messages) #self.assertEqual(len(self.fake_session.message_queue), 2) # the name here is actually wrong (and hardcoded), but needed in the current version # TODO: fix that #self.assertEqual({'specification_update': [ self.name, self.spec ] }, # self.fake_session.get_message("Cmdctl", None)) #self.assertEqual({'commands_update': [ self.name, self.commands ] }, # self.fake_session.get_message("Cmdctl", None)) self._handle_msg_helper({ "command": ["shutdown"] }, {'result': [0]}) def test_run(self): self.fake_session.group_sendmsg({ "command": [ "get_commands_spec" ] }, "ConfigManager") self.fake_session.group_sendmsg({ "command": [ "shutdown" ] }, "ConfigManager") self.cm.run() pass if __name__ == '__main__': if not 'CONFIG_TESTDATA_PATH' in os.environ or not 'CONFIG_WR_TESTDATA_PATH' in os.environ: print("You need to set the environment variable CONFIG_TESTDATA_PATH and CONFIG_WR_TESTDATA_PATH to point to the directory containing the test data files") exit(1) unittest.main()