zone_loader_test.py 7.96 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 92
    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)

    def check_load_incremental(self, loader):
Jelte Jansen's avatar
Jelte Jansen committed
93 94 95 96 97 98 99 100 101 102 103 104
        # 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
105 106 107 108 109 110 111 112 113 114 115 116 117 118
        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)
119 120

    def test_bad_file(self):
Jelte Jansen's avatar
Jelte Jansen committed
121
        self.check_zone_soa(ORIG_SOA_TXT)
122 123
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name,
                                        'no such file')
124
        self.assertRaises(isc.datasrc.MasterFileError, loader.load)
Jelte Jansen's avatar
Jelte Jansen committed
125
        self.check_zone_soa(ORIG_SOA_TXT)
126

127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172
    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)

173 174 175 176 177 178 179 180 181
    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()