cfgmgr_test.py 15.5 KB
Newer Older
1
# Copyright (C) 2010  Internet Systems Consortium.
Jelte Jansen's avatar
Jelte Jansen committed
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
#
# 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.

#
# Tests for the configuration manager module
#

import unittest
import os
from isc.config.cfgmgr import *
Jelte Jansen's avatar
Jelte Jansen committed
23
from unittest_fakesession import FakeModuleCCSession
Jelte Jansen's avatar
Jelte Jansen committed
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

class TestConfigManagerData(unittest.TestCase):
    def setUp(self):
        self.data_path = os.environ['CONFIG_TESTDATA_PATH']
        self.config_manager_data = ConfigManagerData(self.data_path)
        self.assert_(self.config_manager_data)

    def test_init(self):
        self.assertEqual(self.config_manager_data.data['version'],
                         ConfigManagerData.CONFIG_VERSION)
        self.assertEqual(self.config_manager_data.data_path,
                         self.data_path)
        self.assertEqual(self.config_manager_data.db_filename,
                         self.data_path + os.sep + "b10-config.db")

    def test_read_from_file(self):
        ConfigManagerData.read_from_file(self.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")

    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)

60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
    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)
        
Jelte Jansen's avatar
Jelte Jansen committed
75
76

class TestConfigManager(unittest.TestCase):
Jelte Jansen's avatar
Jelte Jansen committed
77
78

    def setUp(self):
Jelte Jansen's avatar
Jelte Jansen committed
79
        self.data_path = os.environ['CONFIG_TESTDATA_PATH']
80
        self.fake_session = FakeModuleCCSession()
Jelte Jansen's avatar
Jelte Jansen committed
81
82
        self.cm = ConfigManager(self.data_path, self.fake_session)
        self.name = "TestModule"
Jelte Jansen's avatar
Jelte Jansen committed
83
        self.spec = isc.config.module_spec_from_file(self.data_path + os.sep + "/spec2.spec")
Jelte Jansen's avatar
Jelte Jansen committed
84
85
    
    def test_init(self):
Jelte Jansen's avatar
Jelte Jansen committed
86
        self.assert_(self.cm.module_specs == {})
Jelte Jansen's avatar
Jelte Jansen committed
87
88
89
90
91
92
93
94
95
96
97
98
        self.assert_(self.cm.data_path == self.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"})
Jelte Jansen's avatar
Jelte Jansen committed
99

100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
    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())
Jelte Jansen's avatar
Jelte Jansen committed
138
139
140
        config_spec = self.cm.get_config_spec('Spec2')
        self.assertEqual(config_spec['Spec2'], module_spec.get_config_spec())
        
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
    
    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())
Jelte Jansen's avatar
Jelte Jansen committed
158
159
        commands_spec = self.cm.get_commands_spec('Spec2')
        self.assertEqual(commands_spec['Spec2'], module_spec.get_commands_spec())
160
161
162
163

    def test_read_config(self):
        self.assertEqual(self.cm.config.data, {'version': 1})
        self.cm.read_config()
Jelte Jansen's avatar
Jelte Jansen committed
164
165
166
167
168
        # 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': 1})
        self.cm.data_path = "/no_such_path"
        self.cm.read_config()
        self.assertEqual(self.cm.config.data, {'version': 1})
169
170
171
172
173

    def test_write_config(self):
        # tested in ConfigManagerData tests
        pass
    
Jelte Jansen's avatar
Jelte Jansen committed
174
175
176
    def _handle_msg_helper(self, msg, expected_answer):
        answer = self.cm.handle_msg(msg)
        self.assertEqual(expected_answer, answer)
Jelte Jansen's avatar
Jelte Jansen committed
177
178

    def test_handle_msg(self):
Jelte Jansen's avatar
Jelte Jansen committed
179
180
        self._handle_msg_helper({}, { 'result': [ 1, 'Unknown message format: {}']})
        self._handle_msg_helper("", { 'result': [ 1, 'Unknown message format: ']})
181
        self._handle_msg_helper({ "command": [ "badcommand" ] }, { 'result': [ 1, "Unknown command: badcommand"]})
Jelte Jansen's avatar
Jelte Jansen committed
182
183
        self._handle_msg_helper({ "command": [ "get_commands_spec" ] }, { 'result': [ 0, {} ]})
        self._handle_msg_helper({ "command": [ "get_module_spec" ] }, { 'result': [ 0, {} ]})
Jelte Jansen's avatar
Jelte Jansen committed
184
        self._handle_msg_helper({ "command": [ "get_module_spec", { "module_name": "Spec2" } ] }, { 'result': [ 0, {} ]})
Jelte Jansen's avatar
Jelte Jansen committed
185
        #self._handle_msg_helper({ "command": [ "get_module_spec", { "module_name": "nosuchmodule" } ] },
186
        #                        {'result': [1, 'No specification for module nosuchmodule']})
Jelte Jansen's avatar
Jelte Jansen committed
187
188
189
190
        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']})
Jelte Jansen's avatar
Jelte Jansen committed
191
192
        self._handle_msg_helper({ "command": [ "get_config" ] }, { 'result': [ 0, { 'version': 1} ]})
        self._handle_msg_helper({ "command": [ "get_config", { "module_name": "nosuchmodule" } ] },
193
                                {'result': [0, { 'version': 1 }]})
Jelte Jansen's avatar
Jelte Jansen committed
194
195
196
197
198
199
        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']})
200
        self._handle_msg_helper({ "command": [ "set_config", [{}]] },
Jelte Jansen's avatar
Jelte Jansen committed
201
202
                                {'result': [0]})
        self.assertEqual(len(self.fake_session.message_queue), 0)
203
204
205
206
207

        # the targets of some of these tests expect specific answers, put
        # those in our fake msgq first.
        my_ok_answer = { 'result': [ 0 ] }

Jelte Jansen's avatar
Jelte Jansen committed
208
209

        # Send the 'ok' that cfgmgr expects back to the fake queue first
210
        self.fake_session.group_sendmsg(my_ok_answer, "ConfigManager")
Jelte Jansen's avatar
Jelte Jansen committed
211
        # then send the command
212
        self._handle_msg_helper({ "command": [ "set_config", [self.name, { "test": 123 }] ] },
213
                                my_ok_answer)
Jelte Jansen's avatar
Jelte Jansen committed
214
        # The cfgmgr should have eaten the ok message, and sent out an update again
Jelte Jansen's avatar
Jelte Jansen committed
215
        self.assertEqual(len(self.fake_session.message_queue), 1)
Jelte Jansen's avatar
Jelte Jansen committed
216
        self.assertEqual({'command': [ 'config_update', {'test': 123}]},
Jelte Jansen's avatar
Jelte Jansen committed
217
                         self.fake_session.get_message(self.name, None))
Jelte Jansen's avatar
Jelte Jansen committed
218
219
        # and the queue should now be empty again
        self.assertEqual(len(self.fake_session.message_queue), 0)
220

Jelte Jansen's avatar
Jelte Jansen committed
221
222
223
224
        # 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)
Jelte Jansen's avatar
Jelte Jansen committed
225
        self.assertEqual(len(self.fake_session.message_queue), 1)
Jelte Jansen's avatar
Jelte Jansen committed
226
        self.assertEqual({'command': [ 'config_update', {'test': 124}]},
Jelte Jansen's avatar
Jelte Jansen committed
227
                         self.fake_session.get_message(self.name, None))
Jelte Jansen's avatar
Jelte Jansen committed
228
229
230
231
232
233
234
235
        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)
Jelte Jansen's avatar
Jelte Jansen committed
236
        self.assertEqual({'command': [ 'config_update', {'test': 125}]},
Jelte Jansen's avatar
Jelte Jansen committed
237
238
239
240
241
242
243
244
                         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)
Jelte Jansen's avatar
Jelte Jansen committed
245
        self.assertEqual({'command': [ 'config_update', {'test': 125}]},
Jelte Jansen's avatar
Jelte Jansen committed
246
247
248
249
250
251
252
253
254
255
256
257
258
                         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)
        #
259
260
        self._handle_msg_helper({ "command": 
                                  ["module_spec", self.spec.get_full_spec()]
Jelte Jansen's avatar
Jelte Jansen committed
261
262
                                },
                                {'result': [0]})
263
        self._handle_msg_helper({ "command": [ "module_spec", { 'foo': 1 } ] },
Jelte Jansen's avatar
Jelte Jansen committed
264
265
266
                                {'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_config_spec() } ]})
        self._handle_msg_helper({ "command": [ "get_commands_spec" ] }, { 'result': [ 0, { self.spec.get_module_name(): self.spec.get_commands_spec() } ]})
267
268
        # 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)
Jelte Jansen's avatar
Jelte Jansen committed
269
270
        # the name here is actually wrong (and hardcoded), but needed in the current version
        # TODO: fix that
271
272
273
274
        #self.assertEqual({'specification_update': [ self.name, self.spec ] },
        #                 self.fake_session.get_message("Cmd-Ctrld", None))
        #self.assertEqual({'commands_update': [ self.name, self.commands ] },
        #                 self.fake_session.get_message("Cmd-Ctrld", None))
Jelte Jansen's avatar
Jelte Jansen committed
275
276
277
278
279

        self._handle_msg_helper({ "command": 
                                  ["shutdown"]
                                },
                                {'result': [0]})
Jelte Jansen's avatar
Jelte Jansen committed
280
281

    def test_run(self):
Jelte Jansen's avatar
Jelte Jansen committed
282
283
        self.fake_session.group_sendmsg({ "command": [ "get_commands_spec" ] }, "ConfigManager")
        self.cm.run()
Jelte Jansen's avatar
Jelte Jansen committed
284
285
286
287
288
289
290
291
292
        pass


if __name__ == '__main__':
    if not 'CONFIG_TESTDATA_PATH' in os.environ:
        print("You need to set the environment variable CONFIG_TESTDATA_PATH to point to the directory containing the test data files")
        exit(1)
    unittest.main()