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