zone_loader_test.py 4.4 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

Jelte Jansen's avatar
Jelte Jansen committed
30
ORIG_DB_FILE = TESTDATA_PATH + '/example.com.sqlite3'
31 32
DB_FILE = TESTDATA_WRITE_PATH + '/zoneloadertest.sqlite3'
DB_CLIENT_CONFIG = '{ "database_file": "' + DB_FILE + '"}'
33

Jelte Jansen's avatar
Jelte Jansen committed
34 35 36 37 38 39
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'


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

    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
60 61 62 63 64 65 66 67 68 69
    def check_zone_soa(self, soa_txt):
        """
        Check that the given RRset exists and matches the expected string
        """
        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())

70
    def test_load_file(self):
Jelte Jansen's avatar
Jelte Jansen committed
71
        self.check_zone_soa(ORIG_SOA_TXT)
72 73

        # Create loader and load the zone
74
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name, self.test_file)
75
        loader.load()
Jelte Jansen's avatar
Jelte Jansen committed
76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97

        self.check_zone_soa(NEW_SOA_TXT)

    def test_load_incremental(self):
        self.check_zone_soa(ORIG_SOA_TXT)

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

        # 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
        self.assertRaises(isc.datasrc.Error, loader.load_incremental, 5)
98 99

    def test_bad_file(self):
Jelte Jansen's avatar
Jelte Jansen committed
100
        self.check_zone_soa(ORIG_SOA_TXT)
101
        loader = isc.datasrc.ZoneLoader(self.client, self.test_name, "no such file")
102
        self.assertRaises(isc.datasrc.MasterFileError, loader.load)
Jelte Jansen's avatar
Jelte Jansen committed
103
        self.check_zone_soa(ORIG_SOA_TXT)
104 105 106 107 108 109 110 111 112 113

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