zone_loader_test.py 9.99 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20
# 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.

import isc.datasrc
import isc.dns

import os
import unittest
Jelte Jansen's avatar
Jelte Jansen committed
21 22 23
import shutil

# Constants and common data used in tests
24

25 26
TESTDATA_PATH = os.environ['TESTDATA_PATH']
TESTDATA_WRITE_PATH = os.environ['TESTDATA_WRITE_PATH']
27

28
ZONE_FILE = TESTDATA_PATH + '/example.com'
29 30
STATIC_ZONE_FILE = '../../../../datasrc/static.zone'
SOURCE_DB_FILE = TESTDATA_PATH + '/example.com.source.sqlite3'
Jelte Jansen's avatar
Jelte Jansen committed
31
ORIG_DB_FILE = TESTDATA_PATH + '/example.com.sqlite3'
32
DB_FILE = TESTDATA_WRITE_PATH + '/zoneloadertest.sqlite3'
33 34
DB_CLIENT_CONFIG = '{ "database_file": "' + DB_FILE + '" }'
DB_SOURCE_CLIENT_CONFIG = '{ "database_file": "' + SOURCE_DB_FILE + '" }'
35

Jelte Jansen's avatar
Jelte Jansen committed
36 37 38 39 40 41
ORIG_SOA_TXT = 'example.com. 3600 IN SOA master.example.com. ' +\
               'admin.example.com. 1234 3600 1800 2419200 7200\n'
NEW_SOA_TXT = 'example.com. 1000 IN SOA a.dns.example.com. ' +\
              'mail.example.com. 1 1 1 1 1\n'


42 43 44
class ZoneLoaderTests(unittest.TestCase):
    def setUp(self):
        self.test_name = isc.dns.Name("example.com")
45 46
        self.test_file = ZONE_FILE
        self.client = isc.datasrc.DataSourceClient("sqlite3", DB_CLIENT_CONFIG)
Jelte Jansen's avatar
Jelte Jansen committed
47 48
        # Make a fresh copy of the database
        shutil.copy(ORIG_DB_FILE, DB_FILE)
49

50 51 52 53 54 55 56
    def tearDown(self):
        # We can only create 1 loader at a time (it locks the db), and it
        # may not be destroyed immediately if there is an exception in a
        # test. So the tests that do create one should put it in self, and
        # we make sure to invalidate it here.
        self.loader = None

57 58 59 60 61 62 63 64 65 66 67 68
    def test_bad_constructor(self):
        self.assertRaises(TypeError, isc.datasrc.ZoneLoader)
        self.assertRaises(TypeError, isc.datasrc.ZoneLoader, 1)
        self.assertRaises(TypeError, isc.datasrc.ZoneLoader,
                          None, self.test_name, self.test_file)
        self.assertRaises(TypeError, isc.datasrc.ZoneLoader,
                          self.client, None, self.test_file)
        self.assertRaises(TypeError, isc.datasrc.ZoneLoader,
                          self.client, self.test_name, None)
        self.assertRaises(TypeError, isc.datasrc.ZoneLoader,
                          self.client, self.test_name, self.test_file, 1)

Jelte Jansen's avatar
Jelte Jansen committed
69 70
    def check_zone_soa(self, soa_txt):
        """
71
        Check that the given SOA RR exists and matches the expected string
Jelte Jansen's avatar
Jelte Jansen committed
72 73 74 75 76 77 78
        """
        result, finder = self.client.find_zone(self.test_name)
        self.assertEqual(self.client.SUCCESS, result)
        result, rrset, _ = finder.find(self.test_name, isc.dns.RRType.SOA())
        self.assertEqual(finder.SUCCESS, result)
        self.assertEqual(soa_txt, rrset.to_text())

79
    def check_load(self):
Jelte Jansen's avatar
Jelte Jansen committed
80
        self.check_zone_soa(ORIG_SOA_TXT)
81
        self.loader.load()
Jelte Jansen's avatar
Jelte Jansen committed
82 83
        self.check_zone_soa(NEW_SOA_TXT)

84
        # And after that, it should throw
85
        self.assertRaises(isc.dns.InvalidOperation, self.loader.load)
Jelte Jansen's avatar
Jelte Jansen committed
86

87
    def test_load_from_file(self):
88 89 90
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             self.test_file)
        self.check_load()
Jelte Jansen's avatar
Jelte Jansen committed
91

92 93 94
    def test_load_from_client(self):
        source_client = isc.datasrc.DataSourceClient('sqlite3',
                                                     DB_SOURCE_CLIENT_CONFIG)
95 96 97
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             source_client)
        self.check_load()
98

Jelte Jansen's avatar
Jelte Jansen committed
99 100
    def test_load_from_file_checkrefs(self):
        # A test to see the refcount is increased properly
101 102 103 104 105
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             self.test_file)
        # Explicitely delete the objects here, so we trigger bad reference
        # counting (best effort, if there are leaked references for these
        # objects themselves, it still won't fail)
106
        self.client = None
Jelte Jansen's avatar
Jelte Jansen committed
107 108 109
        self.client = None
        self.test_name = None
        self.test_file = None
110
        self.loader.load()
Jelte Jansen's avatar
Jelte Jansen committed
111 112 113 114 115

    def test_load_from_client_checkrefs(self):
        # A test to see the refcount is increased properly
        source_client = isc.datasrc.DataSourceClient('sqlite3',
                                                     DB_SOURCE_CLIENT_CONFIG)
116 117 118 119 120
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             source_client)
        # Explicitely delete the objects here, so we trigger bad reference
        # counting (best effort, if there are leaked references for these
        # objects themselves, it still won't fail)
Jelte Jansen's avatar
Jelte Jansen committed
121 122 123
        self.client = None
        self.test_name = None
        source_client = None
124
        self.loader.load()
Jelte Jansen's avatar
Jelte Jansen committed
125

126
    def check_load_incremental(self):
Jelte Jansen's avatar
Jelte Jansen committed
127 128
        # New zone has 8 RRs
        # After 5, it should return False
129
        self.assertFalse(self.loader.load_incremental(5))
Jelte Jansen's avatar
Jelte Jansen committed
130 131 132 133
        # New zone should not have been loaded yet
        self.check_zone_soa(ORIG_SOA_TXT)

        # After 5 more, it should return True (only having read 3)
134
        self.assertTrue(self.loader.load_incremental(5))
Jelte Jansen's avatar
Jelte Jansen committed
135 136 137 138
        # New zone should now be loaded
        self.check_zone_soa(NEW_SOA_TXT)

        # And after that, it should throw
139 140
        self.assertRaises(isc.dns.InvalidOperation,
                          self.loader.load_incremental, 5)
141 142 143

    def test_load_from_file_incremental(self):
        # Create loader and load the zone
144 145 146
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             self.test_file)
        self.check_load_incremental()
147 148 149 150

    def test_load_from_client_incremental(self):
        source_client = isc.datasrc.DataSourceClient('sqlite3',
                                                     DB_SOURCE_CLIENT_CONFIG)
151 152 153
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             source_client)
        self.check_load_incremental()
154 155

    def test_bad_file(self):
Jelte Jansen's avatar
Jelte Jansen committed
156
        self.check_zone_soa(ORIG_SOA_TXT)
157 158 159
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             'no such file')
        self.assertRaises(isc.datasrc.MasterFileError, self.loader.load)
Jelte Jansen's avatar
Jelte Jansen committed
160
        self.check_zone_soa(ORIG_SOA_TXT)
161

162 163
    def test_bad_file_incremental(self):
        self.check_zone_soa(ORIG_SOA_TXT)
164 165
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             'no such file')
166
        self.assertRaises(isc.datasrc.MasterFileError,
167
                          self.loader.load_incremental, 1)
168 169 170 171 172 173 174 175
        self.check_zone_soa(ORIG_SOA_TXT)

    def test_no_such_zone_in_target(self):
        self.assertRaises(isc.datasrc.Error, isc.datasrc.ZoneLoader,
                          self.client, isc.dns.Name("unknownzone"),
                          self.test_file)

    def test_no_such_zone_in_source(self):
176 177
        # Reuse a zone that exists in target but not in source
        zone_name = isc.dns.Name("sql1.example.com")
178 179
        source_client = isc.datasrc.DataSourceClient('sqlite3',
                                                     DB_SOURCE_CLIENT_CONFIG)
180 181 182 183 184 185 186 187

        # make sure the zone exists in the target
        found, _ = self.client.find_zone(zone_name)
        self.assertEqual(self.client.SUCCESS, found)
        # And that it does not in the source
        found, _ = source_client.find_zone(zone_name)
        self.assertNotEqual(source_client.SUCCESS, found)

188
        self.assertRaises(isc.datasrc.Error, isc.datasrc.ZoneLoader,
189
                          self.client, zone_name, source_client)
190 191 192 193 194 195 196 197 198 199 200 201

    def test_no_ds_load_support(self):
        # This may change in the future, but atm, the in-mem ds does
        # not support the API the zone loader uses (it has direct load calls)
        inmem_client = isc.datasrc.DataSourceClient('memory',
                                                    '{ "type": "memory" }');
        self.assertRaises(isc.datasrc.NotImplemented,
                          isc.datasrc.ZoneLoader,
                          inmem_client, self.test_name, self.test_file)

    def test_wrong_class_from_file(self):
        # If the file has wrong class, it is not detected until load time
202 203 204
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             self.test_file + '.ch')
        self.assertRaises(isc.datasrc.MasterFileError, self.loader.load)
205 206 207 208 209 210 211 212 213 214 215 216

    def test_wrong_class_from_client(self):
        # For ds->ds loading, wrong class is detected upon construction
        # Need a bit of the extended setup for CH source client
        clientlist = isc.datasrc.ConfigurableClientList(isc.dns.RRClass.CH())
        clientlist.configure('[ { "type": "static", "params": "' +
                             STATIC_ZONE_FILE +'" } ]', False)
        source_client, _, _ = clientlist.find(isc.dns.Name("bind."),
                                              False, False)
        self.assertRaises(isc.dns.InvalidParameter, isc.datasrc.ZoneLoader,
                          self.client, isc.dns.Name("bind."), source_client)

217 218 219 220 221 222 223 224 225
    def test_exception(self):
        # Just check if masterfileerror is subclass of datasrc.Error
        self.assertTrue(issubclass(isc.datasrc.MasterFileError,
                                   isc.datasrc.Error))

if __name__ == "__main__":
    isc.log.init("bind10")
    isc.log.resetUnitTestRootLogger()
    unittest.main()