cfgmgr_test.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387
  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. #
  16. # Tests for the configuration manager module
  17. #
  18. import unittest
  19. import os
  20. from isc.config.cfgmgr import *
  21. from isc.config import config_data
  22. from unittest_fakesession import FakeModuleCCSession
  23. class TestConfigManagerData(unittest.TestCase):
  24. def setUp(self):
  25. self.data_path = os.environ['CONFIG_TESTDATA_PATH']
  26. self.writable_data_path = os.environ['CONFIG_WR_TESTDATA_PATH']
  27. self.config_manager_data = ConfigManagerData(self.writable_data_path,
  28. file_name="b10-config.db")
  29. self.assert_(self.config_manager_data)
  30. def test_abs_file(self):
  31. """
  32. Test what happens if we give the config manager an absolute path.
  33. It shouldn't append the data path to it.
  34. """
  35. abs_path = self.data_path + os.sep + "b10-config-imaginary.db"
  36. data = ConfigManagerData(os.getcwd(), abs_path)
  37. self.assertEqual(abs_path, data.db_filename)
  38. self.assertEqual(self.data_path, data.data_path)
  39. def test_init(self):
  40. self.assertEqual(self.config_manager_data.data['version'],
  41. config_data.BIND10_CONFIG_DATA_VERSION)
  42. self.assertEqual(self.config_manager_data.data_path,
  43. self.writable_data_path)
  44. self.assertEqual(self.config_manager_data.db_filename,
  45. self.writable_data_path + os.sep + "b10-config.db")
  46. def test_read_from_file(self):
  47. ConfigManagerData.read_from_file(self.writable_data_path, "b10-config.db")
  48. self.assertRaises(ConfigManagerDataEmpty,
  49. ConfigManagerData.read_from_file,
  50. "doesnotexist", "b10-config.db")
  51. self.assertRaises(ConfigManagerDataReadError,
  52. ConfigManagerData.read_from_file,
  53. self.data_path, "b10-config-bad1.db")
  54. self.assertRaises(ConfigManagerDataReadError,
  55. ConfigManagerData.read_from_file,
  56. self.data_path, "b10-config-bad2.db")
  57. self.assertRaises(ConfigManagerDataReadError,
  58. ConfigManagerData.read_from_file,
  59. self.data_path, "b10-config-bad3.db")
  60. self.assertRaises(ConfigManagerDataReadError,
  61. ConfigManagerData.read_from_file,
  62. self.data_path, "b10-config-bad4.db")
  63. def test_write_to_file(self):
  64. output_file_name = "b10-config-write-test"
  65. self.config_manager_data.write_to_file(output_file_name)
  66. new_config = ConfigManagerData(self.data_path, output_file_name)
  67. self.assertEqual(self.config_manager_data, new_config)
  68. os.remove(output_file_name)
  69. def test_equality(self):
  70. # tests the __eq__ function. Equality is only defined
  71. # by equality of the .data element. If data_path or db_filename
  72. # are different, but the contents are the same, it's still
  73. # considered equal
  74. cfd1 = ConfigManagerData(self.data_path, file_name="b10-config.db")
  75. cfd2 = ConfigManagerData(self.data_path, file_name="b10-config.db")
  76. self.assertEqual(cfd1, cfd2)
  77. cfd2.data_path = "some/unknown/path"
  78. self.assertEqual(cfd1, cfd2)
  79. cfd2.db_filename = "bad_file.name"
  80. self.assertEqual(cfd1, cfd2)
  81. cfd2.data['test'] = { 'a': [ 1, 2, 3]}
  82. self.assertNotEqual(cfd1, cfd2)
  83. class TestConfigManager(unittest.TestCase):
  84. def setUp(self):
  85. self.data_path = os.environ['CONFIG_TESTDATA_PATH']
  86. self.writable_data_path = os.environ['CONFIG_WR_TESTDATA_PATH']
  87. self.fake_session = FakeModuleCCSession()
  88. self.cm = ConfigManager(self.writable_data_path,
  89. database_filename="b10-config.db",
  90. session=self.fake_session)
  91. self.name = "TestModule"
  92. self.spec = isc.config.module_spec_from_file(self.data_path + os.sep + "/spec2.spec")
  93. def test_paths(self):
  94. """
  95. Test data_path and database filename is passed trough to
  96. underlying ConfigManagerData.
  97. """
  98. cm = ConfigManager("datapath", "filename", self.fake_session)
  99. self.assertEqual("datapath" + os.sep + "filename",
  100. cm.config.db_filename)
  101. # It should preserve it while reading
  102. cm.read_config()
  103. self.assertEqual("datapath" + os.sep + "filename",
  104. cm.config.db_filename)
  105. def test_init(self):
  106. self.assert_(self.cm.module_specs == {})
  107. self.assert_(self.cm.data_path == self.writable_data_path)
  108. self.assert_(self.cm.config != None)
  109. self.assert_(self.fake_session.has_subscription("ConfigManager"))
  110. self.assert_(self.fake_session.has_subscription("Boss", "ConfigManager"))
  111. self.assertFalse(self.cm.running)
  112. def test_notify_boss(self):
  113. self.cm.notify_boss()
  114. msg = self.fake_session.get_message("Boss", None)
  115. self.assert_(msg)
  116. # this one is actually wrong, but 'current status quo'
  117. self.assertEqual(msg, {"running": "configmanager"})
  118. def test_set_module_spec(self):
  119. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
  120. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  121. self.cm.set_module_spec(module_spec)
  122. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  123. def test_remove_module_spec(self):
  124. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
  125. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  126. self.cm.set_module_spec(module_spec)
  127. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  128. self.cm.remove_module_spec(module_spec.get_module_name())
  129. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  130. def test_get_module_spec(self):
  131. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
  132. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  133. self.cm.set_module_spec(module_spec)
  134. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  135. module_spec2 = self.cm.get_module_spec(module_spec.get_module_name())
  136. self.assertEqual(module_spec, module_spec2)
  137. def test_get_config_spec(self):
  138. config_spec = self.cm.get_config_spec()
  139. self.assertEqual(config_spec, {})
  140. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
  141. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  142. self.cm.set_module_spec(module_spec)
  143. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  144. config_spec = self.cm.get_config_spec()
  145. self.assertEqual(config_spec, { 'Spec1': None })
  146. self.cm.remove_module_spec('Spec1')
  147. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec2.spec")
  148. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  149. self.cm.set_module_spec(module_spec)
  150. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  151. config_spec = self.cm.get_config_spec()
  152. self.assertEqual(config_spec['Spec2'], module_spec.get_config_spec())
  153. config_spec = self.cm.get_config_spec('Spec2')
  154. self.assertEqual(config_spec['Spec2'], module_spec.get_config_spec())
  155. def test_get_commands_spec(self):
  156. commands_spec = self.cm.get_commands_spec()
  157. self.assertEqual(commands_spec, {})
  158. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec1.spec")
  159. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  160. self.cm.set_module_spec(module_spec)
  161. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  162. commands_spec = self.cm.get_commands_spec()
  163. self.assertEqual(commands_spec, { 'Spec1': None })
  164. self.cm.remove_module_spec('Spec1')
  165. module_spec = isc.config.module_spec.module_spec_from_file(self.data_path + os.sep + "spec2.spec")
  166. self.assert_(module_spec.get_module_name() not in self.cm.module_specs)
  167. self.cm.set_module_spec(module_spec)
  168. self.assert_(module_spec.get_module_name() in self.cm.module_specs)
  169. commands_spec = self.cm.get_commands_spec()
  170. self.assertEqual(commands_spec['Spec2'], module_spec.get_commands_spec())
  171. commands_spec = self.cm.get_commands_spec('Spec2')
  172. self.assertEqual(commands_spec['Spec2'], module_spec.get_commands_spec())
  173. def test_read_config(self):
  174. self.assertEqual(self.cm.config.data, {'version': config_data.BIND10_CONFIG_DATA_VERSION})
  175. self.cm.read_config()
  176. # due to what get written, the value here is what the last set_config command in test_handle_msg does
  177. self.assertEqual(self.cm.config.data, {'TestModule': {'test': 125}, 'version': config_data.BIND10_CONFIG_DATA_VERSION})
  178. self.cm.data_path = "/no_such_path"
  179. self.cm.read_config()
  180. self.assertEqual(self.cm.config.data, {'version': config_data.BIND10_CONFIG_DATA_VERSION})
  181. def test_write_config(self):
  182. # tested in ConfigManagerData tests
  183. pass
  184. def _handle_msg_helper(self, msg, expected_answer):
  185. answer = self.cm.handle_msg(msg)
  186. self.assertEqual(expected_answer, answer)
  187. def test_handle_msg(self):
  188. self._handle_msg_helper({}, { 'result': [ 1, 'Unknown message format: {}']})
  189. self._handle_msg_helper("", { 'result': [ 1, 'Unknown message format: ']})
  190. self._handle_msg_helper({ "command": [ "badcommand" ] }, { 'result': [ 1, "Unknown command: badcommand"]})
  191. self._handle_msg_helper({ "command": [ "get_commands_spec" ] }, { 'result': [ 0, {} ]})
  192. self._handle_msg_helper({ "command": [ "get_module_spec" ] }, { 'result': [ 0, {} ]})
  193. self._handle_msg_helper({ "command": [ "get_module_spec", { "module_name": "Spec2" } ] }, { 'result': [ 0, {} ]})
  194. #self._handle_msg_helper({ "command": [ "get_module_spec", { "module_name": "nosuchmodule" } ] },
  195. # {'result': [1, 'No specification for module nosuchmodule']})
  196. self._handle_msg_helper({ "command": [ "get_module_spec", 1 ] },
  197. {'result': [1, 'Bad get_module_spec command, argument not a dict']})
  198. self._handle_msg_helper({ "command": [ "get_module_spec", { } ] },
  199. {'result': [1, 'Bad module_name in get_module_spec command']})
  200. self._handle_msg_helper({ "command": [ "get_config" ] }, { 'result': [ 0, { 'version': config_data.BIND10_CONFIG_DATA_VERSION } ]})
  201. self._handle_msg_helper({ "command": [ "get_config", { "module_name": "nosuchmodule" } ] },
  202. {'result': [0, { 'version': config_data.BIND10_CONFIG_DATA_VERSION }]})
  203. self._handle_msg_helper({ "command": [ "get_config", 1 ] },
  204. {'result': [1, 'Bad get_config command, argument not a dict']})
  205. self._handle_msg_helper({ "command": [ "get_config", { } ] },
  206. {'result': [1, 'Bad module_name in get_config command']})
  207. self._handle_msg_helper({ "command": [ "set_config" ] },
  208. {'result': [1, 'Wrong number of arguments']})
  209. self._handle_msg_helper({ "command": [ "set_config", [{}]] },
  210. {'result': [0]})
  211. self.assertEqual(len(self.fake_session.message_queue), 0)
  212. # the targets of some of these tests expect specific answers, put
  213. # those in our fake msgq first.
  214. my_ok_answer = { 'result': [ 0 ] }
  215. # Send the 'ok' that cfgmgr expects back to the fake queue first
  216. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  217. # then send the command
  218. self._handle_msg_helper({ "command": [ "set_config", [self.name, { "test": 123 }] ] },
  219. my_ok_answer)
  220. # The cfgmgr should have eaten the ok message, and sent out an update again
  221. self.assertEqual(len(self.fake_session.message_queue), 1)
  222. self.assertEqual({'command': [ 'config_update', {'test': 123}]},
  223. self.fake_session.get_message(self.name, None))
  224. # and the queue should now be empty again
  225. self.assertEqual(len(self.fake_session.message_queue), 0)
  226. # below are variations of the theme above
  227. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  228. self._handle_msg_helper({ "command": [ "set_config", [self.name, { "test": 124 }] ] },
  229. my_ok_answer)
  230. self.assertEqual(len(self.fake_session.message_queue), 1)
  231. self.assertEqual({'command': [ 'config_update', {'test': 124}]},
  232. self.fake_session.get_message(self.name, None))
  233. self.assertEqual(len(self.fake_session.message_queue), 0)
  234. # This is the last 'succes' one, the value set here is what test_read_config expects
  235. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  236. self._handle_msg_helper({ "command": [ "set_config", [ { self.name: { "test": 125 } }] ] },
  237. my_ok_answer )
  238. self.assertEqual(len(self.fake_session.message_queue), 1)
  239. self.assertEqual({'command': [ 'config_update', {'test': 125}]},
  240. self.fake_session.get_message(self.name, None))
  241. self.assertEqual(len(self.fake_session.message_queue), 0)
  242. my_bad_answer = { 'result': [1, "bad_answer"] }
  243. self.fake_session.group_sendmsg(my_bad_answer, "ConfigManager")
  244. self._handle_msg_helper({ "command": [ "set_config", [ self.name, { "test": 125 }] ] },
  245. my_bad_answer )
  246. self.assertEqual(len(self.fake_session.message_queue), 1)
  247. self.assertEqual({'command': [ 'config_update', {'test': 125}]},
  248. self.fake_session.get_message(self.name, None))
  249. self.assertEqual(len(self.fake_session.message_queue), 0)
  250. self.fake_session.group_sendmsg(None, 'ConfigManager')
  251. self._handle_msg_helper({ "command": [ "set_config", [ ] ] },
  252. {'result': [1, 'Wrong number of arguments']} )
  253. self._handle_msg_helper({ "command": [ "set_config", [ self.name, { "test": 125 }] ] },
  254. { 'result': [1, 'No answer message from TestModule']} )
  255. #self.assertEqual(len(self.fake_session.message_queue), 1)
  256. #self.assertEqual({'config_update': {'test': 124}},
  257. # self.fake_session.get_message(self.name, None))
  258. #self.assertEqual({'version': 1, 'TestModule': {'test': 124}}, self.cm.config.data)
  259. #
  260. self._handle_msg_helper({ "command":
  261. ["module_spec", self.spec.get_full_spec()]
  262. },
  263. {'result': [0]})
  264. self._handle_msg_helper({ "command": [ "module_spec", { 'foo': 1 } ] },
  265. {'result': [1, 'Error in data definition: no module_name in module_spec']})
  266. self._handle_msg_helper({ "command": [ "get_module_spec" ] }, { 'result': [ 0, { self.spec.get_module_name(): self.spec.get_full_spec() } ]})
  267. self._handle_msg_helper({ "command": [ "get_commands_spec" ] }, { 'result': [ 0, { self.spec.get_module_name(): self.spec.get_commands_spec() } ]})
  268. # re-add this once we have new way to propagate spec changes (1 instead of the current 2 messages)
  269. #self.assertEqual(len(self.fake_session.message_queue), 2)
  270. # the name here is actually wrong (and hardcoded), but needed in the current version
  271. # TODO: fix that
  272. #self.assertEqual({'specification_update': [ self.name, self.spec ] },
  273. # self.fake_session.get_message("Cmdctl", None))
  274. #self.assertEqual({'commands_update': [ self.name, self.commands ] },
  275. # self.fake_session.get_message("Cmdctl", None))
  276. self._handle_msg_helper({ "command":
  277. ["shutdown"]
  278. },
  279. {'result': [0]})
  280. def test_set_config_all(self):
  281. my_ok_answer = { 'result': [ 0 ] }
  282. self.assertEqual({"version": 2}, self.cm.config.data)
  283. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  284. self.cm._handle_set_config_all({"test": { "value1": 123 }})
  285. self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
  286. "test": { "value1": 123 }
  287. }, self.cm.config.data)
  288. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  289. self.cm._handle_set_config_all({"test": { "value1": 124 }})
  290. self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
  291. "test": { "value1": 124 }
  292. }, self.cm.config.data)
  293. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  294. self.cm._handle_set_config_all({"test": { "value2": True }})
  295. self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
  296. "test": { "value1": 124,
  297. "value2": True
  298. }
  299. }, self.cm.config.data)
  300. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  301. self.cm._handle_set_config_all({"test": { "value3": [ 1, 2, 3 ] }})
  302. self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
  303. "test": { "value1": 124,
  304. "value2": True,
  305. "value3": [ 1, 2, 3 ]
  306. }
  307. }, self.cm.config.data)
  308. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  309. self.cm._handle_set_config_all({"test": { "value2": False }})
  310. self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
  311. "test": { "value1": 124,
  312. "value2": False,
  313. "value3": [ 1, 2, 3 ]
  314. }
  315. }, self.cm.config.data)
  316. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  317. self.cm._handle_set_config_all({"test": { "value1": None }})
  318. self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
  319. "test": { "value2": False,
  320. "value3": [ 1, 2, 3 ]
  321. }
  322. }, self.cm.config.data)
  323. self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
  324. self.cm._handle_set_config_all({"test": { "value3": [ 1 ] }})
  325. self.assertEqual({"version": config_data.BIND10_CONFIG_DATA_VERSION,
  326. "test": { "value2": False,
  327. "value3": [ 1 ]
  328. }
  329. }, self.cm.config.data)
  330. def test_run(self):
  331. self.fake_session.group_sendmsg({ "command": [ "get_commands_spec" ] }, "ConfigManager")
  332. self.fake_session.group_sendmsg({ "command": [ "shutdown" ] }, "ConfigManager")
  333. self.cm.run()
  334. pass
  335. if __name__ == '__main__':
  336. if not 'CONFIG_TESTDATA_PATH' in os.environ or not 'CONFIG_WR_TESTDATA_PATH' in os.environ:
  337. 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")
  338. exit(1)
  339. unittest.main()