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

"""
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 stats
import isc.cc.session
from test_utils import BaseModules, ThreadingServerManager, MyStats, SignalHandler, send_command, send_shutdown
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'} ] }
        ]

    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' } })
        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)

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.stats = stats.Stats()
        self.const_timestamp = 1308730448.965706
        self.const_datetime = '2011-06-22T08:14:08Z'
        self.const_default_datetime = '1970-01-01T00:00:00Z'

    def tearDown(self):
        self.base.shutdown()
        # reset the signal handler
        self.sig_handler.reset()

    def test_init(self):
        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.assertTrue('command_set' in self.stats.callbacks)

    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 test_start(self):
        # start without err
        self.stats_server = ThreadingServerManager(MyStats)
        self.stats = self.stats_server.server
        self.assertFalse(self.stats.running)
        self.stats_server.run()
        self.assertEqual(send_command("status", "Stats"),
                (0, "Stats is up. (PID " + str(os.getpid()) + ")"))
        self.assertTrue(self.stats.running)
        # Due to a race-condition related to the threads used in these
        # tests, use of the mock session and the stopped check (see
        # below), are temporarily disabled
        # See ticket #1668
        # Override moduleCCSession so we can check if send_stopping is called
        #self.stats.mccs = MockModuleCCSession()
        self.assertEqual(send_shutdown("Stats"), (0, None))
        self.assertFalse(self.stats.running)
        # Call server.shutdown with argument True so the thread.join() call
        # blocks and we are sure the main loop has finished (and set
        # mccs.stopped)
        self.stats_server.shutdown(True)
        # Also temporarily disabled for #1668, see above
        #self.assertTrue(self.stats.mccs.stopped)

        # start with err
        self.stats = stats.Stats()
        self.stats.update_statistics_data = lambda x,**y: ['an error']
        self.assertRaises(stats.StatsError, self.stats.start)

    def test_handlers(self):
        self.stats_server = ThreadingServerManager(MyStats)
        self.stats = self.stats_server.server
        self.stats_server.run()
        # config_handler
        self.assertEqual(self.stats.config_handler({'foo':'bar'}),
                         isc.config.create_answer(0))

        # command_handler
        self.base.boss.server._started.wait()
        self.base.boss.server._started.clear()
        self.assertEqual(
            send_command(
                'show', 'Stats',
                params={ 'owner' : 'Boss',
                  'name'  : 'boot_time' }),
            (0, {'Boss': {'boot_time': self.const_datetime}}))
        self.assertEqual(
            send_command(
                'set', 'Stats',
                params={ 'owner' : 'Boss',
                  'data'  : { 'boot_time' : self.const_datetime } }),
            (0, None))
        self.assertEqual(
            send_command(
                'show', 'Stats',
                params={ 'owner' : 'Boss',
                  'name'  : 'boot_time' }),
            (0, {'Boss': {'boot_time': self.const_datetime}}))
        self.assertEqual(
            send_command('status', 'Stats'),
            (0, "Stats is up. (PID " + str(os.getpid()) + ")"))

        (rcode, value) = send_command('show', 'Stats')
        self.assertEqual(rcode, 0)
        self.assertEqual(len(value), 3)
        self.assertTrue('Boss' in value)
        self.assertTrue('Stats' in value)
        self.assertTrue('Auth' in value)
        self.assertEqual(len(value['Stats']), 5)
        self.assertEqual(len(value['Boss']), 1)
        self.assertTrue('boot_time' in value['Boss'])
        self.assertEqual(value['Boss']['boot_time'], self.const_datetime)
        self.assertTrue('report_time' in value['Stats'])
        self.assertTrue('boot_time' in value['Stats'])
        self.assertTrue('last_update_time' in value['Stats'])
        self.assertTrue('timestamp' in value['Stats'])
        self.assertTrue('lname' in value['Stats'])
        (rcode, value) = send_command('showschema', 'Stats')
        self.assertEqual(rcode, 0)
        self.assertEqual(len(value), 3)
        self.assertTrue('Boss' in value)
        self.assertTrue('Stats' in value)
        self.assertTrue('Auth' in value)
        self.assertEqual(len(value['Stats']), 5)
        self.assertEqual(len(value['Boss']), 1)
        for item in value['Boss']:
            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)
        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)

        self.assertEqual(
            send_command('__UNKNOWN__', 'Stats'),
            (1, "Unknown command: '__UNKNOWN__'"))

        self.stats_server.shutdown()

    def test_update_modules(self):
        self.assertEqual(len(self.stats.modules), 0)
        self.stats.update_modules()
        self.assertTrue('Stats' in self.stats.modules)
        self.assertTrue('Boss' 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['Boss'].get_statistics_spec())
        self.assertTrue('boot_time' in my_statistics_data)
        self.assertEqual(my_statistics_data['boot_time'], self.const_default_datetime)
        orig_parse_answer = stats.isc.config.ccsession.parse_answer
        stats.isc.config.ccsession.parse_answer = lambda x: (99, 'error')
        self.assertRaises(stats.StatsError, self.stats.update_modules)
        stats.isc.config.ccsession.parse_answer = orig_parse_answer

    def test_get_statistics_data(self):
        my_statistics_data = self.stats.get_statistics_data()
        self.assertTrue('Stats' in my_statistics_data)
        self.assertTrue('Boss' in my_statistics_data)
        self.assertTrue('boot_time' in my_statistics_data['Boss'])
        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.assertEqual(my_statistics_data['Stats']['boot_time'], self.const_default_datetime)
        my_statistics_data = self.stats.get_statistics_data(owner='Stats', name='last_update_time')
        self.assertEqual(my_statistics_data['Stats']['last_update_time'], self.const_default_datetime)
        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.assertEqual(my_statistics_data, {'Stats': {'lname':''}})
        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):
        self.stats.update_statistics_data(owner='Stats', lname='foo@bar')
        self.assertTrue('Stats' in self.stats.statistics_data)
        my_statistics_data = self.stats.statistics_data['Stats']
        self.assertEqual(my_statistics_data['lname'], 'foo@bar')
        self.stats.update_statistics_data(owner='Stats', last_update_time=self.const_datetime)
        self.assertTrue('Stats' in self.stats.statistics_data)
        my_statistics_data = self.stats.statistics_data['Stats']
        self.assertEqual(my_statistics_data['last_update_time'], self.const_datetime)
        self.assertEqual(self.stats.update_statistics_data(owner='Stats', lname=0.0),
                         ['0.0 should be a string'])
        self.assertEqual(self.stats.update_statistics_data(owner='Dummy', foo='bar'),
                         ['unknown module name: Dummy'])

    def test_update_statistics_data_withpid(self):
        # one pid of Auth
        self.stats.update_statistics_data(owner='Auth',
                                          pid=9999,
                                          **{'queries.tcp':1001})
        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'], 1001)
        self.assertTrue('Auth' in self.stats.statistics_data_bypid)
        self.assertTrue(9999 in self.stats.statistics_data_bypid['Auth'])
        self.assertTrue('queries.tcp' in self.stats.statistics_data_bypid['Auth'][9999])
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9999]['queries.tcp'], 1001)
        self.assertEqual(self.stats.statistics_data_bypid,
                         {'Auth': {9999: {'queries.tcp': 1001}}})
        # non-existent pid of Auth, but no changes in statistics data
        self.stats.update_statistics_data(owner='Auth',
                                          pid=10000,
                                          **{'queries.tcp':2001})
        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'], 1001)
        self.assertTrue('Auth' in self.stats.statistics_data_bypid)
        self.assertTrue(9999 in self.stats.statistics_data_bypid['Auth'])
        self.assertTrue('queries.tcp' in self.stats.statistics_data_bypid['Auth'][9999])
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9999]['queries.tcp'], 1001)
        self.assertEqual(self.stats.statistics_data_bypid,
                         {'Auth': {9999: {'queries.tcp': 1001}}})
        # kill running Auth, then statistics is reset
        self.assertEqual(self.base.boss.server.pid_list[0][0], 9999)
        killed = self.base.boss.server.pid_list.pop(0)
        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'], 0)
        self.assertEqual(self.stats.statistics_data['Auth']['queries.udp'], 0)
        self.assertFalse('Auth' in self.stats.statistics_data_bypid)
        # restore statistics data of killed auth
        self.base.boss.server.pid_list = [ killed ] + self.base.boss.server.pid_list[:]
        self.stats.update_statistics_data(owner='Auth',
                                          pid=9999,
                                          **{'queries.tcp':1001})
        # another pid of Auth
        self.stats.update_statistics_data(owner='Auth',
                                          pid=9998,
                                          **{'queries.tcp':1002,
                                             'queries.udp':1003})
        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'], 2003)
        self.assertEqual(self.stats.statistics_data['Auth']['queries.udp'], 1003)
        self.assertTrue('Auth' in self.stats.statistics_data_bypid)
        self.assertTrue(9999 in self.stats.statistics_data_bypid['Auth'])
        self.assertTrue(9998 in self.stats.statistics_data_bypid['Auth'])
        self.assertTrue('queries.tcp' in self.stats.statistics_data_bypid['Auth'][9999])
        self.assertTrue('queries.udp' in self.stats.statistics_data_bypid['Auth'][9998])
        self.assertTrue('queries.udp' in self.stats.statistics_data_bypid['Auth'][9998])
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9999]['queries.tcp'], 1001)
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9998]['queries.tcp'], 1002)
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9998]['queries.udp'], 1003)

    def test_commands(self):
        # 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)

    def test_command_show(self):
        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"))
        self.assertEqual(self.stats.command_show(owner='Auth'),
                         isc.config.create_answer(
                0, {'Auth':{ 'queries.udp': 0,
                     'queries.tcp': 0,
                     'queries.perzone': [{ 'zonename': 'test1.example',
                                           'queries.udp': 1,
                                           'queries.tcp': 2 },
                                         { 'zonename': 'test2.example',
                                           'queries.udp': 3,
                                           'queries.tcp': 4 }] }}))
        self.assertEqual(self.stats.command_show(owner='Auth', name='queries.udp'),
                         isc.config.create_answer(
                0, {'Auth': {'queries.udp':0}}))
        self.assertEqual(self.stats.command_show(owner='Auth', name='queries.perzone'),
                         isc.config.create_answer(
                0, {'Auth': {'queries.perzone': [{ 'zonename': 'test1.example',
                      'queries.udp': 1,
                      'queries.tcp': 2 },
                    { 'zonename': 'test2.example',
                      'queries.udp': 3,
                      'queries.tcp': 4 }]}}))
        orig_get_timestamp = stats.get_timestamp
        orig_get_datetime = stats.get_datetime
        stats.get_timestamp = lambda : self.const_timestamp
        stats.get_datetime = lambda : self.const_datetime
        self.assertEqual(stats.get_timestamp(), self.const_timestamp)
        self.assertEqual(stats.get_datetime(), self.const_datetime)
        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.statistics_data['Stats']['timestamp'], self.const_timestamp)
        self.assertEqual(self.stats.statistics_data['Stats']['boot_time'], self.const_default_datetime)
        stats.get_timestamp = orig_get_timestamp
        stats.get_datetime = orig_get_datetime
        self.stats.mccs.specification = isc.config.module_spec.ModuleSpec(
            { "module_name": self.stats.module_name,
              "statistics": [] } )
        self.assertRaises(
            stats.StatsError, self.stats.command_show, owner='Foo', name='bar')

    def test_command_showchema(self):
        (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('Boss' 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['Boss']
        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), 3)
        for item in schema:
            if item['item_type'] == 'list':
                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('Boss' 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('Boss' 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"
                                    }
                                ]
                            }
                        }]}))
        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='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"))

    def test_command_set(self):
        orig_get_datetime = stats.get_datetime
        stats.get_datetime = lambda : self.const_datetime
        (rcode, value) = isc.config.ccsession.parse_answer(
            self.stats.command_set(owner='Boss',
                                   data={ 'boot_time' : self.const_datetime }))
        stats.get_datetime = orig_get_datetime
        self.assertEqual(rcode, 0)
        self.assertTrue(value is None)
        self.assertEqual(self.stats.statistics_data['Boss']['boot_time'],
                         self.const_datetime)
        self.assertEqual(self.stats.statistics_data['Stats']['last_update_time'],
                         self.const_datetime)
        self.assertEqual(self.stats.command_set(owner='Stats',
                                                data={ 'lname' : 'foo@bar' }),
                         isc.config.create_answer(0, None))
        self.stats.statistics_data['Stats'] = {}
        self.stats.mccs.specification = isc.config.module_spec.ModuleSpec(
            { "module_name": self.stats.module_name,
              "statistics": [] } )
        self.assertEqual(self.stats.command_set(owner='Stats',
                                                data={ 'lname' : '_foo_@_bar_' }),
                         isc.config.create_answer(
                1,
                "errors while setting statistics data: unknown item lname"))
        self.stats.statistics_data['Stats'] = {}
        self.stats.mccs.specification = isc.config.module_spec.ModuleSpec(
            { "module_name": self.stats.module_name } )
        self.assertEqual(self.stats.command_set(owner='Stats',
                                                data={ 'lname' : '_foo_@_bar_' }),
                         isc.config.create_answer(
                1,
                "errors while setting statistics data: No statistics specification"))
        self.stats.statistics_data['Stats'] = {}
        self.stats.mccs.specification = isc.config.module_spec.ModuleSpec(
            { "module_name": self.stats.module_name,
              "statistics": [
                    {
                        "item_name": "dummy",
                        "item_type": "string",
                        "item_optional": False,
                        "item_default": "",
                        "item_title": "Local Name",
                        "item_description": "brabra"
                        } ] } )
        self.assertRaises(stats.StatsError,
                          self.stats.command_set, owner='Stats', data={ 'dummy' : '_xxxx_yyyy_zzz_' })

    def test_command_set_withpid(self):
        # one pid of Auth
        retval = isc.config.ccsession.parse_answer(
            self.stats.command_set(owner='Auth',
                                   pid=9997,
                                   data={ 'queries.tcp' : 1001,
                                          'queries.perzone':
                                              [{ 'zonename': 'test1.example',
                                                 'queries.tcp': 1 },
                                               { 'zonename': 'test2.example',
                                                 'queries.tcp': 2,
                                                 'queries.udp': 3 }]}))
        self.assertEqual(retval, (0,None))
        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'], 1001)
        self.assertEqual(self.stats.statistics_data['Auth']['queries.perzone'],
                         [{ 'zonename': 'test1.example',
                            'queries.tcp': 1 },
                          { 'zonename': 'test2.example',
                            'queries.tcp': 2,
                            'queries.udp': 3 }])
        self.assertTrue('Stats' in self.stats.statistics_data)
        self.assertTrue('last_update_time' in self.stats.statistics_data['Stats'])
        self.assertTrue('Auth' in self.stats.statistics_data_bypid)
        self.assertTrue(9997 in self.stats.statistics_data_bypid['Auth'])
        self.assertTrue('queries.tcp' in self.stats.statistics_data_bypid['Auth'][9997])
        self.assertTrue('queries.perzone' in self.stats.statistics_data_bypid['Auth'][9997])
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9997]['queries.tcp'], 1001)
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9997]['queries.perzone'],
                         [{ 'zonename': 'test1.example',
                            'queries.tcp': 1 },
                          { 'zonename': 'test2.example',
                            'queries.tcp': 2,
                            'queries.udp': 3 }])
        # non-existent pid of Auth, but no changes in statistics data
        retval = isc.config.ccsession.parse_answer(
            self.stats.command_set(owner='Auth',
                                   pid=10000,
                                   data={ 'queries.tcp' : 2001,
                                          'queries.perzone':
                                              [{ 'zonename': 'test1.example',
                                                 'queries.tcp': 101 },
                                               { 'zonename': 'test2.example',
                                                 'queries.tcp': 102,
                                                 'queries.udp': 103 }]}))
        self.assertEqual(retval, (0,None))
        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'], 1001)
        self.assertEqual(self.stats.statistics_data['Auth']['queries.perzone'],
                         [{ 'zonename': 'test1.example',
                            'queries.tcp': 1 },
                          { 'zonename': 'test2.example',
                            'queries.tcp': 2,
                            'queries.udp': 3 }])
        self.assertTrue('Auth' in self.stats.statistics_data_bypid)
        self.assertTrue(9997 in self.stats.statistics_data_bypid['Auth'])
        self.assertTrue('queries.tcp' in self.stats.statistics_data_bypid['Auth'][9997])
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9997]['queries.tcp'], 1001)
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9997]['queries.perzone'],
                         [{ 'zonename': 'test1.example',
                            'queries.tcp': 1 },
                          { 'zonename': 'test2.example',
                            'queries.tcp': 2,
                            'queries.udp': 3 }])
        # another pid of Auth
        retval = isc.config.ccsession.parse_answer(
            self.stats.command_set(owner='Auth',
                                   pid=9996,
                                   data={ 'queries.tcp' : 1002,
                                          'queries.udp' : 1003,
                                          'queries.perzone':
                                              [{ 'zonename': 'test1.example',
                                                 'queries.tcp': 10,
                                                 'queries.udp': 11},
                                               { 'zonename': 'test2.example',
                                                 'queries.tcp': 12,
                                                 'queries.udp': 13 }]}))
        self.assertEqual(retval, (0,None))
        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.assertTrue('queries.perzone' in self.stats.statistics_data['Auth'])
        self.assertEqual(self.stats.statistics_data['Auth']['queries.tcp'], 2003)
        self.assertEqual(self.stats.statistics_data['Auth']['queries.udp'], 1003)
        self.assertEqual(self.stats.statistics_data['Auth']['queries.perzone'],
                         [{ 'zonename': 'test1.example',
                            'queries.tcp': 11,
                            'queries.udp': 11},
                          { 'zonename': 'test2.example',
                            'queries.tcp': 14,
                            'queries.udp': 16 }])
        self.assertTrue('Auth' in self.stats.statistics_data_bypid)
        self.assertTrue(9997 in self.stats.statistics_data_bypid['Auth'])
        self.assertTrue(9996 in self.stats.statistics_data_bypid['Auth'])
        self.assertTrue('queries.tcp' in self.stats.statistics_data_bypid['Auth'][9997])
        self.assertTrue('queries.udp' in self.stats.statistics_data_bypid['Auth'][9996])
        self.assertTrue('queries.udp' in self.stats.statistics_data_bypid['Auth'][9996])
        self.assertTrue('queries.perzone' in self.stats.statistics_data_bypid['Auth'][9996])
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9997]['queries.tcp'], 1001)
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9997]['queries.perzone'],
                         [{ 'zonename': 'test1.example',
                            'queries.tcp': 1 },
                          { 'zonename': 'test2.example',
                            'queries.tcp': 2,
                            'queries.udp': 3 }])
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9996]['queries.tcp'], 1002)
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9996]['queries.udp'], 1003)
        self.assertEqual(self.stats.statistics_data_bypid['Auth'][9996]['queries.perzone'],
                         [{ 'zonename': 'test1.example',
                            'queries.tcp': 10,
                            'queries.udp': 11},
                          { 'zonename': 'test2.example',
                            'queries.tcp': 12,
                            'queries.udp': 13 }])

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)

def test_main():
    unittest.main()

if __name__ == "__main__":
    test_main()
