loadzone_test.py 12.4 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# Copyright (C) 2012  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.

'''Tests for the loadzone module'''

import unittest
from loadzone import *
20
from isc.dns import *
JINMEI Tatuya's avatar
JINMEI Tatuya committed
21
from isc.datasrc import *
22
import isc.log
JINMEI Tatuya's avatar
JINMEI Tatuya committed
23
24
import os
import shutil
25

JINMEI Tatuya's avatar
JINMEI Tatuya committed
26
27
28
29
30
31
# Some common test parameters
TESTDATA_PATH = os.environ['TESTDATA_PATH'] + os.sep
READ_ZONE_DB_FILE = TESTDATA_PATH + "rwtest.sqlite3" # original, to be copied
LOCAL_TESTDATA_PATH = os.environ['LOCAL_TESTDATA_PATH'] + os.sep
READ_ZONE_DB_FILE = TESTDATA_PATH + "rwtest.sqlite3" # original, to be copied
NEW_ZONE_TXT_FILE = LOCAL_TESTDATA_PATH + "example.org.zone"
32
ALT_NEW_ZONE_TXT_FILE = TESTDATA_PATH + "example.com.zone"
JINMEI Tatuya's avatar
JINMEI Tatuya committed
33
34
TESTDATA_WRITE_PATH = os.environ['TESTDATA_WRITE_PATH'] + os.sep
WRITE_ZONE_DB_FILE = TESTDATA_WRITE_PATH + "rwtest.sqlite3.copied"
35
TEST_ZONE_NAME = Name('example.org')
JINMEI Tatuya's avatar
JINMEI Tatuya committed
36
37
38
39
40
41
42
DATASRC_CONFIG = '{"database_file": "' + WRITE_ZONE_DB_FILE + '"}'

# before/after SOAs: different in mname and serial
ORIG_SOA_TXT = 'example.org. 3600 IN SOA ns1.example.org. ' +\
    'admin.example.org. 1234 3600 1800 2419200 7200\n'
NEW_SOA_TXT = 'example.org. 3600 IN SOA ns.example.org. ' +\
    'admin.example.org. 1235 3600 1800 2419200 7200\n'
43
44
45
# This is the brandnew SOA for a newly created zone
ALT_NEW_SOA_TXT = 'example.com. 3600 IN SOA ns.example.com. ' +\
    'admin.example.com. 1234 3600 1800 2419200 7200\n'
46

47
48
class TestLoadZoneRunner(unittest.TestCase):
    def setUp(self):
JINMEI Tatuya's avatar
JINMEI Tatuya committed
49
50
        shutil.copyfile(READ_ZONE_DB_FILE, WRITE_ZONE_DB_FILE)

51
        # default command line arguments
JINMEI Tatuya's avatar
JINMEI Tatuya committed
52
        self.__args = ['-c', DATASRC_CONFIG, 'example.org', NEW_ZONE_TXT_FILE]
53
        self.__runner = LoadZoneRunner(self.__args)
54
55

    def tearDown(self):
56
57
58
59
        # Delete the used DB file; if some of the tests unexpectedly fail
        # unexpectedly in the middle of updating the DB, a lock could stay
        # there and would affect the other tests that would otherwise succeed.
        os.unlink(WRITE_ZONE_DB_FILE)
60

61
    def test_init(self):
62
        '''
63
        Checks initial class attributes
64
        '''
65
66
        self.assertIsNone(self.__runner._zone_class)
        self.assertIsNone(self.__runner._zone_name)
67
68
69
        self.assertIsNone(self.__runner._zone_file)
        self.assertIsNone(self.__runner._datasrc_config)
        self.assertIsNone(self.__runner._datasrc_type)
70
        self.assertEqual(10000, self.__runner._load_iteration_limit)
71
72
        self.assertEqual('INFO', self.__runner._log_severity)
        self.assertEqual(0, self.__runner._log_debuglevel)
73
74
75
76

    def test_parse_args(self):
        self.__runner._parse_args()
        self.assertEqual(TEST_ZONE_NAME, self.__runner._zone_name)
JINMEI Tatuya's avatar
JINMEI Tatuya committed
77
        self.assertEqual(NEW_ZONE_TXT_FILE, self.__runner._zone_file)
JINMEI Tatuya's avatar
JINMEI Tatuya committed
78
79
        self.assertEqual(DATASRC_CONFIG, self.__runner._datasrc_config)
        self.assertEqual('sqlite3', self.__runner._datasrc_type) # default
80
        self.assertEqual(10000, self.__runner._load_iteration_limit) # default
81
        self.assertEqual(RRClass.IN(), self.__runner._zone_class) # default
82
83
84
85
86
87
88
89
        self.assertEqual('INFO', self.__runner._log_severity) # default
        self.assertEqual(0, self.__runner._log_debuglevel)

    def test_set_loglevel(self):
        runner = LoadZoneRunner(['-d', '1'] + self.__args)
        runner._parse_args()
        self.assertEqual('DEBUG', runner._log_severity)
        self.assertEqual(1, runner._log_debuglevel)
90
91

    def test_parse_bad_args(self):
JINMEI Tatuya's avatar
JINMEI Tatuya committed
92
93
94
95
96
        # -c cannot be omitted (right now)
        self.assertRaises(BadArgument,
                          LoadZoneRunner(['example', 'example.zone']).
                          _parse_args)

97
98
        copt = ['-c', '0']      # template for the mandatory -c option

99
        # There must be exactly 2 non-option arguments: zone name and zone file
JINMEI Tatuya's avatar
JINMEI Tatuya committed
100
101
102
        self.assertRaises(BadArgument, LoadZoneRunner(copt)._parse_args)
        self.assertRaises(BadArgument, LoadZoneRunner(copt + ['example']).
                          _parse_args)
103
104
105
106
107
        self.assertRaises(BadArgument, LoadZoneRunner(self.__args + ['0']).
                          _parse_args)

        # Bad zone name
        self.assertRaises(BadArgument,
JINMEI Tatuya's avatar
JINMEI Tatuya committed
108
                          LoadZoneRunner(copt + ['bad..name', 'example.zone']).
109
                          _parse_args)
110

111
112
113
114
115
116
117
118
119
        # Bad class name
        self.assertRaises(BadArgument,
                          LoadZoneRunner(copt + ['-C', 'badclass']).
                          _parse_args)
        # Unsupported class
        self.assertRaises(BadArgument,
                          LoadZoneRunner(copt + ['-C', 'CH']).
                          _parse_args)

120
121
122
123
124
125
126
127
128
        # bad debug level
        args = copt + ['example.org', 'example.zone'] # otherwise valid args
        self.assertRaises(BadArgument,
                          LoadZoneRunner(['-d', '-10'] + args)._parse_args)

        # bad report interval
        self.assertRaises(BadArgument,
                          LoadZoneRunner(['-i', '-5'] + args)._parse_args)

129
130
131
132
133
134
    def __common_load_setup(self):
        self.__runner._zone_class = RRClass.IN()
        self.__runner._zone_name = TEST_ZONE_NAME
        self.__runner._zone_file = NEW_ZONE_TXT_FILE
        self.__runner._datasrc_type = 'sqlite3'
        self.__runner._datasrc_config = DATASRC_CONFIG
135
136
137
        self.__runner._load_iteration_limit = 1
        self.__reports = []
        self.__runner._report_progress = lambda x: self.__reports.append(x)
138
139
140
141
142

    def __check_zone_soa(self, soa_txt, zone_name=TEST_ZONE_NAME):
        """Check that the given SOA RR exists and matches the expected string

        If soa_txt is None, the zone is expected to be non-existent.
143
144
        Otherwise, if soa_txt is False, the zone should exist but SOA is
        expected to be missing.
145
146

        """
JINMEI Tatuya's avatar
JINMEI Tatuya committed
147
148

        client = DataSourceClient('sqlite3', DATASRC_CONFIG)
149
150
151
152
        result, finder = client.find_zone(zone_name)
        if soa_txt is None:
            self.assertEqual(client.NOTFOUND, result)
            return
JINMEI Tatuya's avatar
JINMEI Tatuya committed
153
        self.assertEqual(client.SUCCESS, result)
154
        result, rrset, _ = finder.find(zone_name, RRType.SOA())
155
156
157
158
159
        if soa_txt:
            self.assertEqual(finder.SUCCESS, result)
            self.assertEqual(soa_txt, rrset.to_text())
        else:
            self.assertEqual(finder.NXRRSET, result)
JINMEI Tatuya's avatar
JINMEI Tatuya committed
160
161
162

    def test_load_update(self):
        '''successful case to loading new contents to an existing zone.'''
163
164
165
        self.__common_load_setup()
        self.__check_zone_soa(ORIG_SOA_TXT)
        self.__runner._do_load()
166
167
168
        # In this test setup every loaded RR will be reported, and there will
        # be 3 RRs
        self.assertEqual([1, 2, 3], self.__reports)
169
170
        self.__check_zone_soa(NEW_SOA_TXT)

171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
    def test_load_update_skipped_report(self):
        '''successful loading, with reports for every 2 RRs'''
        self.__common_load_setup()
        self.__runner._load_iteration_limit = 2
        self.__runner._do_load()
        self.assertEqual([2], self.__reports)

    def test_load_update_no_report(self):
        '''successful loading, without progress reports'''
        self.__common_load_setup()
        self.__runner._load_iteration_limit = 0
        self.__runner._do_load()
        self.assertEqual([], self.__reports) # no report
        self.__check_zone_soa(NEW_SOA_TXT)   # but load is completed

186
187
188
189
190
191
    def test_create_and_load(self):
        '''successful case to loading contents to a new zone (created).'''
        self.__common_load_setup()
        self.__runner._zone_name = Name('example.com')
        self.__runner._zone_file = ALT_NEW_ZONE_TXT_FILE
        self.__check_zone_soa(None, zone_name=Name('example.com'))
JINMEI Tatuya's avatar
JINMEI Tatuya committed
192
        self.__runner._do_load()
193
        self.__check_zone_soa(ALT_NEW_SOA_TXT, zone_name=Name('example.com'))
JINMEI Tatuya's avatar
JINMEI Tatuya committed
194

195
196
197
198
199
    def test_load_fail_badconfig(self):
        '''Load attempt fails due to broken datasrc config.'''
        self.__common_load_setup()
        self.__runner._datasrc_config = "invalid config"
        self.__check_zone_soa(ORIG_SOA_TXT)
200
        self.assertRaises(LoadFailure, self.__runner._do_load)
201
202
203
204
205
206
207
208
        self.__check_zone_soa(ORIG_SOA_TXT) # no change to the zone

    def test_load_fail_badzone(self):
        '''Load attempt fails due to broken zone file.'''
        self.__common_load_setup()
        self.__runner._zone_file = \
            LOCAL_TESTDATA_PATH + '/broken-example.org.zone'
        self.__check_zone_soa(ORIG_SOA_TXT)
209
        self.assertRaises(LoadFailure, self.__runner._do_load)
210
211
212
213
214
215
216
217
        self.__check_zone_soa(ORIG_SOA_TXT)

    def test_load_fail_noloader(self):
        '''Load attempt fails because loading isn't supported'''
        self.__common_load_setup()
        self.__runner._datasrc_type = 'memory'
        self.__runner._datasrc_config = '{"type": "memory"}'
        self.__check_zone_soa(ORIG_SOA_TXT)
218
        self.assertRaises(LoadFailure, self.__runner._do_load)
219
220
        self.__check_zone_soa(ORIG_SOA_TXT)

221
222
223
224
225
226
227
228
229
230
    def test_load_fail_create_cancel(self):
        '''Load attempt fails and new creation of zone is canceled'''
        self.__common_load_setup()
        self.__runner._zone_name = Name('example.com')
        self.__runner._zone_file = 'no-such-file'
        self.__check_zone_soa(None, zone_name=Name('example.com'))
        self.assertRaises(LoadFailure, self.__runner._do_load)
        # _do_load() should have once created the zone but then canceled it.
        self.__check_zone_soa(None, zone_name=Name('example.com'))

231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
    def __common_post_load_setup(self, zone_file):
        '''Common setup procedure for post load tests.'''
        # replace the LoadZoneRunner's original _post_load_warning() for
        # inspection
        self.__warnings = []
        self.__runner._post_load_warning = \
            lambda msg: self.__warnings.append(msg)

        # perform load and invoke checks
        self.__common_load_setup()
        self.__runner._zone_file = zone_file
        self.__check_zone_soa(ORIG_SOA_TXT)
        self.__runner._do_load()
        self.__runner._post_load_checks()

    def test_load_fail_create_cancel(self):
        '''Load succeeds but warns about missing SOA, should cause warn'''
        self.__common_load_setup()
        self.__common_post_load_setup(LOCAL_TESTDATA_PATH +
                                      '/example-nosoa.org.zone')
        self.__check_zone_soa(False)
        self.assertEqual(1, len(self.__warnings))
        self.assertEqual('zone has no SOA', self.__warnings[0])

    def test_load_fail_create_cancel(self):
        '''Load succeeds but warns about missing NS, should cause warn'''
        self.__common_load_setup()
        self.__common_post_load_setup(LOCAL_TESTDATA_PATH +
                                      '/example-nons.org.zone')
        self.__check_zone_soa(NEW_SOA_TXT)
        self.assertEqual(1, len(self.__warnings))
        self.assertEqual('zone has no NS', self.__warnings[0])

JINMEI Tatuya's avatar
JINMEI Tatuya committed
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
    def test_run_success(self):
        '''Check for the top-level method.

        Detailed behavior is tested in other tests.  We only check the
        return value of run(), and the zone is successfully loaded.

        '''
        self.__check_zone_soa(ORIG_SOA_TXT)
        self.assertEqual(0, self.__runner.run())
        self.__check_zone_soa(NEW_SOA_TXT)

    def test_run_fail(self):
        '''Check for the top-level method, failure case.

        Similar to the success test, but loading will fail, and return
        value should be 1.

        '''
        runner = LoadZoneRunner(['-c', DATASRC_CONFIG, 'example.org',
                                 LOCAL_TESTDATA_PATH +
                                 '/broken-example.org.zone'])
        self.__check_zone_soa(ORIG_SOA_TXT)
        self.assertEqual(1, runner.run())
        self.__check_zone_soa(ORIG_SOA_TXT)

289
290
if __name__== "__main__":
    isc.log.resetUnitTestRootLogger()
291
292
293
    # Disable the internal logging setup so the test output won't be too
    # verbose by default.
    LoadZoneRunner._config_log = lambda x: None
294
    unittest.main()