cfgmgr_test.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298
  1. # Copyright (C) 2010 Internet Systems Consortium.
  2. #
  3. # Permission to use, copy, modify, and distribute this software for any
  4. # purpose with or without fee is hereby granted, provided that the above
  5. # copyright notice and this permission notice appear in all copies.
  6. #
  7. # THE SOFTWARE IS PROVIDED "AS IS" AND INTERNET SYSTEMS CONSORTIUM
  8. # DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL
  9. # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
  10. # INTERNET SYSTEMS CONSORTIUM BE LIABLE FOR ANY SPECIAL, DIRECT,
  11. # INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING
  12. # FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
  13. # NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
  14. # WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  15. # $Id$
  16. #
  17. # Tests for the configuration manager module
  18. #
  19. import unittest
  20. import os
  21. from isc.config.cfgmgr import *
  22. from isc.config import config_data
  23. from unittest_fakesession import FakeModuleCCSession
  24. class TestConfigManagerData(unittest.TestCase):
  25. def setUp(self):
  26. self.data_path = os.environ['CONFIG_TESTDATA_PATH']
  27. self.config_manager_data = ConfigManagerData(self.data_path)
  28. self.assert_(self.config_manager_data)
  29. def test_init(self):
  30. self.assertEqual(self.config_manager_data.data['version'],
  31. config_data.BIND10_CONFIG_DATA_VERSION)
  32. self.assertEqual(self.config_manager_data.data_path,
  33. self.data_path)
  34. self.assertEqual(self.config_manager_data.db_filename,
  35. self.data_path + os.sep + "b10-config.db")
  36. def test_read_from_file(self):
  37. ConfigManagerData.read_from_file(self.data_path)
  38. self.assertRaises(ConfigManagerDataEmpty,
  39. ConfigManagerData.read_from_file,
  40. "doesnotexist")
  41. self.assertRaises(ConfigManagerDataReadError,
  42. ConfigManagerData.read_from_file,
  43. self.data_path, "b10-config-bad1.db")
  44. self.assertRaises(ConfigManagerDataReadError,
  45. ConfigManagerData.read_from_file,
  46. self.data_path, "b10-config-bad2.db")
  47. self.assertRaises(ConfigManagerDataReadError,
  48. ConfigManagerData.read_from_file,
  49. self.data_path, "b10-config-bad3.db")
  50. self.assertRaises(ConfigManagerDataReadError,
  51. ConfigManagerData.read_from_file,
  52. self.data_path, "b10-config-bad4.db")
  53. def test_write_to_file(self):
  54. output_file_name = "b10-config-write-test";
  55. self.config_manager_data.write_to_file(output_file_name)
  56. new_config = ConfigManagerData(self.data_path, output_file_name)
  57. self.assertEqual(self.config_manager_data, new_config)
  58. def test_equality(self):
  59. # tests the __eq__ function. Equality is only defined
  60. # by equality of the .data element. If data_path or db_filename
  61. # are different, but the contents are the same, it's still
  62. # considered equal
  63. cfd1 = ConfigManagerData(self.data_path)
  64. cfd2 = ConfigManagerData(self.data_path)
  65. self.assertEqual(cfd1, cfd2)
  66. cfd2.data_path = "some/unknown/path"
  67. self.assertEqual(cfd1, cfd2)
  68. cfd2.db_filename = "bad_file.name"
  69. self.assertEqual(cfd1, cfd2)
  70. cfd2.data['test'] = { 'a': [ 1, 2, 3]}
  71. self.assertNotEqual(cfd1, cfd2)
  72. class TestConfigManager(unittest.TestCase):
  73. def setUp(self):
  74. self.data_path = os.environ['CONFIG_TESTDATA_PATH']
  75. self.fake_session = FakeModuleCCSession()
  76. self.cm = ConfigManager(self.data_path, self.fake_session)
  77. self.name = "TestModule"
  78. self.spec = isc.config.module_spec_from_file(self.data_path + os.sep + "/spec2.spec")
  79. def test_init(self):
  80. self.assert_(self.cm.module_specs == {})
  81. self.assert_(self.cm.data_path == self.data_path)
  82. self.assert_(self.cm.config != None)
  83. self.assert_(self.fake_session.has_subscription("ConfigManager"))
  84. self.assert_(self.fake_session.has_subscription("Boss", "ConfigManager"))
  85. self.assertFalse(self.cm.running)
  86. def test_notify_boss(self):
  87. self.cm.notify_boss()
  88. msg = self.fake_session.get_message("Boss", None)
  89. self.assert_(msg)
  90. # this one is actually wrong, but 'current status quo'
  91. self.assertEqual(msg, {"running": "configmanager"})
  92. def test_set_module_spec(self):
  93. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
  94. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  95. self.cm.set_module_spec(module_spec)
  96. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  97. def test_remove_module_spec(self):
  98. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
  99. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  100. self.cm.set_module_spec(module_spec)
  101. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  102. self.cm.remove_module_spec(module_spec.get_module_name())
  103. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  104. def test_get_module_spec(self):
  105. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
  106. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  107. self.cm.set_module_spec(module_spec)
  108. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  109. module_spec2 = self.cm.get_module_spec(module_spec.get_module_name())
  110. self.assertEqual(module_spec, module_spec2)
  111. def test_get_config_spec(self):
  112. config_spec = self.cm.get_config_spec()
  113. self.assertEqual(config_spec, {})
  114. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
  115. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  116. self.cm.set_module_spec(module_spec)
  117. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  118. config_spec = self.cm.get_config_spec()
  119. self.assertEqual(config_spec, { 'Spec1': None })
  120. self.cm.remove_module_spec('Spec1')
  121. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec2.spec")
  122. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  123. self.cm.set_module_spec(module_spec)
  124. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  125. config_spec = self.cm.get_config_spec()
  126. self.assertEqual(config_spec['Spec2'], module_spec.get_config_spec())
  127. config_spec = self.cm.get_config_spec('Spec2')
  128. self.assertEqual(config_spec['Spec2'], module_spec.get_config_spec())
  129. def test_get_commands_spec(self):
  130. commands_spec = self.cm.get_commands_spec()
  131. self.assertEqual(commands_spec, {})
  132. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
  133. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  134. self.cm.set_module_spec(module_spec)
  135. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  136. commands_spec = self.cm.get_commands_spec()
  137. self.assertEqual(commands_spec, { 'Spec1': None })
  138. self.cm.remove_module_spec('Spec1')
  139. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec2.spec")
  140. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  141. self.cm.set_module_spec(module_spec)
  142. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  143. commands_spec = self.cm.get_commands_spec()
  144. self.assertEqual(commands_spec['Spec2'], module_spec.get_commands_spec())
  145. commands_spec = self.cm.get_commands_spec('Spec2')
  146. self.assertEqual(commands_spec['Spec2'], module_spec.get_commands_spec())
  147. def test_read_config(self):
  148. self.assertEqual(self.cm.config.data, {'version': config_data.BIND10_CONFIG_DATA_VERSION})
  149. self.cm.read_config()
  150. # due to what get written, the value here is what the last set_config command in test_handle_msg does
  151. self.assertEqual(self.cm.config.data, {'TestModule': {'test': 125}, 'version': config_data.BIND10_CONFIG_DATA_VERSION})
  152. self.cm.data_path = "/no_such_path"
  153. self.cm.read_config()
  154. self.assertEqual(self.cm.config.data, {'version': config_data.BIND10_CONFIG_DATA_VERSION})
  155. def test_write_config(self):
  156. # tested in ConfigManagerData tests
  157. pass
  158. def _handle_msg_helper(self, msg, expected_answer):
  159. answer = self.cm.handle_msg(msg)
  160. self.assertEqual(expected_answer, answer)
  161. def test_handle_msg(self):
  162. self._handle_msg_helper({}, { 'result': [ 1, 'Unknown message format: {}']})
  163. self._handle_msg_helper("", { 'result': [ 1, 'Unknown message format: ']})
  164. self._handle_msg_helper({ "command": [ "badcommand" ] }, { 'result': [ 1, "Unknown command: badcommand"]})
  165. self._handle_msg_helper({ "command": [ "get_commands_spec" ] }, { 'result': [ 0, {} ]})
  166. self._handle_msg_helper({ "command": [ "get_module_spec" ] }, { 'result': [ 0, {} ]})
  167. self._handle_msg_helper({ "command": [ "get_module_spec", { "module_name": "Spec2" } ] }, { 'result': [ 0, {} ]})
  168. #self._handle_msg_helper({ "command": [ "get_module_spec", { "module_name": "nosuchmodule" } ] },
  169. # {'result': [1, 'No specification for module nosuchmodule']})
  170. self._handle_msg_helper({ "command": [ "get_module_spec", 1 ] },
  171. {'result': [1, 'Bad get_module_spec command, argument not a dict']})
  172. self._handle_msg_helper({ "command": [ "get_module_spec", { } ] },
  173. {'result': [1, 'Bad module_name in get_module_spec command']})
  174. self._handle_msg_helper({ "command": [ "get_config" ] }, { 'result': [ 0, { 'version': config_data.BIND10_CONFIG_DATA_VERSION } ]})
  175. self._handle_msg_helper({ "command": [ "get_config", { "module_name": "nosuchmodule" } ] },
  176. {'result': [0, { 'version': config_data.BIND10_CONFIG_DATA_VERSION }]})
  177. self._handle_msg_helper({ "command": [ "get_config", 1 ] },
  178. {'result': [1, 'Bad get_config command, argument not a dict']})
  179. self._handle_msg_helper({ "command": [ "get_config", { } ] },
  180. {'result': [1, 'Bad module_name in get_config command']})
  181. self._handle_msg_helper({ "command": [ "set_config" ] },
  182. {'result': [1, 'Wrong number of arguments']})
  183. self._handle_msg_helper({ "command": [ "set_config", [{}]] },
  184. {'result': [0]})
  185. self.assertEqual(len(self.fake_session.message_queue), 0)
  186. # the targets of some of these tests expect specific answers, put
  187. # those in our fake msgq first.
  188. my_ok_answer = { 'result': [ 0 ] }
  189. # Send the 'ok' that cfgmgr expects back to the fake queue first
  190. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  191. # then send the command
  192. self._handle_msg_helper({ "command": [ "set_config", [self.name, { "test": 123 }] ] },
  193. my_ok_answer)
  194. # The cfgmgr should have eaten the ok message, and sent out an update again
  195. self.assertEqual(len(self.fake_session.message_queue), 1)
  196. self.assertEqual({'command': [ 'config_update', {'test': 123}]},
  197. self.fake_session.get_message(self.name, None))
  198. # and the queue should now be empty again
  199. self.assertEqual(len(self.fake_session.message_queue), 0)
  200. # below are variations of the theme above
  201. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  202. self._handle_msg_helper({ "command": [ "set_config", [self.name, { "test": 124 }] ] },
  203. my_ok_answer)
  204. self.assertEqual(len(self.fake_session.message_queue), 1)
  205. self.assertEqual({'command': [ 'config_update', {'test': 124}]},
  206. self.fake_session.get_message(self.name, None))
  207. self.assertEqual(len(self.fake_session.message_queue), 0)
  208. # This is the last 'succes' one, the value set here is what test_read_config expects
  209. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  210. self._handle_msg_helper({ "command": [ "set_config", [ { self.name: { "test": 125 } }] ] },
  211. my_ok_answer )
  212. self.assertEqual(len(self.fake_session.message_queue), 1)
  213. self.assertEqual({'command': [ 'config_update', {'test': 125}]},
  214. self.fake_session.get_message(self.name, None))
  215. self.assertEqual(len(self.fake_session.message_queue), 0)
  216. my_bad_answer = { 'result': [1, "bad_answer"] }
  217. self.fake_session.group_sendmsg(my_bad_answer, "ConfigManager")
  218. self._handle_msg_helper({ "command": [ "set_config", [ self.name, { "test": 125 }] ] },
  219. my_bad_answer )
  220. self.assertEqual(len(self.fake_session.message_queue), 1)
  221. self.assertEqual({'command': [ 'config_update', {'test': 125}]},
  222. self.fake_session.get_message(self.name, None))
  223. self.assertEqual(len(self.fake_session.message_queue), 0)
  224. self._handle_msg_helper({ "command": [ "set_config", [ ] ] },
  225. {'result': [1, 'Wrong number of arguments']} )
  226. self._handle_msg_helper({ "command": [ "set_config", [ self.name, { "test": 125 }] ] },
  227. { 'result': [1, 'No answer message from TestModule']} )
  228. #self.assertEqual(len(self.fake_session.message_queue), 1)
  229. #self.assertEqual({'config_update': {'test': 124}},
  230. # self.fake_session.get_message(self.name, None))
  231. #self.assertEqual({'version': 1, 'TestModule': {'test': 124}}, self.cm.config.data)
  232. #
  233. self._handle_msg_helper({ "command":
  234. ["module_spec", self.spec.get_full_spec()]
  235. },
  236. {'result': [0]})
  237. self._handle_msg_helper({ "command": [ "module_spec", { 'foo': 1 } ] },
  238. {'result': [1, 'Error in data definition: no module_name in module_spec']})
  239. self._handle_msg_helper({ "command": [ "get_module_spec" ] }, { 'result': [ 0, { self.spec.get_module_name(): self.spec.get_full_spec() } ]})
  240. self._handle_msg_helper({ "command": [ "get_commands_spec" ] }, { 'result': [ 0, { self.spec.get_module_name(): self.spec.get_commands_spec() } ]})
  241. # re-add this once we have new way to propagate spec changes (1 instead of the current 2 messages)
  242. #self.assertEqual(len(self.fake_session.message_queue), 2)
  243. # the name here is actually wrong (and hardcoded), but needed in the current version
  244. # TODO: fix that
  245. #self.assertEqual({'specification_update': [ self.name, self.spec ] },
  246. # self.fake_session.get_message("Cmdctl", None))
  247. #self.assertEqual({'commands_update': [ self.name, self.commands ] },
  248. # self.fake_session.get_message("Cmdctl", None))
  249. self._handle_msg_helper({ "command":
  250. ["shutdown"]
  251. },
  252. {'result': [0]})
  253. def test_run(self):
  254. self.fake_session.group_sendmsg({ "command": [ "get_commands_spec" ] }, "ConfigManager")
  255. self.cm.run()
  256. pass
  257. if __name__ == '__main__':
  258. if not 'CONFIG_TESTDATA_PATH' in os.environ:
  259. print("You need to set the environment variable CONFIG_TESTDATA_PATH to point to the directory containing the test data files")
  260. exit(1)
  261. unittest.main()