kea/src/bin/stats/tests/b10-stats_test.py

# Copyright (C) 2010, 2011, 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.

"""
In each of these tests we start several virtual components. They are
not the real components, no external processes are started. They are
just simple mock objects running each in its own thread and pretending
to be bind10 modules. This helps testing the stats module in a close
to real environment.
"""

import unittest
import os
import threading
import io
import time
import imp
import sys

import stats
import isc.log
import isc.cc.session
from test_utils import BaseModules, ThreadingServerManager, MyStats, \
    SimpleStats, SignalHandler, MyModuleCCSession, send_command
from isc.testutils.ccsession_mock import MockModuleCCSession

class TestUtilties(unittest.TestCase):
    items = [
        { 'item_name': 'test_int1',  'item_type': 'integer', 'item_default': 12345      },
        { 'item_name': 'test_real1', 'item_type': 'real',    'item_default': 12345.6789 },
        { 'item_name': 'test_bool1', 'item_type': 'boolean', 'item_default': True       },
        { 'item_name': 'test_str1',  'item_type': 'string',  'item_default': 'ABCD'     },
        { 'item_name': 'test_list1', 'item_type': 'list',    'item_default': [1,2,3],
          'list_item_spec' : { 'item_name': 'number',   'item_type': 'integer' } },
        { 'item_name': 'test_map1',  'item_type': 'map',     'item_default': {'a':1,'b':2,'c':3},
          'map_item_spec'  : [ { 'item_name': 'a',   'item_type': 'integer'},
                               { 'item_name': 'b',   'item_type': 'integer'},
                               { 'item_name': 'c', 'item_type': 'integer'} ] },
        { 'item_name': 'test_int2',  'item_type': 'integer' },
        { 'item_name': 'test_real2', 'item_type': 'real'    },
        { 'item_name': 'test_bool2', 'item_type': 'boolean' },
        { 'item_name': 'test_str2',  'item_type': 'string'  },
        { 'item_name': 'test_list2', 'item_type': 'list',
          'list_item_spec' : { 'item_name': 'number',   'item_type': 'integer' } },
        { 'item_name': 'test_map2',  'item_type': 'map',
          'map_item_spec'  : [ { 'item_name': 'A', 'item_type': 'integer'},
                               { 'item_name': 'B', 'item_type': 'integer'},
                               { 'item_name': 'C', 'item_type': 'integer'} ] },
        { 'item_name': 'test_none',  'item_type': 'none'    },
        { 'item_name': 'test_list3', 'item_type': 'list',    'item_default': ["one","two","three"],
          'list_item_spec' : { 'item_name': 'number', 'item_type': 'string' } },
        { 'item_name': 'test_map3',  'item_type': 'map',     'item_default': {'a':'one','b':'two','c':'three'},
          'map_item_spec'  : [ { 'item_name': 'a', 'item_type': 'string'},
                               { 'item_name': 'b', 'item_type': 'string'},
                               { 'item_name': 'c', 'item_type': 'string'} ] },
        {
          'item_name': 'test_named_set',
          'item_type': 'named_set',
          'item_default': { },
          'named_set_item_spec': {
            'item_name': 'name',
            'item_type': 'map',
            'item_default': { },
            'map_item_spec': [
              {
                'item_name': 'number1',
                'item_type': 'integer'
                },
              {
                'item_name': 'number2',
                'item_type': 'integer'
                }
              ]
            }
          }
        ]

    def setUp(self):
        self.const_timestamp = 1308730448.965706
        self.const_timetuple = (2011, 6, 22, 8, 14, 8, 2, 173, 0)
        self.const_datetime = '2011-06-22T08:14:08Z'
        stats.time = lambda : self.const_timestamp
        stats.gmtime = lambda : self.const_timetuple

    def test_get_spec_defaults(self):
        self.assertEqual(
            stats.get_spec_defaults(self.items), {
                'test_int1'  : 12345              ,
                'test_real1' : 12345.6789         ,
                'test_bool1' : True               ,
                'test_str1'  : 'ABCD'             ,
                'test_list1' : [1,2,3]            ,
                'test_map1'  : {'a':1,'b':2,'c':3},
                'test_int2'  : 0 ,
                'test_real2' : 0.0,
                'test_bool2' : False,
                'test_str2'  : "",
                'test_list2' : [0],
                'test_map2'  : { 'A' : 0, 'B' : 0, 'C' : 0 },
                'test_none'  : None,
                'test_list3' : [ "one", "two", "three" ],
                'test_map3'  : { 'a' : 'one', 'b' : 'two', 'c' : 'three' },
                'test_named_set' : {} })
        self.assertEqual(stats.get_spec_defaults(None), {})
        self.assertRaises(KeyError, stats.get_spec_defaults, [{'item_name':'Foo'}])

    def test_get_timestamp(self):
        self.assertEqual(stats.get_timestamp(), self.const_timestamp)

    def test_get_datetime(self):
        self.assertEqual(stats.get_datetime(), self.const_datetime)
        self.assertNotEqual(stats.get_datetime(
                (2011, 6, 22, 8, 23, 40, 2, 173, 0)), self.const_datetime)

    def test__accum(self):
        self.assertEqual(stats._accum(None, None), None)
        self.assertEqual(stats._accum(None, "b"), "b")
        self.assertEqual(stats._accum("a", None), "a")
        self.assertEqual(stats._accum(1, 2), 3)
        self.assertEqual(stats._accum(0.5, 0.3), 0.8)
        self.assertEqual(stats._accum('aa','bb'), 'bb')
        self.assertEqual(stats._accum('1970-01-01T09:00:00Z','2012-08-09T09:33:31Z'),
                         '2012-08-09T09:33:31Z')
        self.assertEqual(stats._accum(
                [1, 2, 3], [4, 5]), [5, 7, 3])
        self.assertEqual(stats._accum(
                [4, 5], [1, 2, 3]), [5, 7, 3])
        self.assertEqual(stats._accum(
                [1, 2, 3], [None, 5, 6]), [1, 7, 9])
        self.assertEqual(stats._accum(
                [None, 5, 6], [1, 2, 3]), [1, 7, 9])
        self.assertEqual(stats._accum(
                [1, 2, 3], [None, None, None, None]), [1,2,3,None])
        self.assertEqual(stats._accum(
                [[1,2],3],[[],5,6]), [[1,2],8,6])
        self.assertEqual(stats._accum(
                {'one': 1, 'two': 2, 'three': 3},
                {'one': 4, 'two': 5}),
                         {'one': 5, 'two': 7, 'three': 3})
        self.assertEqual(stats._accum(
                {'one': 1, 'two': 2, 'three': 3},
                {'four': 4, 'five': 5}),
                         {'one': 1, 'two': 2, 'three': 3, 'four': 4, 'five': 5})
        self.assertEqual(stats._accum(
                {'one': [1, 2], 'two': [3, None, 5], 'three': [None, 3, None]},
                {'one': [2], 'two': [4, 5], 'three': [None, None, None], 'four': 'FOUR'}),
                         {'one':[3,2], 'two':[7,5,5], 'three':[None,3,None], 'four': 'FOUR'})
        self.assertEqual(stats._accum(
                [ {'one': 1, 'two': 2, 'three': 3}, {'four': 4, 'five': 5, 'six': 6} ],
                [ {}, {'four': 1, 'five': 2, 'six': 3} ]),
                [ {'one': 1, 'two': 2, 'three': 3}, {'four': 5, 'five': 7, 'six': 9} ])

    def test_merge_oldnre(self):
        self.assertEqual(stats.merge_oldnew(1, 2), 2)
        self.assertEqual(stats.merge_oldnew(0.5, 0.3), 0.3)
        self.assertEqual(stats.merge_oldnew('aa','bb'), 'bb')
        self.assertEqual(stats.merge_oldnew(
                [1, 2, 3], [4, 5]), [4, 5, 3])
        self.assertEqual(stats.merge_oldnew(
                [4, 5], [1, 2, 3]), [1, 2, 3])
        self.assertEqual(stats.merge_oldnew(
                [1, 2, 3], [None, 5, 6]), [None, 5, 6])
        self.assertEqual(stats.merge_oldnew(
                [None, 5, 6], [1, 2, 3]), [1, 2, 3])
        self.assertEqual(stats.merge_oldnew(
                [1, 2, 3], [None, None, None, None]), [None, None, None, None])
        self.assertEqual(stats.merge_oldnew(
                [[1,2],3],[[],5,6]), [[1,2],5,6])
        self.assertEqual(stats.merge_oldnew(
                {'one': 1, 'two': 2, 'three': 3},
                {'one': 4, 'two': 5}),
                         {'one': 4, 'two': 5, 'three': 3})
        self.assertEqual(stats.merge_oldnew(
                {'one': 1, 'two': 2, 'three': 3},
                {'four': 4, 'five': 5}),
                         {'one': 1, 'two': 2, 'three': 3, 'four': 4, 'five': 5})
        self.assertEqual(stats.merge_oldnew(
                {'one': [1, 2], 'two': [3, None, 5], 'three': [None, 3, None]},
                {'one': [2], 'two': [4, 5], 'three': [None, None, None], 'four': 'FOUR'}),
                         {'one':[2,2], 'two':[4,5,5], 'three':[None,None,None], 'four': 'FOUR'})
        self.assertEqual(stats.merge_oldnew(
                [ {'one': 1, 'two': 2, 'three': 3}, {'four': 4, 'five': 5, 'six': 6} ],
                [ {}, {'four': 1, 'five': 2, 'six': 3} ]),
                [ {'one': 1, 'two': 2, 'three': 3}, {'four': 1, 'five': 2, 'six': 3} ])

class TestCallback(unittest.TestCase):
    def setUp(self):
        self.dummy_func = lambda *x, **y : (x, y)
        self.dummy_args = (1,2,3)
        self.dummy_kwargs = {'a':1,'b':2,'c':3}
        self.cback1 = stats.Callback(
            command=self.dummy_func,
            args=self.dummy_args,
            kwargs=self.dummy_kwargs
            )
        self.cback2 = stats.Callback(
            args=self.dummy_args,
            kwargs=self.dummy_kwargs
            )
        self.cback3 = stats.Callback(
            command=self.dummy_func,
            kwargs=self.dummy_kwargs
            )
        self.cback4 = stats.Callback(
            command=self.dummy_func,
            args=self.dummy_args
            )

    def test_init(self):
        self.assertEqual((self.cback1.command, self.cback1.args, self.cback1.kwargs),
                         (self.dummy_func, self.dummy_args, self.dummy_kwargs))
        self.assertEqual((self.cback2.command, self.cback2.args, self.cback2.kwargs),
                         (None, self.dummy_args, self.dummy_kwargs))
        self.assertEqual((self.cback3.command, self.cback3.args, self.cback3.kwargs),
                         (self.dummy_func, (), self.dummy_kwargs))
        self.assertEqual((self.cback4.command, self.cback4.args, self.cback4.kwargs),
                         (self.dummy_func, self.dummy_args, {}))

    def test_call(self):
        self.assertEqual(self.cback1(), (self.dummy_args, self.dummy_kwargs))
        self.assertEqual(self.cback1(100, 200), ((100, 200), self.dummy_kwargs))
        self.assertEqual(self.cback1(a=100, b=200), (self.dummy_args, {'a':100, 'b':200}))
        self.assertEqual(self.cback2(), None)
        self.assertEqual(self.cback3(), ((), self.dummy_kwargs))
        self.assertEqual(self.cback3(100, 200), ((100, 200), self.dummy_kwargs))
        self.assertEqual(self.cback3(a=100, b=200), ((), {'a':100, 'b':200}))
        self.assertEqual(self.cback4(), (self.dummy_args, {}))
        self.assertEqual(self.cback4(100, 200), ((100, 200), {}))
        self.assertEqual(self.cback4(a=100, b=200), (self.dummy_args, {'a':100, 'b':200}))

class TestStats(unittest.TestCase):
    def setUp(self):
        # set the signal handler for deadlock
        self.sig_handler = SignalHandler(self.fail)
        self.base = BaseModules()
        self.const_timestamp = 1308730448.965706
        self.const_datetime = '2011-06-22T08:14:08Z'
        self.const_default_datetime = '1970-01-01T00:00:00Z'
        # Record original module-defined functions in case we replace them
        self.__orig_timestamp = stats.get_timestamp
        self.__orig_get_datetime = stats.get_datetime

    def tearDown(self):
        self.base.shutdown()
        # reset the signal handler
        self.sig_handler.reset()
        # restore the stored original function in case we replaced them
        stats.get_timestamp = self.__orig_timestamp
        stats.get_datetime = self.__orig_get_datetime

    def test_init(self):
        self.stats = stats.Stats()
        self.assertEqual(self.stats.module_name, 'Stats')
        self.assertFalse(self.stats.running)
        self.assertTrue('command_show' in self.stats.callbacks)
        self.assertTrue('command_status' in self.stats.callbacks)
        self.assertTrue('command_shutdown' in self.stats.callbacks)
        self.assertTrue('command_show' in self.stats.callbacks)
        self.assertTrue('command_showschema' in self.stats.callbacks)
        self.assertEqual(self.stats.config['poll-interval'], 60)

    def test_init_undefcmd(self):
        spec_str = """\
{
  "module_spec": {
    "module_name": "Stats",
    "module_description": "Stats daemon",
    "config_data": [],
    "commands": [
      {
        "command_name": "_undef_command_",
        "command_description": "a undefined command in stats",
        "command_args": []
      }
    ],
    "statistics": []
  }
}
"""
        orig_spec_location = stats.SPECFILE_LOCATION
        stats.SPECFILE_LOCATION = io.StringIO(spec_str)
        self.assertRaises(stats.StatsError, stats.Stats)
        stats.SPECFILE_LOCATION = orig_spec_location

    def __send_command(self, stats, command_name, params=None):
        '''Emulate a command arriving to stats by directly calling callback'''
        return isc.config.ccsession.parse_answer(
            stats.command_handler(command_name, params))

    def test_start(self):
        # Define a separate exception class so we can be sure that's actually
        # the one raised in __check_start() below
        class CheckException(Exception):
            pass

        def __check_start(tested_stats):
            self.assertTrue(tested_stats.running)
            raise CheckException # terminate the loop

        # start without err
        stats = SimpleStats()
        self.assertFalse(stats.running)
        stats._check_command = lambda: __check_start(stats)
        # We are going to confirm start() will set running to True, avoiding
        # to fall into a loop with the exception trick.
        self.assertRaises(CheckException, stats.start)
        self.assertEqual(self.__send_command(stats, "status"),
                         (0, "Stats is up. (PID " + str(os.getpid()) + ")"))

    def test_shutdown(self):
        def __check_shutdown(tested_stats):
            self.assertTrue(tested_stats.running)
            self.assertEqual(self.__send_command(tested_stats, "shutdown"),
                             (0, None))
            self.assertFalse(tested_stats.running)
            # override get_interval() so it won't go poll statistics
            tested_stats.get_interval = lambda : 0

        stats = SimpleStats()
        stats._check_command = lambda: __check_shutdown(stats)
        stats.start()
        self.assertTrue(stats.mccs.stopped)

    def test_handlers(self):
        """Test command_handler"""

        __stats = SimpleStats()

        # 'show' command.  We're going to check the expected methods are
        # called in the expected order, and check the resulting response.
        # Details of each method are tested separately.
        call_log = []
        def __steal_method(fn_name, *arg):
            call_log.append((fn_name, arg))
            if fn_name == 'update_stat':
                return False        # "no error"
            if fn_name == 'showschema':
                return isc.config.create_answer(0, 'no error')

        # Fake some methods and attributes for inspection
        __stats.do_polling = lambda: __steal_method('polling')
        __stats.update_statistics_data = \
            lambda x, y, z: __steal_method('update_stat', x, y, z)
        __stats.update_modules = lambda: __steal_method('update_module')
        __stats.mccs.lname = 'test lname'
        __stats.statistics_data = {'Init': {'boot_time': self.const_datetime}}

        # skip initial polling
        stats.get_timestamp = lambda: 0
        __stats._lasttime_poll = 0

        stats.get_datetime = lambda: 42 # make the result predictable

        # now send the command
        self.assertEqual(
            self.__send_command(
                __stats, 'show',
                params={ 'owner' : 'Init', 'name'  : 'boot_time' }),
            (0, {'Init': {'boot_time': self.const_datetime}}))
        # Check if expected methods are called
        self.assertEqual([('update_stat',
                           ('Stats', 'test lname',
                            {'timestamp': 0,
                             'report_time': 42})),
                          ('update_module', ())], call_log)

        # Then update faked timestamp so the initial polling will happen, and
        # confirm that.
        call_log = []
        stats.get_timestamp = lambda: 10
        self.assertEqual(
            self.__send_command(
                __stats, 'show',
                params={ 'owner' : 'Init', 'name'  : 'boot_time' }),
            (0, {'Init': {'boot_time': self.const_datetime}}))
        self.assertEqual([('polling', ()),
                          ('update_stat',
                           ('Stats', 'test lname',
                            {'timestamp': 10,
                             'report_time': 42})),
                          ('update_module', ())], call_log)

        # 'status' command.  We can confirm the behavior without any fake
        self.assertEqual(
            self.__send_command(__stats, 'status'),
            (0, "Stats is up. (PID " + str(os.getpid()) + ")"))

        # 'showschema' command.  update_modules() will be called, which
        # (implicitly) confirms the correct method is called; further details
        # are tested separately.
        call_log = []
        (rcode, value) = self.__send_command(__stats, 'showschema')
        self.assertEqual([('update_module', ())], call_log)

        # Unknown command.  Error should be returned
        self.assertEqual(
            self.__send_command(__stats, '__UNKNOWN__'),
            (1, "Unknown command: '__UNKNOWN__'"))

    def test_update_modules(self):
        """Confirm the behavior of Stats.update_modules().

        It checks whether the expected command is sent to ConfigManager,
        and whether the answer from ConfigManager is handled as expected.

        """

        def __check_rpc_call(command, group):
            self.assertEqual('ConfigManager', group)
            self.assertEqual(command,
                             isc.config.ccsession.COMMAND_GET_STATISTICS_SPEC)
            answer_value = {'Init': [{
                        "item_name": "boot_time",
                        "item_type": "string",
                        "item_optional": False,
                        # Use a different default so we can check it below
                        "item_default": "2013-01-01T00:00:01Z",
                        "item_title": "Boot time",
                        "item_description": "dummy desc",
                        "item_format": "date-time"
                        }]}
            return answer_value

        self.stats = SimpleStats()
        self.stats.cc_session.rpc_call = __check_rpc_call

        self.stats.update_modules()

        # Stats is always incorporated.  For others, only the ones returned
        # by group_recvmsg() above is available.
        self.assertTrue('Stats' in self.stats.modules)
        self.assertTrue('Init' in self.stats.modules)
        self.assertFalse('Dummy' in self.stats.modules)

        my_statistics_data = stats.get_spec_defaults(
            self.stats.modules['Stats'].get_statistics_spec())
        self.assertTrue('report_time' in my_statistics_data)
        self.assertTrue('boot_time' in my_statistics_data)
        self.assertTrue('last_update_time' in my_statistics_data)
        self.assertTrue('timestamp' in my_statistics_data)
        self.assertTrue('lname' in my_statistics_data)
        self.assertEqual(my_statistics_data['report_time'],
                         self.const_default_datetime)
        self.assertEqual(my_statistics_data['boot_time'],
                         self.const_default_datetime)
        self.assertEqual(my_statistics_data['last_update_time'],
                         self.const_default_datetime)
        self.assertEqual(my_statistics_data['timestamp'], 0.0)
        self.assertEqual(my_statistics_data['lname'], "")
        my_statistics_data = stats.get_spec_defaults(
            self.stats.modules['Init'].get_statistics_spec())
        self.assertTrue('boot_time' in my_statistics_data)
        self.assertEqual(my_statistics_data['boot_time'],
                         "2013-01-01T00:00:01Z")

        # Error case
        def __raise_on_rpc_call(x, y):
            raise isc.config.RPCError(99, 'error')
        orig_parse_answer = stats.isc.config.ccsession.parse_answer
        self.stats.cc_session.rpc_call = __raise_on_rpc_call
        self.assertRaises(stats.StatsError, self.stats.update_modules)

    def test_get_statistics_data(self):
        """Confirm the behavior of Stats.get_statistics_data().

        It should first call update_modules(), and then retrieve the requested
        data from statistics_data.  We confirm this by fake update_modules()
        where we set the expected data in statistics_data.

        """
        self.stats = SimpleStats()
        def __faked_update_modules():
            self.stats.statistics_data = { \
                'Stats': {
                    'report_time': self.const_default_datetime,
                    'boot_time': None,
                    'last_update_time': None,
                    'timestamp': 0.0,
                    'lname': 'dummy name'
                    },
                'Init': { 'boot_time': None }
                }

        self.stats.update_modules = __faked_update_modules

        my_statistics_data = self.stats.get_statistics_data()
        self.assertTrue('Stats' in my_statistics_data)
        self.assertTrue('Init' in my_statistics_data)
        self.assertTrue('boot_time' in my_statistics_data['Init'])

        my_statistics_data = self.stats.get_statistics_data(owner='Stats')
        self.assertTrue('Stats' in my_statistics_data)
        self.assertTrue('report_time' in my_statistics_data['Stats'])
        self.assertTrue('boot_time' in my_statistics_data['Stats'])
        self.assertTrue('last_update_time' in my_statistics_data['Stats'])
        self.assertTrue('timestamp' in my_statistics_data['Stats'])
        self.assertTrue('lname' in my_statistics_data['Stats'])
        self.assertRaises(stats.StatsError, self.stats.get_statistics_data,
                          owner='Foo')

        my_statistics_data = self.stats.get_statistics_data(
            owner='Stats', name='report_time')
        self.assertEqual(my_statistics_data['Stats']['report_time'],
                         self.const_default_datetime)

        my_statistics_data = self.stats.get_statistics_data(
            owner='Stats', name='boot_time')
        self.assertTrue('boot_time' in my_statistics_data['Stats'])

        my_statistics_data = self.stats.get_statistics_data(
            owner='Stats', name='last_update_time')
        self.assertTrue('last_update_time' in my_statistics_data['Stats'])

        my_statistics_data = self.stats.get_statistics_data(
            owner='Stats', name='timestamp')
        self.assertEqual(my_statistics_data['Stats']['timestamp'], 0.0)

        my_statistics_data = self.stats.get_statistics_data(
            owner='Stats', name='lname')
        self.assertTrue(len(my_statistics_data['Stats']['lname']) >0)
        self.assertRaises(stats.StatsError, self.stats.get_statistics_data,
                          owner='Stats', name='Bar')
        self.assertRaises(stats.StatsError, self.stats.get_statistics_data,
                          owner='Foo', name='Bar')
        self.assertRaises(stats.StatsError, self.stats.get_statistics_data,
                          name='Bar')

    def test_update_statistics_data(self):
        """test for list-type statistics"""
        self.stats = SimpleStats()
        _test_exp1 = {
              'zonename': 'test1.example',
              'queries.tcp': 5,
              'queries.udp': 4
            }
        _test_exp2 = {
              'zonename': 'test2.example',
              'queries.tcp': 3,
              'queries.udp': 2
            }
        _test_exp3 = {}
        _test_exp4 = {
              'queries.udp': 4
            }
        _test_exp5_1 = {
              'queries.perzone': [
                { },
                {
                  'queries.udp': 9876
                }
              ]
            }
        _test_exp5_2 = {
              'queries.perzone[1]/queries.udp':
                  isc.cc.data.find(_test_exp5_1,
                                   'queries.perzone[1]/queries.udp')
            }
        # Success cases
        self.assertEqual(self.stats.statistics_data['Stats']['lname'],
                         self.stats.cc_session.lname)
        self.stats.update_statistics_data(
            'Stats', self.stats.cc_session.lname,
            {'lname': 'foo@bar'})
        self.assertEqual(self.stats.statistics_data['Stats']['lname'],
                         'foo@bar')
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo1', {'queries.perzone': [_test_exp1]}))
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['queries.perzone'],\
                             [_test_exp1])
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo1', {'queries.perzone': [_test_exp2]}))
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['queries.perzone'],\
                             [_test_exp2])
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo1', {'queries.perzone': [_test_exp1,_test_exp2]}))
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['queries.perzone'],
                         [_test_exp1,_test_exp2])
        # differential update
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo1', {'queries.perzone': [_test_exp3,_test_exp4]}))
        _new_data = stats.merge_oldnew(_test_exp2,_test_exp4)
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['queries.perzone'], \
                             [_test_exp1,_new_data])
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo1', _test_exp5_2))
        _new_data = stats.merge_oldnew(_new_data,
                                       _test_exp5_1['queries.perzone'][1])
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['queries.perzone'], \
                             [_test_exp1,_new_data])
        # Error cases
        self.assertEqual(self.stats.update_statistics_data('Stats', None,
                                                           {'lname': 0.0}),
                         ['0.0 should be a string'])
        self.assertEqual(self.stats.update_statistics_data('Dummy', None,
                                                           {'foo': 'bar'}),
                         ['unknown module name: Dummy'])
        self.assertEqual(self.stats.update_statistics_data(
                'Auth', 'foo1', {'queries.perzone': [None]}), ['None should be a map'])

    def test_update_statistics_data_pt2(self):
        """test for named_set-type statistics"""
        self.stats = SimpleStats()
        _test_exp1 = \
            { 'test10.example': { 'queries.tcp': 5, 'queries.udp': 4 } }
        _test_exp2 = \
            { 'test20.example': { 'queries.tcp': 3, 'queries.udp': 2 } }
        _test_exp3 = {}
        _test_exp4 = { 'test20.example': { 'queries.udp': 4 } }
        _test_exp5_1 = { 'test10.example': { 'queries.udp': 5432 } }
        _test_exp5_2 ={
              'nds_queries.perzone/test10.example/queries.udp':
                  isc.cc.data.find(_test_exp5_1, 'test10.example/queries.udp')
            }
        _test_exp6 = { 'foo/bar':  'brabra' }
        _test_exp7 = { 'foo[100]': 'bar' }
        # Success cases
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo1', {'nds_queries.perzone': _test_exp1}))
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['nds_queries.perzone'],\
                             _test_exp1)
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo1', {'nds_queries.perzone': _test_exp2}))
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['nds_queries.perzone'],\
                         dict(_test_exp1,**_test_exp2))
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo1', {'nds_queries.perzone':
                                 dict(_test_exp1, **_test_exp2)}))
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['nds_queries.perzone'],
                         dict(_test_exp1, **_test_exp2))
        # differential update
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo1', {'nds_queries.perzone':
                                 dict(_test_exp3, **_test_exp4)}))
        _new_val = dict(_test_exp1,
                        **stats.merge_oldnew(_test_exp2,_test_exp4))
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['nds_queries.perzone'],\
                             _new_val)
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo1', _test_exp5_2))
        _new_val = stats.merge_oldnew(_new_val, _test_exp5_1)
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['nds_queries.perzone'],\
                             _new_val)
        self.assertIsNone(self.stats.update_statistics_data(
            'Auth', 'foo2', _test_exp5_2))
        _new_val = stats.merge_oldnew(_new_val, _test_exp5_1)
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo2']['nds_queries.perzone'],\
                             _test_exp5_1)
        # Error cases
        self.assertEqual(self.stats.update_statistics_data(
                'Auth', 'foo1', {'nds_queries.perzone': None}),
                         ['None should be a map'])
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['nds_queries.perzone'],\
                             _new_val)
        self.assertEqual(self.stats.update_statistics_data(
                'Auth', 'foo1', _test_exp6), ['unknown item foo'])
        self.assertEqual(self.stats.statistics_data_bymid['Auth']\
                             ['foo1']['nds_queries.perzone'],\
                             _new_val)
        self.assertEqual(self.stats.update_statistics_data(
                'Init', 'bar1', _test_exp7), ["KeyError: 'foo'"])
        self.assertEqual(self.stats.update_statistics_data(
                'Foo', 'foo1', _test_exp6), ['unknown module name: Foo'])

    def test_update_statistics_data_withmid(self):
        self.stats = SimpleStats()

        # This test relies on existing statistics data at the Stats object.
        # This version of test prepares the data using the do_polling() method;
        # that's a bad practice because a unittest for a method
        # (update_statistics_data) would heavily depend on details of another
        # method (do_polling).  However, there's currently no direct test
        # for do_polling (which is also bad), so we still keep that approach,
        # partly for testing do_polling indirectly.  #2781 should provide
        # direct test for do_polling, with which this test scenario should
        # also be changed to be more stand-alone.

        # We use the knowledge of what kind of messages are sent via
        # do_polling, and return the following faked answer directly.
        create_answer = isc.config.ccsession.create_answer # shortcut
        self.stats._answers = [\
            # Answer for "show_processes"
            (create_answer(0, [[1034, 'b10-auth-1', 'Auth'],
                               [1035, 'b10-auth-2', 'Auth']]),  None),
            # Answers for "getstats".  2 for Auth instances and 1 for Init.
            # we return some bogus values for Init, but the rest of the test
            # doesn't need it, so it's okay.
            (create_answer(0, self.stats._auth_sdata), {'from': 'auth1'}),
            (create_answer(0, self.stats._auth_sdata), {'from': 'auth2'}),
            (create_answer(0, self.stats._auth_sdata), {'from': 'auth3'})
            ]
        # do_polling calls update_modules internally; in our scenario there's
        # no change in modules, so we make it no-op.
        self.stats.update_modules = lambda: None
        # Now call do_polling.
        self.stats.do_polling()

        # samples of query number
        bar1_tcp = 1001
        bar2_tcp = 2001
        bar3_tcp = 1002
        bar3_udp = 1003
        # two auth instances invoked, so we double the pre-set stat values
        sum_qtcp = self.stats._queries_tcp * 2
        sum_qudp = self.stats._queries_udp * 2
        self.stats.update_statistics_data('Auth', "bar1@foo",
                                          {'queries.tcp': bar1_tcp})
        self.assertTrue('Auth' in self.stats.statistics_data)
        self.assertTrue('queries.tcp' in self.stats.statistics_data['Auth'])
        self.assertEqual(self.stats.statistics_data['Auth']['queries.tcp'],
                         bar1_tcp + sum_qtcp)
        self.assertTrue('Auth' in self.stats.statistics_data_bymid)
        self.assertTrue('bar1@foo' in self.stats.statistics_data_bymid['Auth'])
        self.assertTrue('queries.tcp' in self.stats.statistics_data_bymid
                        ['Auth']['bar1@foo'])
        self.assertEqual(self.stats.statistics_data_bymid['Auth']['bar1@foo'],
                         {'queries.tcp': bar1_tcp})
        # check consolidation of statistics data even if there is
        # non-existent mid of Auth
        self.stats.update_statistics_data('Auth', "bar2@foo",
                                          {'queries.tcp': bar2_tcp})
        self.assertTrue('Auth' in self.stats.statistics_data)
        self.assertTrue('queries.tcp' in self.stats.statistics_data['Auth'])
        self.assertEqual(self.stats.statistics_data['Auth']['queries.tcp'],
                         bar1_tcp + bar2_tcp + sum_qtcp)
        self.assertTrue('Auth' in self.stats.statistics_data_bymid)
        self.assertTrue('bar1@foo' in self.stats.statistics_data_bymid['Auth'])
        self.assertTrue('queries.tcp' in self.stats.statistics_data_bymid['Auth']['bar1@foo'])
        self.assertEqual(self.stats.statistics_data_bymid['Auth']['bar1@foo'],
                         {'queries.tcp': bar1_tcp})
        self.assertEqual(self.stats.statistics_data_bymid['Auth']['bar2@foo'],
                         {'queries.tcp': bar2_tcp})
        # kill running Auth but the statistics data doesn't change
        self.base.auth2.server.shutdown()
        self.stats.update_statistics_data()
        self.assertTrue('Auth' in self.stats.statistics_data)
        self.assertTrue('queries.tcp' in self.stats.statistics_data['Auth'])
        self.assertTrue('queries.udp' in self.stats.statistics_data['Auth'])
        self.assertEqual(self.stats.statistics_data['Auth']['queries.tcp'],
                         bar1_tcp + bar2_tcp + sum_qtcp)
        self.assertEqual(self.stats.statistics_data['Auth']['queries.udp'],
                         sum_qudp)
        self.assertTrue('Auth' in self.stats.statistics_data_bymid)
        # restore statistics data of killed auth
        # self.base.b10_init.server.pid_list = [ killed ] + self.base.b10_init.server.pid_list[:]
        self.stats.update_statistics_data('Auth',
                                          "bar1@foo",
                                          {'queries.tcp': bar1_tcp})
        # set another mid of Auth
        self.stats.update_statistics_data('Auth',
                                          "bar3@foo",
                                          {'queries.tcp':bar3_tcp,
                                           'queries.udp':bar3_udp})
        self.assertTrue('Auth' in self.stats.statistics_data)
        self.assertTrue('queries.tcp' in self.stats.statistics_data['Auth'])
        self.assertTrue('queries.udp' in self.stats.statistics_data['Auth'])
        self.assertEqual(self.stats.statistics_data['Auth']['queries.tcp'],
                         bar1_tcp + bar2_tcp + bar3_tcp + sum_qtcp)
        self.assertEqual(self.stats.statistics_data['Auth']['queries.udp'],
                         bar3_udp + sum_qudp)
        self.assertTrue('Auth' in self.stats.statistics_data_bymid)
        self.assertTrue('bar1@foo' in self.stats.statistics_data_bymid['Auth'])
        self.assertTrue('bar3@foo' in self.stats.statistics_data_bymid['Auth'])
        self.assertTrue('queries.tcp' in self.stats.statistics_data_bymid['Auth']['bar1@foo'])
        self.assertTrue('queries.udp' in self.stats.statistics_data_bymid['Auth']['bar3@foo'])
        self.assertTrue('queries.udp' in self.stats.statistics_data_bymid['Auth']['bar3@foo'])
        self.assertEqual(self.stats.statistics_data_bymid['Auth']['bar1@foo']['queries.tcp'], bar1_tcp)
        self.assertEqual(self.stats.statistics_data_bymid['Auth']['bar3@foo']['queries.tcp'], bar3_tcp)
        self.assertEqual(self.stats.statistics_data_bymid['Auth']['bar3@foo']['queries.udp'], bar3_udp)

    def test_config(self):
        orig_get_timestamp = stats.get_timestamp
        stats.get_timestamp = lambda : self.const_timestamp
        stat = SimpleStats()

        # test updating poll-interval
        self.assertEqual(stat.config['poll-interval'], 60)
        self.assertEqual(stat.get_interval(), 60)
        self.assertEqual(stat.next_polltime, self.const_timestamp + 60)
        self.assertEqual(stat.config_handler({'poll-interval': 120}),
                         isc.config.create_answer(0))
        self.assertEqual(stat.config['poll-interval'], 120)
        self.assertEqual(stat.get_interval(), 120)
        self.assertEqual(stat.next_polltime, self.const_timestamp + 120)
        stats.get_timestamp = orig_get_timestamp
        self.assertEqual(stat.config_handler({'poll-interval': "foo"}),
                         isc.config.create_answer(1, 'foo should be an integer'))
        self.assertEqual(stat.config_handler({'poll-interval': -1}),
                         isc.config.create_answer(1, 'Negative integer ignored'))
        # unknown item
        self.assertEqual(
            stat.config_handler({'_UNKNOWN_KEY_': None}),
            isc.config.ccsession.create_answer(
                1, "unknown item _UNKNOWN_KEY_"))
        # test no change if zero interval time
        self.assertEqual(stat.config_handler({'poll-interval': 0}),
                         isc.config.create_answer(0))
        self.assertEqual(stat.config['poll-interval'], 0)

        # see the comment for test_update_statistics_data_withmid.  We abuse
        # do_polling here, too.  With #2781 we should make it more direct.
        create_answer = isc.config.ccsession.create_answer # shortcut
        stat._answers = [\
            # Answer for "show_processes"
            (create_answer(0, []),  None),
            # Answers for "getstats" for Init (the other one for Auth, but
            # that doesn't matter for this test)
            (create_answer(0, stat._init_sdata), {'from': 'init'}),
            (create_answer(0, stat._init_sdata), {'from': 'init'})
            ]
        stat.update_modules = lambda: None

        self.assertEqual(
            self.__send_command(
                stat, 'show',
                params={ 'owner' : 'Init', 'name'  : 'boot_time' }),
            (0, {'Init': {'boot_time': self.const_datetime}}))

    def test_commands(self):
        self.stats = stats.Stats()

        # status
        self.assertEqual(self.stats.command_status(),
                isc.config.create_answer(
                0, "Stats is up. (PID " + str(os.getpid()) + ")"))

        # shutdown
        self.stats.running = True
        self.assertEqual(self.stats.command_shutdown(),
                         isc.config.create_answer(0))
        self.assertFalse(self.stats.running)

    @unittest.skipIf(sys.version_info >= (3, 3), "Unsupported in Python 3.3 or higher")
    def test_command_show(self):
        # two auth instances invoked
        list_auth = [ self.base.auth.server,
                      self.base.auth2.server ]
        sum_qtcp = 0
        sum_qudp = 0
        sum_qtcp_perzone1 = 0
        sum_qudp_perzone1 = 0
        sum_qtcp_perzone2 = 4 * len(list_auth)
        sum_qudp_perzone2 = 3 * len(list_auth)
        sum_qtcp_nds_perzone10 = 0
        sum_qudp_nds_perzone10 = 0
        sum_qtcp_nds_perzone20 = 4 * len(list_auth)
        sum_qudp_nds_perzone20 = 3 * len(list_auth)
        self.stats = stats.Stats()
        self.assertEqual(self.stats.command_show(owner='Foo', name=None),
                         isc.config.create_answer(
                1, "specified arguments are incorrect: owner: Foo, name: None"))
        self.assertEqual(self.stats.command_show(owner='Foo', name='_bar_'),
                         isc.config.create_answer(
                1, "specified arguments are incorrect: owner: Foo, name: _bar_"))
        self.assertEqual(self.stats.command_show(owner='Foo', name='bar'),
                         isc.config.create_answer(
                1, "specified arguments are incorrect: owner: Foo, name: bar"))

        for a in list_auth:
            sum_qtcp += a.queries_tcp
            sum_qudp += a.queries_udp
            sum_qtcp_perzone1 += a.queries_per_zone[0]['queries.tcp']
            sum_qudp_perzone1 += a.queries_per_zone[0]['queries.udp']
            sum_qtcp_nds_perzone10 += a.nds_queries_per_zone['test10.example']['queries.tcp']
            sum_qudp_nds_perzone10 += a.nds_queries_per_zone['test10.example']['queries.udp']

        self.assertEqual(self.stats.command_show(owner='Auth'),
                         isc.config.create_answer(
                0, {'Auth':{ 'queries.udp': sum_qudp,
                     'queries.tcp': sum_qtcp,
                     'queries.perzone': [{ 'zonename': 'test1.example',
                                           'queries.udp': sum_qudp_perzone1,
                                           'queries.tcp': sum_qtcp_perzone1 },
                                         { 'zonename': 'test2.example',
                                           'queries.udp': sum_qudp_perzone2,
                                           'queries.tcp': sum_qtcp_perzone2 }
                                         ],
                     'nds_queries.perzone': { 'test10.example' : {
                                              'queries.udp': sum_qudp_nds_perzone10,
                                              'queries.tcp': sum_qtcp_nds_perzone10 },
                                              'test20.example' : {
                                              'queries.udp': sum_qudp_nds_perzone20,
                                              'queries.tcp': sum_qtcp_nds_perzone20 }
                             }}}))
        self.assertEqual(self.stats.command_show(owner='Auth', name='queries.udp'),
                         isc.config.create_answer(
                0, {'Auth': {'queries.udp': sum_qudp}}))
        self.assertEqual(self.stats.command_show(owner='Auth', name='queries.perzone'),
                         isc.config.create_answer(
                0, {'Auth': {'queries.perzone': [
                            { 'zonename': 'test1.example',
                              'queries.udp': sum_qudp_perzone1,
                              'queries.tcp': sum_qtcp_perzone1 },
                            { 'zonename': 'test2.example',
                              'queries.udp': sum_qudp_perzone2,
                              'queries.tcp': sum_qtcp_perzone2 }]}}))
        self.assertEqual(self.stats.command_show(owner='Auth', name='nds_queries.perzone'),
                         isc.config.create_answer(
                0, {'Auth': {'nds_queries.perzone': {
                            'test10.example': {
                                'queries.udp': sum_qudp_nds_perzone10,
                                'queries.tcp': sum_qtcp_nds_perzone10 },
                            'test20.example': {
                                'queries.udp': sum_qudp_nds_perzone20,
                                'queries.tcp': sum_qtcp_nds_perzone20 }}}}))
        orig_get_datetime = stats.get_datetime
        orig_get_timestamp = stats.get_timestamp
        stats.get_datetime = lambda x=None: self.const_datetime
        stats.get_timestamp = lambda : self.const_timestamp
        self.assertEqual(self.stats.command_show(owner='Stats', name='report_time'),
                         isc.config.create_answer(
                0, {'Stats': {'report_time':self.const_datetime}}))
        self.assertEqual(self.stats.command_show(owner='Stats', name='timestamp'),
                         isc.config.create_answer(
                0, {'Stats': {'timestamp':self.const_timestamp}}))
        stats.get_datetime = orig_get_datetime
        stats.get_timestamp = orig_get_timestamp
        self.stats.modules[self.stats.module_name] = isc.config.module_spec.ModuleSpec(
            { "module_name": self.stats.module_name,
              "statistics": [] } )
        self.assertRaises(
            stats.StatsError, self.stats.command_show, owner=self.stats.module_name, name='bar')

    def test_command_showchema(self):
        self.stats = stats.Stats()
        (rcode, value) = isc.config.ccsession.parse_answer(
            self.stats.command_showschema())
        self.assertEqual(rcode, 0)
        self.assertEqual(len(value), 3)
        self.assertTrue('Stats' in value)
        self.assertTrue('Init' in value)
        self.assertTrue('Auth' in value)
        self.assertFalse('__Dummy__' in value)
        schema = value['Stats']
        self.assertEqual(len(schema), 5)
        for item in schema:
            self.assertTrue(len(item) == 6 or len(item) == 7)
            self.assertTrue('item_name' in item)
            self.assertTrue('item_type' in item)
            self.assertTrue('item_optional' in item)
            self.assertTrue('item_default' in item)
            self.assertTrue('item_title' in item)
            self.assertTrue('item_description' in item)
            if len(item) == 7:
                self.assertTrue('item_format' in item)

        schema = value['Init']
        self.assertEqual(len(schema), 1)
        for item in schema:
            self.assertTrue(len(item) == 7)
            self.assertTrue('item_name' in item)
            self.assertTrue('item_type' in item)
            self.assertTrue('item_optional' in item)
            self.assertTrue('item_default' in item)
            self.assertTrue('item_title' in item)
            self.assertTrue('item_description' in item)
            self.assertTrue('item_format' in item)

        schema = value['Auth']
        self.assertEqual(len(schema), 4)
        for item in schema:
            if item['item_type'] == 'list' or item['item_type'] == 'named_set':
                self.assertEqual(len(item), 7)
            else:
                self.assertEqual(len(item), 6)
            self.assertTrue('item_name' in item)
            self.assertTrue('item_type' in item)
            self.assertTrue('item_optional' in item)
            self.assertTrue('item_default' in item)
            self.assertTrue('item_title' in item)
            self.assertTrue('item_description' in item)

        (rcode, value) = isc.config.ccsession.parse_answer(
            self.stats.command_showschema(owner='Stats'))
        self.assertEqual(rcode, 0)
        self.assertTrue('Stats' in value)
        self.assertFalse('Init' in value)
        self.assertFalse('Auth' in value)
        for item in value['Stats']:
            self.assertTrue(len(item) == 6 or len(item) == 7)
            self.assertTrue('item_name' in item)
            self.assertTrue('item_type' in item)
            self.assertTrue('item_optional' in item)
            self.assertTrue('item_default' in item)
            self.assertTrue('item_title' in item)
            self.assertTrue('item_description' in item)
            if len(item) == 7:
                self.assertTrue('item_format' in item)

        (rcode, value) = isc.config.ccsession.parse_answer(
            self.stats.command_showschema(owner='Stats', name='report_time'))
        self.assertEqual(rcode, 0)
        self.assertTrue('Stats' in value)
        self.assertFalse('Init' in value)
        self.assertFalse('Auth' in value)
        self.assertEqual(len(value['Stats'][0]), 7)
        self.assertTrue('item_name' in value['Stats'][0])
        self.assertTrue('item_type' in value['Stats'][0])
        self.assertTrue('item_optional' in value['Stats'][0])
        self.assertTrue('item_default' in value['Stats'][0])
        self.assertTrue('item_title' in value['Stats'][0])
        self.assertTrue('item_description' in value['Stats'][0])
        self.assertTrue('item_format' in value['Stats'][0])
        self.assertEqual(value['Stats'][0]['item_name'], 'report_time')
        self.assertEqual(value['Stats'][0]['item_format'], 'date-time')

        self.assertEqual(self.stats.command_showschema(owner='Foo'),
                         isc.config.create_answer(
                1, "specified arguments are incorrect: owner: Foo, name: None"))
        self.assertEqual(self.stats.command_showschema(owner='Foo', name='bar'),
                         isc.config.create_answer(
                1, "specified arguments are incorrect: owner: Foo, name: bar"))
        self.assertEqual(self.stats.command_showschema(owner='Auth'),
                         isc.config.create_answer(
                0, {'Auth': [{
                        "item_default": 0,
                        "item_description": "A number of total query counts which all auth servers receive over TCP since they started initially",
                        "item_name": "queries.tcp",
                        "item_optional": False,
                        "item_title": "Queries TCP",
                        "item_type": "integer"
                        },
                    {
                        "item_default": 0,
                        "item_description": "A number of total query counts which all auth servers receive over UDP since they started initially",
                        "item_name": "queries.udp",
                        "item_optional": False,
                        "item_title": "Queries UDP",
                        "item_type": "integer"
                        },
                    {
                        "item_name": "queries.perzone",
                        "item_type": "list",
                        "item_optional": False,
                        "item_default": [
                            {
                                "zonename" : "test1.example",
                                "queries.udp" : 1,
                                "queries.tcp" : 2
                                },
                            {
                                "zonename" : "test2.example",
                                "queries.udp" : 3,
                                "queries.tcp" : 4
                                }
                        ],
                        "item_title": "Queries per zone",
                        "item_description": "Queries per zone",
                        "list_item_spec": {
                            "item_name": "zones",
                            "item_type": "map",
                            "item_optional": False,
                            "item_default": {},
                            "map_item_spec": [
                                {
                                    "item_name": "zonename",
                                    "item_type": "string",
                                    "item_optional": False,
                                    "item_default": "",
                                    "item_title": "Zonename",
                                    "item_description": "Zonename"
                                    },
                                {
                                    "item_name": "queries.udp",
                                    "item_type": "integer",
                                    "item_optional": False,
                                    "item_default": 0,
                                    "item_title": "Queries UDP per zone",
                                    "item_description": "A number of UDP query counts per zone"
                                    },
                                {
                                    "item_name": "queries.tcp",
                                    "item_type": "integer",
                                    "item_optional": False,
                                    "item_default": 0,
                                    "item_title": "Queries TCP per zone",
                                    "item_description": "A number of TCP query counts per zone"
                                    }
                                ]
                            }
                        },
                    {
                        "item_name": "nds_queries.perzone",
                        "item_type": "named_set",
                        "item_optional": False,
                        "item_default": {
                            "test10.example" : {
                                "queries.udp" : 1,
                                "queries.tcp" : 2
                            },
                            "test20.example" : {
                                "queries.udp" : 3,
                                "queries.tcp" : 4
                            }
                        },
                        "item_title": "Queries per zone",
                        "item_description": "Queries per zone",
                        "named_set_item_spec": {
                            "item_name": "zonename",
                            "item_type": "map",
                            "item_optional": False,
                            "item_default": {},
                            "item_title": "Zonename",
                            "item_description": "Zonename",
                            "map_item_spec": [
                                {
                                    "item_name": "queries.udp",
                                    "item_type": "integer",
                                    "item_optional": False,
                                    "item_default": 0,
                                    "item_title": "Queries UDP per zone",
                                    "item_description": "A number of UDP query counts per zone"
                                    },
                                {
                                    "item_name": "queries.tcp",
                                    "item_type": "integer",
                                    "item_optional": False,
                                    "item_default": 0,
                                    "item_title": "Queries TCP per zone",
                                    "item_description": "A number of TCP query counts per zone"
                                    }
                                ]
                            }
                        }]}))
        self.assertEqual(self.stats.command_showschema(owner='Auth', name='queries.tcp'),
                         isc.config.create_answer(
                0, {'Auth': [{
                    "item_default": 0,
                    "item_description": "A number of total query counts which all auth servers receive over TCP since they started initially",
                    "item_name": "queries.tcp",
                    "item_optional": False,
                    "item_title": "Queries TCP",
                    "item_type": "integer"
                    }]}))
        self.assertEqual(self.stats.command_showschema(owner='Auth', name='queries.perzone'),
                         isc.config.create_answer(
                0, {'Auth':[{
                    "item_name": "queries.perzone",
                    "item_type": "list",
                    "item_optional": False,
                    "item_default": [
                        {
                            "zonename" : "test1.example",
                            "queries.udp" : 1,
                            "queries.tcp" : 2
                            },
                        {
                            "zonename" : "test2.example",
                            "queries.udp" : 3,
                            "queries.tcp" : 4
                            }
                    ],
                    "item_title": "Queries per zone",
                    "item_description": "Queries per zone",
                    "list_item_spec": {
                        "item_name": "zones",
                        "item_type": "map",
                        "item_optional": False,
                        "item_default": {},
                        "map_item_spec": [
                            {
                                "item_name": "zonename",
                                "item_type": "string",
                                "item_optional": False,
                                "item_default": "",
                                "item_title": "Zonename",
                                "item_description": "Zonename"
                                },
                            {
                                "item_name": "queries.udp",
                                "item_type": "integer",
                                "item_optional": False,
                                "item_default": 0,
                                "item_title": "Queries UDP per zone",
                                "item_description": "A number of UDP query counts per zone"
                                },
                            {
                                "item_name": "queries.tcp",
                                "item_type": "integer",
                                "item_optional": False,
                                "item_default": 0,
                                "item_title": "Queries TCP per zone",
                                "item_description": "A number of TCP query counts per zone"
                                }
                            ]
                         }
                     }]}))
        self.assertEqual(self.stats.command_showschema(owner='Auth', name='nds_queries.perzone'),
                         isc.config.create_answer(
                0, {'Auth':[{
                    "item_name": "nds_queries.perzone",
                    "item_type": "named_set",
                    "item_optional": False,
                    "item_default": {
                        "test10.example" : {
                            "queries.udp" : 1,
                            "queries.tcp" : 2
                        },
                        "test20.example" : {
                            "queries.udp" : 3,
                            "queries.tcp" : 4
                        }
                    },
                    "item_title": "Queries per zone",
                    "item_description": "Queries per zone",
                    "named_set_item_spec": {
                        "item_name": "zonename",
                        "item_type": "map",
                        "item_optional": False,
                        "item_default": {},
                        "item_title": "Zonename",
                        "item_description": "Zonename",
                        "map_item_spec": [
                            {
                                "item_name": "queries.udp",
                                "item_type": "integer",
                                "item_optional": False,
                                "item_default": 0,
                                "item_title": "Queries UDP per zone",
                                "item_description": "A number of UDP query counts per zone"
                                },
                            {
                                "item_name": "queries.tcp",
                                "item_type": "integer",
                                "item_optional": False,
                                "item_default": 0,
                                "item_title": "Queries TCP per zone",
                                "item_description": "A number of TCP query counts per zone"
                                }
                            ]
                        }
                    }]}))

        self.assertEqual(self.stats.command_showschema(owner='Stats', name='bar'),
                         isc.config.create_answer(
                1, "specified arguments are incorrect: owner: Stats, name: bar"))
        self.assertEqual(self.stats.command_showschema(name='bar'),
                         isc.config.create_answer(
                1, "module name is not specified"))

    @unittest.skipIf(sys.version_info >= (3, 3), "Unsupported in Python 3.3 or higher")
    def test_polling(self):
        stats_server = ThreadingServerManager(MyStats)
        stat = stats_server.server
        stats_server.run()
        self.assertEqual(
            send_command('show', 'Stats'),
            (0, stat.statistics_data))
        # check statistics data of 'Init'
        b10_init = self.base.b10_init.server
        self.assertEqual(
            stat.statistics_data_bymid['Init'][b10_init.cc_session.lname],
            {'boot_time': self.const_datetime})
        self.assertEqual(
            len(stat.statistics_data_bymid['Init']), 1)
        self.assertEqual(
            stat.statistics_data['Init'],
            {'boot_time': self.const_datetime})
        # check statistics data of each 'Auth' instances
        list_auth = ['', '2']
        for i in list_auth:
            auth = getattr(self.base,"auth"+i).server
            for s in stat.statistics_data_bymid['Auth'].values():
                self.assertEqual(
                    s, {'queries.perzone': auth.queries_per_zone,
                        'nds_queries.perzone': auth.nds_queries_per_zone,
                        'queries.tcp': auth.queries_tcp,
                        'queries.udp': auth.queries_udp})
            n = len(stat.statistics_data_bymid['Auth'])
            self.assertEqual(n, len(list_auth))
            # check consolidation of statistics data of the auth
            # instances
            self.assertEqual(
                stat.statistics_data['Auth'],
                {'queries.perzone': [
                        {'zonename':
                             auth.queries_per_zone[0]['zonename'],
                         'queries.tcp':
                             auth.queries_per_zone[0]['queries.tcp']*n,
                         'queries.udp':
                             auth.queries_per_zone[0]['queries.udp']*n},
                        {'zonename': "test2.example",
                         'queries.tcp': 4*n,
                         'queries.udp': 3*n },
                        ],
                 'nds_queries.perzone': {
                         'test10.example': {
                             'queries.tcp':
                                 auth.nds_queries_per_zone['test10.example']['queries.tcp']*n,
                             'queries.udp':
                                 auth.nds_queries_per_zone['test10.example']['queries.udp']*n},
                         'test20.example': {
                             'queries.tcp':
                                 4*n,
                             'queries.udp':
                                 3*n},
                         },
                 'queries.tcp': auth.queries_tcp*n,
                 'queries.udp': auth.queries_udp*n})
        # check statistics data of 'Stats'
        self.assertEqual(
            len(stat.statistics_data['Stats']), 5)
        self.assertTrue('boot_time' in
            stat.statistics_data['Stats'])
        self.assertTrue('last_update_time' in
            stat.statistics_data['Stats'])
        self.assertTrue('report_time' in
            stat.statistics_data['Stats'])
        self.assertTrue('timestamp' in
            stat.statistics_data['Stats'])
        self.assertEqual(
            stat.statistics_data['Stats']['lname'],
            stat.mccs._session.lname)
        stats_server.shutdown()

    def test_polling2(self):
        # set invalid statistics
        b10_init = self.base.b10_init.server
        b10_init.statistics_data = {'boot_time':1}
        stats_server = ThreadingServerManager(MyStats)
        stat = stats_server.server
        stats_server.run()
        self.assertEqual(
            send_command('status', 'Stats'),
            (0, "Stats is up. (PID " + str(os.getpid()) + ")"))
        # check default statistics data of 'Init'
        self.assertEqual(
            stat.statistics_data['Init'],
            {'boot_time': self.const_default_datetime})
        stats_server.shutdown()

class TestOSEnv(unittest.TestCase):
    def test_osenv(self):
        """
        test for the environ variable "B10_FROM_SOURCE"
        "B10_FROM_SOURCE" is set in Makefile
        """
        # test case having B10_FROM_SOURCE
        self.assertTrue("B10_FROM_SOURCE" in os.environ)
        self.assertEqual(stats.SPECFILE_LOCATION, \
                             os.environ["B10_FROM_SOURCE"] + os.sep + \
                             "src" + os.sep + "bin" + os.sep + "stats" + \
                             os.sep + "stats.spec")
        # test case not having B10_FROM_SOURCE
        path = os.environ["B10_FROM_SOURCE"]
        os.environ.pop("B10_FROM_SOURCE")
        self.assertFalse("B10_FROM_SOURCE" in os.environ)
        # import stats again
        imp.reload(stats)
        # revert the changes
        os.environ["B10_FROM_SOURCE"] = path
        imp.reload(stats)

if __name__ == "__main__":
    isc.log.resetUnitTestRootLogger()
    unittest.main()