zone_loader_test.py 9.65 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 57 58 59 60 61

    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
62 63
    def check_zone_soa(self, soa_txt):
        """
64
        Check that the given SOA RR exists and matches the expected string
Jelte Jansen's avatar
Jelte Jansen committed
65 66 67 68 69 70 71
        """
        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())

72
    def check_load(self, loader):
Jelte Jansen's avatar
Jelte Jansen committed
73
        self.check_zone_soa(ORIG_SOA_TXT)
74
        loader.load()
Jelte Jansen's avatar
Jelte Jansen committed
75 76
        self.check_zone_soa(NEW_SOA_TXT)

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

80 81 82 83
    def test_load_from_file(self):
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                        self.test_file)
        self.check_load(loader)
Jelte Jansen's avatar
Jelte Jansen committed
84

85 86 87 88 89 90 91
    def test_load_from_client(self):
        source_client = isc.datasrc.DataSourceClient('sqlite3',
                                                     DB_SOURCE_CLIENT_CONFIG)
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                        source_client)
        self.check_load(loader)

Jelte Jansen's avatar
Jelte Jansen committed
92 93 94 95 96
    def test_load_from_file_checkrefs(self):
        # A test to see the refcount is increased properly
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                        self.test_file)
        # Explicitely delete the objects here, so we trigger wrong
97 98 99 100
        # DECREF calls, should there be any (best effort, if
        # there are leaked references for these objects themselves,
        # it still won't fail)
        self.client = None
Jelte Jansen's avatar
Jelte Jansen committed
101 102 103 104 105 106 107 108 109 110 111 112 113
        self.client = None
        self.test_name = None
        self.test_file = None
        loader.load()
        loader = None

    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)
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                        source_client)
        # Explicitely delete the objects here, so we trigger wrong
114 115 116
        # DECREF calls, should there be any (best effort, if
        # there are leaked references for these objects themselves,
        # it still won't fail)
Jelte Jansen's avatar
Jelte Jansen committed
117 118 119 120 121 122
        self.client = None
        self.test_name = None
        source_client = None
        loader.load()
        loader = None

123
    def check_load_incremental(self, loader):
Jelte Jansen's avatar
Jelte Jansen committed
124 125 126 127 128 129 130 131 132 133 134 135
        # New zone has 8 RRs
        # After 5, it should return False
        self.assertFalse(loader.load_incremental(5))
        # 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)
        self.assertTrue(loader.load_incremental(5))
        # New zone should now be loaded
        self.check_zone_soa(NEW_SOA_TXT)

        # And after that, it should throw
136 137 138 139 140 141 142 143 144 145 146 147 148 149
        self.assertRaises(isc.dns.InvalidOperation, loader.load_incremental, 5)

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

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

    def test_bad_file(self):
Jelte Jansen's avatar
Jelte Jansen committed
152
        self.check_zone_soa(ORIG_SOA_TXT)
153 154
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                        'no such file')
155
        self.assertRaises(isc.datasrc.MasterFileError, loader.load)
Jelte Jansen's avatar
Jelte Jansen committed
156
        self.check_zone_soa(ORIG_SOA_TXT)
157

158 159 160 161 162 163 164 165 166 167 168 169 170 171
    def test_bad_file_incremental(self):
        self.check_zone_soa(ORIG_SOA_TXT)
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                        'no such file')
        self.assertRaises(isc.datasrc.MasterFileError,
                          loader.load_incremental, 1)
        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):
172 173
        # Reuse a zone that exists in target but not in source
        zone_name = isc.dns.Name("sql1.example.com")
174 175
        source_client = isc.datasrc.DataSourceClient('sqlite3',
                                                     DB_SOURCE_CLIENT_CONFIG)
176 177 178 179 180 181 182 183

        # 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)

184
        self.assertRaises(isc.datasrc.Error, isc.datasrc.ZoneLoader,
185
                          self.client, zone_name, source_client)
186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212

    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
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                        self.test_file + '.ch')
        self.assertRaises(isc.datasrc.MasterFileError, loader.load)

    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)

213 214 215 216 217 218 219 220 221
    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()