zone_loader_test.py 8.98 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 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117
    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
        # DECREF calls
        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
        # DECREF calls
        self.client = None
        self.test_name = None
        source_client = None
        loader.load()
        loader = None

118
    def check_load_incremental(self, loader):
Jelte Jansen's avatar
Jelte Jansen committed
119 120 121 122 123 124 125 126 127 128 129 130
        # 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
131 132 133 134 135 136 137 138 139 140 141 142 143 144
        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)
145 146

    def test_bad_file(self):
Jelte Jansen's avatar
Jelte Jansen committed
147
        self.check_zone_soa(ORIG_SOA_TXT)
148 149
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                        'no such file')
150
        self.assertRaises(isc.datasrc.MasterFileError, loader.load)
Jelte Jansen's avatar
Jelte Jansen committed
151
        self.check_zone_soa(ORIG_SOA_TXT)
152

153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198
    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):
        source_client = isc.datasrc.DataSourceClient('sqlite3',
                                                     DB_SOURCE_CLIENT_CONFIG)
        self.assertRaises(isc.datasrc.Error, isc.datasrc.ZoneLoader,
                          self.client, isc.dns.Name("unknownzone"),
                          source_client)

    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)

199 200 201 202 203 204 205 206 207
    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()