bind10_test.py.in 15.8 KB
Newer Older
1
from bind10 import ProcessInfo, BoB, dump_pid, unlink_pid_file
Shane Kerr's avatar
Shane Kerr committed
2

3
4
5
# XXX: environment tests are currently disabled, due to the preprocessor
#      setup that we have now complicating the environment

Shane Kerr's avatar
Shane Kerr committed
6
7
8
9
import unittest
import sys
import os
import signal
Evan Hunt's avatar
Evan Hunt committed
10
import socket
11
from isc.net.addr import IPAddr
Shane Kerr's avatar
Shane Kerr committed
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

class TestProcessInfo(unittest.TestCase):
    def setUp(self):
        # redirect stdout to a pipe so we can check that our
        # process spawning is doing the right thing with stdout
        self.old_stdout = os.dup(sys.stdout.fileno())
        self.pipes = os.pipe()
        os.dup2(self.pipes[1], sys.stdout.fileno())
        os.close(self.pipes[1])
        # note that we use dup2() to restore the original stdout
        # to the main program ASAP in each test... this prevents
        # hangs reading from the child process (as the pipe is only
        # open in the child), and also insures nice pretty output

    def tearDown(self):
        # clean up our stdout munging
        os.dup2(self.old_stdout, sys.stdout.fileno())
        os.close(self.pipes[0])

    def test_init(self):
        pi = ProcessInfo('Test Process', [ '/bin/echo', 'foo' ])
        os.dup2(self.old_stdout, sys.stdout.fileno())
        self.assertEqual(pi.name, 'Test Process')
        self.assertEqual(pi.args, [ '/bin/echo', 'foo' ])
36
37
#        self.assertEqual(pi.env, { 'PATH': os.environ['PATH'],
#                                   'PYTHON_EXEC': os.environ['PYTHON_EXEC'] })
Shane Kerr's avatar
Shane Kerr committed
38
39
40
41
42
        self.assertEqual(pi.dev_null_stdout, False)
        self.assertEqual(os.read(self.pipes[0], 100), b"foo\n")
        self.assertNotEqual(pi.process, None)
        self.assertTrue(type(pi.pid) is int)

43
44
45
46
47
48
#    def test_setting_env(self):
#        pi = ProcessInfo('Test Process', [ '/bin/true' ], env={'FOO': 'BAR'})
#        os.dup2(self.old_stdout, sys.stdout.fileno())
#        self.assertEqual(pi.env, { 'PATH': os.environ['PATH'],
#                                   'PYTHON_EXEC': os.environ['PYTHON_EXEC'],
#                                   'FOO': 'BAR' })
Shane Kerr's avatar
Shane Kerr committed
49
50

    def test_setting_null_stdout(self):
51
        pi = ProcessInfo('Test Process', [ '/bin/echo', 'foo' ],
Shane Kerr's avatar
Shane Kerr committed
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
                         dev_null_stdout=True)
        os.dup2(self.old_stdout, sys.stdout.fileno())
        self.assertEqual(pi.dev_null_stdout, True)
        self.assertEqual(os.read(self.pipes[0], 100), b"")

    def test_respawn(self):
        pi = ProcessInfo('Test Process', [ '/bin/echo', 'foo' ])
        # wait for old process to work...
        self.assertEqual(os.read(self.pipes[0], 100), b"foo\n")
        # respawn it
        old_pid = pi.pid
        pi.respawn()
        os.dup2(self.old_stdout, sys.stdout.fileno())
        # make sure the new one started properly
        self.assertEqual(pi.name, 'Test Process')
        self.assertEqual(pi.args, [ '/bin/echo', 'foo' ])
68
69
#        self.assertEqual(pi.env, { 'PATH': os.environ['PATH'],
#                                   'PYTHON_EXEC': os.environ['PYTHON_EXEC'] })
Shane Kerr's avatar
Shane Kerr committed
70
71
72
73
74
75
        self.assertEqual(pi.dev_null_stdout, False)
        self.assertEqual(os.read(self.pipes[0], 100), b"foo\n")
        self.assertNotEqual(pi.process, None)
        self.assertTrue(type(pi.pid) is int)
        self.assertNotEqual(pi.pid, old_pid)

Shane Kerr's avatar
Shane Kerr committed
76
77
78
79
class TestBoB(unittest.TestCase):
    def test_init(self):
        bob = BoB()
        self.assertEqual(bob.verbose, False)
Jelte Jansen's avatar
Jelte Jansen committed
80
        self.assertEqual(bob.msgq_socket_file, None)
81
82
        self.assertEqual(bob.cc_session, None)
        self.assertEqual(bob.ccs, None)
Shane Kerr's avatar
Shane Kerr committed
83
84
85
        self.assertEqual(bob.processes, {})
        self.assertEqual(bob.dead_processes, {})
        self.assertEqual(bob.runnable, False)
86
87
88
89
        self.assertEqual(bob.uid, None)
        self.assertEqual(bob.username, None)
        self.assertEqual(bob.nocache, False)
        self.assertEqual(bob.cfg_start_auth, True)
90
        self.assertEqual(bob.cfg_start_resolver, False)
Shane Kerr's avatar
Shane Kerr committed
91

Jelte Jansen's avatar
Jelte Jansen committed
92
93
    def test_init_alternate_socket(self):
        bob = BoB("alt_socket_file")
Shane Kerr's avatar
Shane Kerr committed
94
        self.assertEqual(bob.verbose, False)
Jelte Jansen's avatar
Jelte Jansen committed
95
        self.assertEqual(bob.msgq_socket_file, "alt_socket_file")
96
97
        self.assertEqual(bob.cc_session, None)
        self.assertEqual(bob.ccs, None)
Evan Hunt's avatar
Evan Hunt committed
98
99
100
        self.assertEqual(bob.processes, {})
        self.assertEqual(bob.dead_processes, {})
        self.assertEqual(bob.runnable, False)
101
102
103
104
        self.assertEqual(bob.uid, None)
        self.assertEqual(bob.username, None)
        self.assertEqual(bob.nocache, False)
        self.assertEqual(bob.cfg_start_auth, True)
105
        self.assertEqual(bob.cfg_start_resolver, False)
106

107
# Class for testing the BoB start/stop components routines.
108
109
110
111
112
#
# Although testing that external processes start is outside the scope
# of the unit test, by overriding the process start methods we can check
# that the right processes are started depending on the configuration
# options.
113
class StartStopCheckBob(BoB):
114
115
116
117
118
119
120
121
    def __init__(self):
        BoB.__init__(self)

# Set flags as to which of the overridden methods has been run.
        self.msgq = False
        self.cfgmgr = False
        self.ccsession = False
        self.auth = False
122
        self.resolver = False
123
124
125
126
127
        self.xfrout = False
        self.xfrin = False
        self.zonemgr = False
        self.stats = False
        self.cmdctl = False
128
        self.c_channel_env = {}
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

    def read_bind10_config(self):
        # Configuration options are set directly
        pass

    def start_msgq(self, c_channel_env):
        self.msgq = True

    def start_cfgmgr(self, c_channel_env):
        self.cfgmgr = True

    def start_ccsession(self, c_channel_env):
        self.ccsession = True

    def start_auth(self, c_channel_env):
        self.auth = True

146
147
    def start_resolver(self, c_channel_env):
        self.resolver = True
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163

    def start_xfrout(self, c_channel_env):
        self.xfrout = True

    def start_xfrin(self, c_channel_env):
        self.xfrin = True

    def start_zonemgr(self, c_channel_env):
        self.zonemgr = True

    def start_stats(self, c_channel_env):
        self.stats = True

    def start_cmdctl(self, c_channel_env):
        self.cmdctl = True

164
165
166
    # We don't really use all of these stop_ methods. But it might turn out
    # someone would add some stop_ method to BoB and we want that one overriden
    # in case he forgets to update the tests.
167
    def stop_msgq(self):
168
169
        self.msgq = False

170
    def stop_cfgmgr(self):
171
172
        self.cfgmgr = False

173
    def stop_ccsession(self):
174
175
        self.ccsession = False

176
    def stop_auth(self):
177
178
        self.auth = False

179
    def stop_resolver(self):
180
181
        self.resolver = False

182
    def stop_xfrout(self):
183
184
        self.xfrout = False

185
    def stop_xfrin(self):
186
187
        self.xfrin = False

188
    def stop_zonemgr(self):
189
190
        self.zonemgr = False

191
    def stop_stats(self):
192
193
        self.stats = False

194
    def stop_cmdctl(self):
195
196
197
198
199
200
201
202
        self.cmdctl = False

class TestStartStopProcessesBob(unittest.TestCase):
    """
    Check that the start_all_processes method starts the right combination
    of processes and that the right processes are started and stopped
    according to changes in configuration.
    """
203
204
    def check_started(self, bob, core, auth, resolver):
        """
205
206
        Check that the right sets of services are started. The ones that
        should be running are specified by the core, auth and resolver parameters
207
208
209
210
211
212
213
214
215
216
217
218
219
220
        (they are groups of processes, eg. auth means b10-auth, -xfrout, -xfrin
        and -zonemgr).
        """
        self.assertEqual(bob.msgq, core)
        self.assertEqual(bob.cfgmgr, core)
        self.assertEqual(bob.ccsession, core)
        self.assertEqual(bob.auth, auth)
        self.assertEqual(bob.resolver, resolver)
        self.assertEqual(bob.xfrout, auth)
        self.assertEqual(bob.xfrin, auth)
        self.assertEqual(bob.zonemgr, auth)
        self.assertEqual(bob.stats, core)
        self.assertEqual(bob.cmdctl, core)

221
    def check_preconditions(self, bob):
222
        self.check_started(bob, False, False, False)
223

224
225
226
227
228
    def check_started_none(self, bob):
        """
        Check that the situation is according to configuration where no servers
        should be started. Some processes still need to be running.
        """
229
        self.check_started(bob, True, False, False)
230
231
232
233
234
235

    def check_started_both(self, bob):
        """
        Check the situation is according to configuration where both servers
        (auth and resolver) are enabled.
        """
236
        self.check_started(bob, True, True, True)
237
238
239
240
241

    def check_started_auth(self, bob):
        """
        Check the set of processes needed to run auth only is started.
        """
242
        self.check_started(bob, True, True, False)
243
244
245
246
247

    def check_started_resolver(self, bob):
        """
        Check the set of processes needed to run resolver only is started.
        """
248
        self.check_started(bob, True, False, True)
249

250
    # Checks the processes started when starting neither auth nor resolver
251
252
    # is specified.
    def test_start_none(self):
253
        # Create BoB and ensure correct initialization
254
        bob = StartStopCheckBob()
255
256
257
258
        self.check_preconditions(bob)

        # Start processes and check what was started
        bob.cfg_start_auth = False
259
        bob.cfg_start_resolver = False
260

261
        bob.start_all_processes()
262
        self.check_started_none(bob)
263
264
265

    # Checks the processes started when starting only the auth process
    def test_start_auth(self):
266
        # Create BoB and ensure correct initialization
267
        bob = StartStopCheckBob()
268
269
270
271
        self.check_preconditions(bob)

        # Start processes and check what was started
        bob.cfg_start_auth = True
272
        bob.cfg_start_resolver = False
273

274
        bob.start_all_processes()
275

276
        self.check_started_auth(bob)
277

278
279
    # Checks the processes started when starting only the resolver process
    def test_start_resolver(self):
280
        # Create BoB and ensure correct initialization
281
        bob = StartStopCheckBob()
282
283
284
285
        self.check_preconditions(bob)

        # Start processes and check what was started
        bob.cfg_start_auth = False
286
        bob.cfg_start_resolver = True
287

288
        bob.start_all_processes()
289

290
        self.check_started_resolver(bob)
291

292
    # Checks the processes started when starting both auth and resolver process
293
    def test_start_both(self):
294
        # Create BoB and ensure correct initialization
295
        bob = StartStopCheckBob()
296
297
298
299
        self.check_preconditions(bob)

        # Start processes and check what was started
        bob.cfg_start_auth = True
300
        bob.cfg_start_resolver = True
301

302
        bob.start_all_processes()
303

304
305
306
307
308
309
310
311
        self.check_started_both(bob)

    def test_config_start(self):
        """
        Test that the configuration starts and stops processes according
        to configuration changes.
        """

312
        # Create BoB and ensure correct initialization
313
        bob = StartStopCheckBob()
314
315
316
317
318
319
320
        self.check_preconditions(bob)

        # Start processes (nothing much should be started, as in
        # test_start_none)
        bob.cfg_start_auth = False
        bob.cfg_start_resolver = False

321
        bob.start_all_processes()
322
        bob.runnable = True
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
        self.check_started_none(bob)

        # Enable both at once
        bob.config_handler({'start_auth': True, 'start_resolver': True})
        self.check_started_both(bob)

        # Not touched by empty change
        bob.config_handler({})
        self.check_started_both(bob)

        # Not touched by change to the same configuration
        bob.config_handler({'start_auth': True, 'start_resolver': True})
        self.check_started_both(bob)

        # Turn them both off again
        bob.config_handler({'start_auth': False, 'start_resolver': False})
        self.check_started_none(bob)

        # Not touched by empty change
        bob.config_handler({})
        self.check_started_none(bob)

        # Not touched by change to the same configuration
        bob.config_handler({'start_auth': False, 'start_resolver': False})
        self.check_started_none(bob)

        # Start and stop auth separately
        bob.config_handler({'start_auth': True})
        self.check_started_auth(bob)

        bob.config_handler({'start_auth': False})
        self.check_started_none(bob)

        # Start and stop resolver separately
        bob.config_handler({'start_resolver': True})
        self.check_started_resolver(bob)

        bob.config_handler({'start_resolver': False})
        self.check_started_none(bob)

        # Alternate
        bob.config_handler({'start_auth': True})
        self.check_started_auth(bob)

        bob.config_handler({'start_auth': False, 'start_resolver': True})
        self.check_started_resolver(bob)
369

370
371
        bob.config_handler({'start_auth': True, 'start_resolver': False})
        self.check_started_auth(bob)
Shane Kerr's avatar
Shane Kerr committed
372

373
374
375
376
    def test_config_start_once(self):
        """
        Tests that a process is started only once.
        """
377
        # Create BoB and ensure correct initialization
378
        bob = StartStopCheckBob()
379
380
381
382
383
384
385
        self.check_preconditions(bob)

        # Start processes (both)
        bob.cfg_start_auth = True
        bob.cfg_start_resolver = True

        bob.start_all_processes()
386
        bob.runnable = True
387
388
389
390
391
392
393
394
395
396
397
398
        self.check_started_both(bob)

        bob.start_auth = lambda: self.fail("Started auth again")
        bob.start_xfrout = lambda: self.fail("Started xfrout again")
        bob.start_xfrin = lambda: self.fail("Started xfrin again")
        bob.start_zonemgr = lambda: self.fail("Started zonemgr again")
        bob.start_resolver = lambda: self.fail("Started resolver again")

        # Send again we want to start them. Should not do it, as they are.
        bob.config_handler({'start_auth': True})
        bob.config_handler({'start_resolver': True})

399
400
401
402
403
    def test_config_not_started_early(self):
        """
        Test that processes are not started by the config handler before
        startup.
        """
404
        bob = StartStopCheckBob()
405
406
407
408
409
410
411
412
413
414
        self.check_preconditions(bob)

        bob.start_auth = lambda: self.fail("Started auth again")
        bob.start_xfrout = lambda: self.fail("Started xfrout again")
        bob.start_xfrin = lambda: self.fail("Started xfrin again")
        bob.start_zonemgr = lambda: self.fail("Started zonemgr again")
        bob.start_resolver = lambda: self.fail("Started resolver again")

        bob.config_handler({'start_auth': True, 'start_resolver': True})

415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
class TestPIDFile(unittest.TestCase):
    def setUp(self):
        self.pid_file = '@builddir@' + os.sep + 'bind10.pid'
        if os.path.exists(self.pid_file):
            os.unlink(self.pid_file)

    def tearDown(self):
        if os.path.exists(self.pid_file):
            os.unlink(self.pid_file)

    def check_pid_file(self):
        # dump PID to the file, and confirm the content is correct
        dump_pid(self.pid_file)
        my_pid = os.getpid()
        self.assertEqual(my_pid, int(open(self.pid_file, "r").read()))

    def test_dump_pid(self):
        self.check_pid_file()

        # make sure any existing content will be removed
        open(self.pid_file, "w").write('dummy data\n')
        self.check_pid_file()

    def test_dump_pid_notexist(self):
        dummy_data = 'dummy_data\n'
        open(self.pid_file, "w").write(dummy_data)
        dump_pid('no_such_pid_file')
        # the original content of the file should be intact.
        self.assertEqual(dummy_data, open(self.pid_file, "r").read())

    def test_dump_pid_with_none(self):
        # Check the behavior of dump_pid() and unlink_pid_file() with None.
        # This should be no-op.
        dump_pid(None)
        self.assertFalse(os.path.exists(self.pid_file))

        dummy_data = 'dummy_data\n'
        open(self.pid_file, "w").write(dummy_data)
        unlink_pid_file(None)
        self.assertEqual(dummy_data, open(self.pid_file, "r").read())

    def test_dump_pid_failure(self):
        # the attempt to open file will fail, which should result in exception.
        self.assertRaises(IOError, dump_pid,
                          'nonexistent_dir' + os.sep + 'bind10.pid')

Shane Kerr's avatar
Shane Kerr committed
461
462
if __name__ == '__main__':
    unittest.main()