Coverage for src/bin/xfrin/tests/xfrin_test : 97%

# Copyright (C) 2009-2011  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 unittest 

import re 

import shutil 

import socket 

import sys 

import io 

from isc.testutils.tsigctx_mock import MockTSIGContext 

from isc.testutils.ccsession_mock import MockModuleCCSession 

from isc.testutils.rrset_utils import * 

from xfrin import * 

import xfrin 

from isc.xfrin.diff import Diff 

import isc.log 

# If we use any python library that is basically a wrapper for 

# a library we use as well (like sqlite3 in our datasources), 

# we must make sure we import ours first; If we have special 

# rpath or libtool rules to pick the correct version, python might 

# choose the wrong one first, if those rules aren't hit first. 

# This would result in missing symbols later. 

import sqlite3 

# 

# Commonly used (mostly constant) test parameters 

# 

TEST_ZONE_NAME_STR = "example.com." 

TEST_ZONE_NAME = Name(TEST_ZONE_NAME_STR) 

TEST_RRCLASS = RRClass.IN() 

TEST_RRCLASS_STR = 'IN' 

TEST_DB_FILE = 'db_file' 

TEST_MASTER_IPV4_ADDRESS = '127.0.0.1' 

TEST_MASTER_IPV4_ADDRINFO = (socket.AF_INET, socket.SOCK_STREAM, 

                             (TEST_MASTER_IPV4_ADDRESS, 53)) 

TEST_MASTER_IPV6_ADDRESS = '::1' 

TEST_MASTER_IPV6_ADDRINFO = (socket.AF_INET6, socket.SOCK_STREAM, 

                             (TEST_MASTER_IPV6_ADDRESS, 53)) 

TESTDATA_SRCDIR = os.getenv("TESTDATASRCDIR") 

TESTDATA_OBJDIR = os.getenv("TESTDATAOBJDIR") 

# XXX: This should be a non priviledge port that is unlikely to be used. 

# If some other process uses this port test will fail. 

TEST_MASTER_PORT = '53535' 

TSIG_KEY = TSIGKey("example.com:SFuWd/q99SzF8Yzd1QbB9g==") 

# SOA intended to be used for the new SOA as a result of transfer. 

soa_rdata = Rdata(RRType.SOA(), TEST_RRCLASS, 

                  'master.example.com. admin.example.com ' + 

                  '1234 3600 1800 2419200 7200') 

soa_rrset = RRset(TEST_ZONE_NAME, TEST_RRCLASS, RRType.SOA(), RRTTL(3600)) 

soa_rrset.add_rdata(soa_rdata) 

# SOA intended to be used for the current SOA at the secondary side. 

# Note that its serial is smaller than that of soa_rdata. 

begin_soa_rdata = Rdata(RRType.SOA(), TEST_RRCLASS, 

                        'master.example.com. admin.example.com ' + 

                        '1230 3600 1800 2419200 7200') 

begin_soa_rrset = RRset(TEST_ZONE_NAME, TEST_RRCLASS, RRType.SOA(), RRTTL(3600)) 

begin_soa_rrset.add_rdata(begin_soa_rdata) 

example_axfr_question = Question(TEST_ZONE_NAME, TEST_RRCLASS, RRType.AXFR()) 

example_soa_question = Question(TEST_ZONE_NAME, TEST_RRCLASS, RRType.SOA()) 

default_questions = [example_axfr_question] 

default_answers = [soa_rrset] 

def get_fake_time_time(): 

    '''Returns a temporary replacement function for time.time(), which 

       always returns 0.1 more than the previous call. This is to make 

       sure these tests do not fail on systems where the time.time() 

       function has a high minimal accuracy. 

       This fake time.time() is usually set in place of the real one 

       where we need testing of get_running_time(). It is done is 

       as low a scope as possible, so as not to mess up unit test 

       framework time related tests. It must be set before 

       XfrinTransferState (or any class that initializes that) is 

       initialized. 

       And every time it is set up, in must be reset later (again, so 

       as not to mess up the framework's concept of time). 

    ''' 

    fake_time = 0.0 

    def fake_time_time(): 

        nonlocal fake_time 

        fake_time += 0.1 

        return fake_time 

    return fake_time_time 

def check_diffs(assert_fn, expected, actual): 

    '''A helper function checking the differences made in the XFR session. 

    This is expected called from some subclass of unittest.TestCase and 

    assert_fn is generally expected to be 'self.assertEqual' of that class. 

    ''' 

    assert_fn(len(expected), len(actual)) 

    for (diffs_exp, diffs_actual) in zip(expected, actual): 

        assert_fn(len(diffs_exp), len(diffs_actual)) 

        for (diff_exp, diff_actual) in zip(diffs_exp, diffs_actual): 

            # operation should match 

            assert_fn(diff_exp[0], diff_actual[0]) 

            # The diff as RRset should be equal (for simplicity we assume 

            # all RRsets contain exactly one RDATA) 

            assert_fn(diff_exp[1].get_name(), diff_actual[1].get_name()) 

            assert_fn(diff_exp[1].get_type(), diff_actual[1].get_type()) 

            assert_fn(diff_exp[1].get_class(), diff_actual[1].get_class()) 

            assert_fn(diff_exp[1].get_rdata_count(), 

                      diff_actual[1].get_rdata_count()) 

            assert_fn(1, diff_exp[1].get_rdata_count()) 

            assert_fn(diff_exp[1].get_rdata()[0], 

                      diff_actual[1].get_rdata()[0]) 

class XfrinTestException(Exception): 

    pass 

class XfrinTestTimeoutException(Exception): 

    pass 

class MockCC(MockModuleCCSession): 

    def get_default_value(self, identifier): 

        # The returned values should be identical to the spec file 

        # XXX: these should be retrieved from the spec file 

        # (see MyCCSession of xfrout_test.py.in) 

        if identifier == "zones/master_port": 

            return TEST_MASTER_PORT 

        if identifier == "zones/class": 

            return TEST_RRCLASS_STR 

exit        if identifier == "zones/use_ixfr": 

            return False 

    def remove_remote_config(self, module_name): 

        pass 

class MockDataSourceClient(): 

    '''A simple mock data source client. 

    This class provides a minimal set of wrappers related the data source 

    API that would be used by Diff objects.  For our testing purposes they 

    only keep truck of the history of the changes. 

    ''' 

    def __init__(self): 

        self.force_fail = False # if True, raise an exception on commit 

        self.committed_diffs = [] 

        self.diffs = [] 

    def get_class(self): 

        '''Mock version of get_class(). 

        We simply return the commonly used constant RR class.  If and when 

        we use this mock for a different RR class we need to adjust it 

        accordingly. 

        ''' 

        return TEST_RRCLASS 

    def find_zone(self, zone_name): 

        '''Mock version of find_zone(). 

        It returns itself (subsequently acting as a mock ZoneFinder) for 

        some test zone names.  For some others it returns either NOTFOUND 

        or PARTIALMATCH. 

        ''' 

        if zone_name == TEST_ZONE_NAME or \ 

                zone_name == Name('no-soa.example') or \ 

                zone_name == Name('dup-soa.example'): 

            return (isc.datasrc.DataSourceClient.SUCCESS, self) 

        elif zone_name == Name('no-such-zone.example'): 

            return (DataSourceClient.NOTFOUND, None) 

184        elif zone_name == Name('partial-match-zone.example'): 

            return (DataSourceClient.PARTIALMATCH, self) 

        raise ValueError('Unexpected input to mock client: bug in test case?') 

    def find(self, name, rrtype, options=ZoneFinder.FIND_DEFAULT): 

        '''Mock ZoneFinder.find(). 

        It returns the predefined SOA RRset to queries for SOA of the common 

        test zone name.  It also emulates some unusual cases for special 

        zone names. 

        ''' 

        if name == TEST_ZONE_NAME and rrtype == RRType.SOA(): 

            return (ZoneFinder.SUCCESS, begin_soa_rrset, 0) 

        if name == Name('no-soa.example'): 

            return (ZoneFinder.NXDOMAIN, None, 0) 

203        if name == Name('dup-soa.example'): 

            dup_soa_rrset = RRset(name, TEST_RRCLASS, RRType.SOA(), RRTTL(0)) 

            dup_soa_rrset.add_rdata(begin_soa_rdata) 

            dup_soa_rrset.add_rdata(soa_rdata) 

            return (ZoneFinder.SUCCESS, dup_soa_rrset, 0) 

        raise ValueError('Unexpected input to mock finder: bug in test case?') 

    def get_updater(self, zone_name, replace, journaling=False): 

        self._journaling_enabled = journaling 

        return self 

    def add_rrset(self, rrset): 

        self.diffs.append(('add', rrset)) 

    def delete_rrset(self, rrset): 

        self.diffs.append(('delete', rrset)) 

    def commit(self): 

        if self.force_fail: 

            raise isc.datasrc.Error('Updater.commit() failed') 

        self.committed_diffs.append(self.diffs) 

        self.diffs = [] 

class MockXfrin(Xfrin): 

    # This is a class attribute of a callable object that specifies a non 

    # default behavior triggered in _cc_check_command().  Specific test methods 

    # are expected to explicitly set this attribute before creating a 

    # MockXfrin object (when it needs a non default behavior). 

    # See the TestMain class. 

    check_command_hook = None 

    def _cc_setup(self): 

        self._tsig_key = None 

        self._module_cc = MockCC() 

        pass 

    def _get_db_file(self): 

        pass 

    def _cc_check_command(self): 

        self._shutdown_event.set() 

        if MockXfrin.check_command_hook: 

            MockXfrin.check_command_hook() 

    def xfrin_start(self, zone_name, rrclass, db_file, master_addrinfo, 

                    tsig_key, request_type, check_soa=True): 

        # store some of the arguments for verification, then call this 

        # method in the superclass 

        self.xfrin_started_master_addr = master_addrinfo[2][0] 

        self.xfrin_started_master_port = master_addrinfo[2][1] 

        self.xfrin_started_request_type = request_type 

        return Xfrin.xfrin_start(self, zone_name, rrclass, None, 

                                 master_addrinfo, tsig_key, 

                                 request_type, check_soa) 

class MockXfrinConnection(XfrinConnection): 

    def __init__(self, sock_map, zone_name, rrclass, datasrc_client, 

                 shutdown_event, master_addr, tsig_key=None): 

        super().__init__(sock_map, zone_name, rrclass, MockDataSourceClient(), 

                         shutdown_event, master_addr, TEST_DB_FILE) 

        self.query_data = b'' 

        self.reply_data = b'' 

        self.force_time_out = False 

        self.force_close = False 

        self.qlen = None 

        self.qid = None 

        self.response_generator = None 

    def _asyncore_loop(self): 

        if self.force_close: 

            self.handle_close() 

        elif not self.force_time_out: 

            self.handle_read() 

    def connect_to_master(self): 

        return True 

    def recv(self, size): 

        data = self.reply_data[:size] 

        self.reply_data = self.reply_data[size:] 

        if len(data) == 0: 

            raise XfrinTestTimeoutException('Emulated timeout') 

        if len(data) < size: 

            raise XfrinTestException('cannot get reply data (' + str(size) + 

                                     ' bytes)') 

        return data 

    def send(self, data): 

        if self.qlen != None and len(self.query_data) >= self.qlen: 

            # This is a new query.  reset the internal state. 

            self.qlen = None 

            self.qid = None 

            self.query_data = b'' 

        self.query_data += data 

        # when the outgoing data is sufficiently large to contain the length 

        # and the QID fields (4 octets or more), extract these fields. 

        # The length will be reset the internal query data to support multiple 

        # queries in a single test. 

        # The QID will be used to construct a matching response. 

        if len(self.query_data) >= 4 and self.qid == None: 

            self.qlen = socket.htons(struct.unpack('H', 

                                                   self.query_data[0:2])[0]) 

            self.qid = socket.htons(struct.unpack('H', self.query_data[2:4])[0]) 

            # if the response generator method is specified, invoke it now. 

            if self.response_generator != None: 

                self.response_generator() 

        return len(data)