zone_loader_test.py 11.2 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
import shutil
22
import sys
Jelte Jansen's avatar
Jelte Jansen committed
23 24

# Constants and common data used in tests
25

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

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

Jelte Jansen's avatar
Jelte Jansen committed
37 38 39 40 41 42
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'


43 44 45
class ZoneLoaderTests(unittest.TestCase):
    def setUp(self):
        self.test_name = isc.dns.Name("example.com")
46 47
        self.test_file = ZONE_FILE
        self.client = isc.datasrc.DataSourceClient("sqlite3", DB_CLIENT_CONFIG)
Jelte Jansen's avatar
Jelte Jansen committed
48 49
        # Make a fresh copy of the database
        shutil.copy(ORIG_DB_FILE, DB_FILE)
50 51 52 53 54 55
        # Some tests set source client; if so, check refcount in
        # tearDown, since most tests don't, set it to None by default.
        self.source_client = None
        self.loader = None
        self.assertEqual(2, sys.getrefcount(self.test_name))
        self.assertEqual(2, sys.getrefcount(self.client))
56

57 58 59 60 61
    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.
62 63 64 65 66 67 68 69 70 71

        # We can also use this to check reference counts; if a loader
        # exists, the client and source client (if any) should have
        # an increased reference count (but the name should not, this
        # is only used in the initializer)
        if self.loader is not None:
            self.assertEqual(2, sys.getrefcount(self.test_name))
            self.assertEqual(3, sys.getrefcount(self.client))
            if (self.source_client is not None):
                self.assertEqual(3, sys.getrefcount(self.source_client))
72 73
        self.loader = None

74 75 76 77 78 79 80
        # Now that the loader has been destroyed, the refcounts
        # of its arguments should be back to their originals
        self.assertEqual(2, sys.getrefcount(self.test_name))
        self.assertEqual(2, sys.getrefcount(self.client))
        if (self.source_client is not None):
            self.assertEqual(2, sys.getrefcount(self.source_client))

81 82 83 84 85 86 87 88 89 90 91 92
    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
93 94
    def check_zone_soa(self, soa_txt):
        """
95
        Check that the given SOA RR exists and matches the expected string
Jelte Jansen's avatar
Jelte Jansen committed
96 97 98 99 100 101 102
        """
        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())

103
    def check_load(self):
Jelte Jansen's avatar
Jelte Jansen committed
104
        self.check_zone_soa(ORIG_SOA_TXT)
105
        self.loader.load()
Jelte Jansen's avatar
Jelte Jansen committed
106 107
        self.check_zone_soa(NEW_SOA_TXT)

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

111
    def test_load_from_file(self):
112 113 114
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             self.test_file)
        self.check_load()
Jelte Jansen's avatar
Jelte Jansen committed
115

116
    def test_load_from_client(self):
117 118
        self.source_client = isc.datasrc.DataSourceClient('sqlite3',
                                    DB_SOURCE_CLIENT_CONFIG)
119
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
120
                                             self.source_client)
121
        self.check_load()
122

123
    def Xtest_load_from_file_checkrefs(self):
Jelte Jansen's avatar
Jelte Jansen committed
124
        # A test to see the refcount is increased properly
125 126 127 128 129
        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)
130
        self.client = None
Jelte Jansen's avatar
Jelte Jansen committed
131 132 133
        self.client = None
        self.test_name = None
        self.test_file = None
134
        self.loader.load()
Jelte Jansen's avatar
Jelte Jansen committed
135

136
    def Xtest_load_from_client_checkrefs(self):
Jelte Jansen's avatar
Jelte Jansen committed
137 138 139
        # A test to see the refcount is increased properly
        source_client = isc.datasrc.DataSourceClient('sqlite3',
                                                     DB_SOURCE_CLIENT_CONFIG)
140 141 142 143 144
        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
145 146 147
        self.client = None
        self.test_name = None
        source_client = None
148
        self.loader.load()
Jelte Jansen's avatar
Jelte Jansen committed
149

150
    def check_load_incremental(self):
Jelte Jansen's avatar
Jelte Jansen committed
151 152
        # New zone has 8 RRs
        # After 5, it should return False
153
        self.assertFalse(self.loader.load_incremental(5))
Jelte Jansen's avatar
Jelte Jansen committed
154 155 156 157
        # 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)
158
        self.assertTrue(self.loader.load_incremental(5))
Jelte Jansen's avatar
Jelte Jansen committed
159 160 161 162
        # New zone should now be loaded
        self.check_zone_soa(NEW_SOA_TXT)

        # And after that, it should throw
163 164
        self.assertRaises(isc.dns.InvalidOperation,
                          self.loader.load_incremental, 5)
165 166 167

    def test_load_from_file_incremental(self):
        # Create loader and load the zone
168 169 170
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             self.test_file)
        self.check_load_incremental()
171 172

    def test_load_from_client_incremental(self):
173 174
        self.source_client = isc.datasrc.DataSourceClient('sqlite3',
                                            DB_SOURCE_CLIENT_CONFIG)
175
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
176
                                             self.source_client)
177
        self.check_load_incremental()
178 179

    def test_bad_file(self):
Jelte Jansen's avatar
Jelte Jansen committed
180
        self.check_zone_soa(ORIG_SOA_TXT)
181 182 183
        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
184
        self.check_zone_soa(ORIG_SOA_TXT)
185

186 187
    def test_bad_file_incremental(self):
        self.check_zone_soa(ORIG_SOA_TXT)
188 189
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             'no such file')
190
        self.assertRaises(isc.datasrc.MasterFileError,
191
                          self.loader.load_incremental, 1)
192 193 194 195 196 197 198 199
        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):
200 201
        # Reuse a zone that exists in target but not in source
        zone_name = isc.dns.Name("sql1.example.com")
202 203
        self.source_client = isc.datasrc.DataSourceClient('sqlite3',
                                            DB_SOURCE_CLIENT_CONFIG)
204 205 206 207 208

        # 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
209 210
        found, _ = self.source_client.find_zone(zone_name)
        self.assertNotEqual(self.source_client.SUCCESS, found)
211

212
        self.assertRaises(isc.datasrc.Error, isc.datasrc.ZoneLoader,
213
                          self.client, zone_name, self.source_client)
214 215 216 217 218 219 220 221 222 223 224 225

    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
226 227 228
        self.loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                             self.test_file + '.ch')
        self.assertRaises(isc.datasrc.MasterFileError, self.loader.load)
229 230 231 232 233 234 235

    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)
236 237
        self.source_client, _, _ = clientlist.find(isc.dns.Name("bind."),
                                                   False, False)
238
        self.assertRaises(isc.dns.InvalidParameter, isc.datasrc.ZoneLoader,
239 240
                          self.client, isc.dns.Name("bind."),
                          self.source_client)
241

242 243 244 245 246 247 248 249 250
    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()