def tearDown(self):
self.counter.clear_counters()
- def test_perzone_counters(self):
- # for per-zone counters
- for counter_name in self._zones_item_list:
- if counter_name.find('time_to_') == 0:
- isc.cc.data.set(\
- self._statistics_data,
- '%s/%s/%s' % (self._perzone_prefix,
- TEST_ZONE_NAME_STR,
- counter_name), 0.0)
- continue
- incrementer = self.counter._to_global\
- ['inc_%s' % counter_name]
- getter = self.counter._to_global\
- ['get_%s' % counter_name]
- incrementer(TEST_ZONE_NAME_STR)
- self.assertEqual(getter(TEST_ZONE_NAME_STR), 1)
- # checks disable/enable
- self.counter.disable()
- incrementer(TEST_ZONE_NAME_STR)
- self.assertEqual(getter(TEST_ZONE_NAME_STR), 1)
- self.counter.enable()
- incrementer(TEST_ZONE_NAME_STR)
- self.assertEqual(getter(TEST_ZONE_NAME_STR), 2)
- for zone_str in (self._entire_server, TEST_ZONE_NAME_STR):
- isc.cc.data.set(\
- self._statistics_data,
- '%s/%s/%s' % (self._perzone_prefix,
- zone_str,
- counter_name), 2)
- # checks other counters
- for counter_name in self._zones_item_list:
- getter = self.counter._to_global\
- ['get_%s' % counter_name]
- self.assertGreaterEqual(getter(TEST_ZONE_NAME_STR), 0)
- self.check_dump_statistics()
-
- def test_xfrrunning_counters(self):
- # for counters of xfer running
- for counter_name in self._xfrrunning_names:
- incrementer = self.counter._to_global\
- ['inc_%s' % counter_name]
- getter = self.counter._to_global\
- ['get_%s' % counter_name]
- decrementer = self.counter._to_global\
- ['dec_%s' % counter_name]
- incrementer()
- self.assertEqual(getter(), 1)
- decrementer()
- self.assertEqual(getter(), 0)
- # checks disable/enable
- self.counter.disable()
- incrementer()
- self.assertEqual(getter(), 0)
- self.counter.enable()
- incrementer()
- self.assertEqual(getter(), 1)
- self.counter.disable()
- decrementer()
- self.assertEqual(getter(), 1)
- self.counter.enable()
- decrementer()
- self.assertEqual(getter(), 0)
- self._statistics_data[counter_name] = 0
- self.check_dump_statistics()
-
def check_dump_statistics(self):
"""Checks no differences between the value returned from
dump_statistics() and locally collected statistics data. Also
self.assertTrue(self._module_spec.validate_statistics(
False, self._statistics_data))
- def test_unixsocket_counters(self):
- # for unixsocket counters
- for counter_name in self._unixsocket_names:
- incrementer = self.counter._to_global\
- ['inc_unixsocket_%s' % counter_name]
- getter = self.counter._to_global\
- ['get_unixsocket_%s' % counter_name]
- incrementer()
- self.assertEqual(getter(), 1)
- # checks disable/enable
- self.counter.disable()
- incrementer()
- self.assertEqual(getter(), 1)
- self.counter.enable()
- incrementer()
- self.assertEqual(getter(), 2)
- isc.cc.data.set(
- self._statistics_data,
- 'socket/unixdomain/%s' % counter_name, 2)
- self.check_dump_statistics()
-
- def test_perzone_timers(self):
- # should not do this test unless there is no timer
- # parameter(time_to_xxx) in statistics spec
- has_timer = False
- for counter_name in self._zones_item_list:
- if counter_name.find('time_to_') == 0:
- has_timer = True
- if not has_timer: return
- # for timer counters
- for counter_name in self._zones_item_list:
- if counter_name.find('time_to_') == -1:
- isc.cc.data.set(\
- self._statistics_data,
- '%s/%s/%s' % (self._perzone_prefix,
- TEST_ZONE_NAME_STR,
- counter_name), 0)
- continue
- starter = self.counter._to_global\
- ['start_%s' % counter_name]
- stopper = self.counter._to_global\
- ['stop_%s' % counter_name]
- getter = self.counter._to_global\
- ['get_%s' % counter_name]
- starter(TEST_ZONE_NAME_STR)
- stopper(TEST_ZONE_NAME_STR)
- self.assertGreater(getter(TEST_ZONE_NAME_STR), 0)
- sec = getter(TEST_ZONE_NAME_STR)
- for zone_str in (self._entire_server, TEST_ZONE_NAME_STR):
- isc.cc.data.set(\
- self._statistics_data,
- '%s/%s/%s' % (self._perzone_prefix,
- zone_str,
- counter_name), sec)
- # twice exec stopper, then second is not changed
- stopper(TEST_ZONE_NAME_STR)
- self.assertEqual(getter(TEST_ZONE_NAME_STR), sec)
- self.check_dump_statistics()
-
- def test_ipsocket_counters(self):
- # for ipsocket counters
- for counter_name in self._ipsocket_names:
- incrementer = self.counter._to_global\
- ['inc_%ssocket_%s' % counter_name]
- getter = self.counter._to_global\
- ['get_%ssocket_%s' % counter_name]
- incrementer()
- self.assertEqual(getter(), 1)
- # checks disable/enable
- self.counter.disable()
- incrementer()
- self.assertEqual(getter(), 1)
- self.counter.enable()
- incrementer()
- self.assertEqual(getter(), 2)
- isc.cc.data.set(
- self._statistics_data,
- 'socket/%s/tcp/%s' % counter_name, 2)
- self.check_dump_statistics()
-
class TestCounter1(unittest.TestCase, BaseTestCounter):
TEST_SPECFILE_LOCATION = TESTDATA_SRCDIR + os.sep + 'test_spec1.spec'
def setUp(self):