diff --git a/src/bin/stats/tests/b10-stats-httpd_test.py b/src/bin/stats/tests/b10-stats-httpd_test.py index 6d72dc2f38..ae07aa9f27 100644 --- a/src/bin/stats/tests/b10-stats-httpd_test.py +++ b/src/bin/stats/tests/b10-stats-httpd_test.py @@ -15,145 +15,259 @@ import unittest import os -import http.server -import string -import fake_select import imp -import sys -import fake_socket - -import isc.cc +import socket +import errno +import select +import string +import time +import threading +import http.client +import xml.etree.ElementTree +import isc import stats_httpd -stats_httpd.socket = fake_socket -stats_httpd.select = fake_select +import stats +from test_utils import BaseModules, ThreadingServerManager, MyStats, MyStatsHttpd, TIMEOUT_SEC DUMMY_DATA = { - "auth.queries.tcp": 10000, - "auth.queries.udp": 12000, - "bind10.boot_time": "2011-03-04T11:59:05Z", - "report_time": "2011-03-04T11:59:19Z", - "stats.boot_time": "2011-03-04T11:59:06Z", - "stats.last_update_time": "2011-03-04T11:59:07Z", - "stats.lname": "4d70d40a_c@host", - "stats.start_time": "2011-03-04T11:59:06Z", - "stats.timestamp": 1299239959.560846 + 'Boss' : { + "boot_time": "2011-03-04T11:59:06Z" + }, + 'Auth' : { + "queries.tcp": 2, + "queries.udp": 3 + }, + 'Stats' : { + "report_time": "2011-03-04T11:59:19Z", + "boot_time": "2011-03-04T11:59:06Z", + "last_update_time": "2011-03-04T11:59:07Z", + "lname": "4d70d40a_c@host", + "timestamp": 1299239959.560846 + } } -def push_answer(stats_httpd): - stats_httpd.cc_session.group_sendmsg( - { 'result': - [ 0, DUMMY_DATA ] }, "Stats") - -def pull_query(stats_httpd): - (msg, env) = stats_httpd.cc_session.group_recvmsg() - if 'result' in msg: - (ret, arg) = isc.config.ccsession.parse_answer(msg) - else: - (ret, arg) = isc.config.ccsession.parse_command(msg) - return (ret, arg, env) - class TestHttpHandler(unittest.TestCase): """Tests for HttpHandler class""" def setUp(self): - self.stats_httpd = stats_httpd.StatsHttpd() - self.assertTrue(type(self.stats_httpd.httpd) is list) - self.httpd = self.stats_httpd.httpd + self.base = BaseModules() + self.stats_server = ThreadingServerManager(MyStats) + self.stats = self.stats_server.server + self.stats_server.run() + + def tearDown(self): + self.stats_server.shutdown() + self.base.shutdown() def test_do_GET(self): - for ht in self.httpd: - self._test_do_GET(ht._handler) - - def _test_do_GET(self, handler): + (address, port) = ('127.0.0.1', 65450) + statshttpd_server = ThreadingServerManager(MyStatsHttpd) + self.stats_httpd = statshttpd_server.server + self.stats_httpd.load_config({'listen_on' : [{ 'address': address, 'port' : port }]}) + self.assertTrue(type(self.stats_httpd.httpd) is list) + self.assertEqual(len(self.stats_httpd.httpd), 0) + statshttpd_server.run() + time.sleep(TIMEOUT_SEC*5) + client = http.client.HTTPConnection(address, port) + client._http_vsn_str = 'HTTP/1.0\n' + client.connect() # URL is '/bind10/statistics/xml' - handler.path = stats_httpd.XML_URL_PATH - push_answer(self.stats_httpd) - handler.do_GET() - (ret, arg, env) = pull_query(self.stats_httpd) - self.assertEqual(ret, "show") - self.assertIsNone(arg) - self.assertTrue('group' in env) - self.assertEqual(env['group'], 'Stats') - self.assertEqual(handler.response.code, 200) - self.assertEqual(handler.response.headers["Content-type"], "text/xml") - self.assertTrue(handler.response.headers["Content-Length"] > 0) - self.assertTrue(handler.response.wrote_headers) - self.assertTrue(handler.response.body.find(stats_httpd.XSD_NAMESPACE)>0) - self.assertTrue(handler.response.body.find(stats_httpd.XSD_URL_PATH)>0) - for (k, v) in DUMMY_DATA.items(): - self.assertTrue(handler.response.body.find(str(k))>0) - self.assertTrue(handler.response.body.find(str(v))>0) + client.putrequest('GET', stats_httpd.XML_URL_PATH) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.getheader("Content-type"), "text/xml") + self.assertTrue(int(response.getheader("Content-Length")) > 0) + self.assertEqual(response.status, 200) + root = xml.etree.ElementTree.parse(response).getroot() + self.assertTrue(root.tag.find('stats_data') > 0) + for (k,v) in root.attrib.items(): + if k.find('schemaLocation') > 0: + self.assertEqual(v, stats_httpd.XSD_NAMESPACE + ' ' + stats_httpd.XSD_URL_PATH) + for mod in DUMMY_DATA: + for (item, value) in DUMMY_DATA[mod].items(): + self.assertIsNotNone(root.find(mod + '/' + item)) # URL is '/bind10/statitics/xsd' - handler.path = stats_httpd.XSD_URL_PATH - handler.do_GET() - self.assertEqual(handler.response.code, 200) - self.assertEqual(handler.response.headers["Content-type"], "text/xml") - self.assertTrue(handler.response.headers["Content-Length"] > 0) - self.assertTrue(handler.response.wrote_headers) - self.assertTrue(handler.response.body.find(stats_httpd.XSD_NAMESPACE)>0) - for (k, v) in DUMMY_DATA.items(): - self.assertTrue(handler.response.body.find(str(k))>0) + client.putrequest('GET', stats_httpd.XSD_URL_PATH) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.getheader("Content-type"), "text/xml") + self.assertTrue(int(response.getheader("Content-Length")) > 0) + self.assertEqual(response.status, 200) + root = xml.etree.ElementTree.parse(response).getroot() + url_xmlschema = '{http://www.w3.org/2001/XMLSchema}' + tags = [ url_xmlschema + t for t in [ 'element', 'complexType', 'all', 'element' ] ] + xsdpath = '/'.join(tags) + self.assertTrue(root.tag.find('schema') > 0) + self.assertTrue(hasattr(root, 'attrib')) + self.assertTrue('targetNamespace' in root.attrib) + self.assertEqual(root.attrib['targetNamespace'], + stats_httpd.XSD_NAMESPACE) + for elm in root.findall(xsdpath): + self.assertIsNotNone(elm.attrib['name']) + self.assertTrue(elm.attrib['name'] in DUMMY_DATA) # URL is '/bind10/statitics/xsl' - handler.path = stats_httpd.XSL_URL_PATH - handler.do_GET() - self.assertEqual(handler.response.code, 200) - self.assertEqual(handler.response.headers["Content-type"], "text/xml") - self.assertTrue(handler.response.headers["Content-Length"] > 0) - self.assertTrue(handler.response.wrote_headers) - self.assertTrue(handler.response.body.find(stats_httpd.XSD_NAMESPACE)>0) - for (k, v) in DUMMY_DATA.items(): - self.assertTrue(handler.response.body.find(str(k))>0) + client.putrequest('GET', stats_httpd.XSL_URL_PATH) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.getheader("Content-type"), "text/xml") + self.assertTrue(int(response.getheader("Content-Length")) > 0) + self.assertEqual(response.status, 200) + root = xml.etree.ElementTree.parse(response).getroot() + url_trans = '{http://www.w3.org/1999/XSL/Transform}' + url_xhtml = '{http://www.w3.org/1999/xhtml}' + xslpath = url_trans + 'template/' + url_xhtml + 'tr' + self.assertEqual(root.tag, url_trans + 'stylesheet') + for tr in root.findall(xslpath): + tds = tr.findall(url_xhtml + 'td') + self.assertIsNotNone(tds) + self.assertEqual(type(tds), list) + self.assertTrue(len(tds) > 2) + self.assertTrue(hasattr(tds[0], 'text')) + self.assertTrue(tds[0].text in DUMMY_DATA) + valueof = tds[2].find(url_trans + 'value-of') + self.assertIsNotNone(valueof) + self.assertTrue(hasattr(valueof, 'attrib')) + self.assertIsNotNone(valueof.attrib) + self.assertTrue('select' in valueof.attrib) + self.assertTrue(valueof.attrib['select'] in \ + [ tds[0].text+'/'+item for item in DUMMY_DATA[tds[0].text].keys() ]) # 302 redirect - handler.path = '/' - handler.headers = {'Host': 'my.host.domain'} - handler.do_GET() - self.assertEqual(handler.response.code, 302) - self.assertEqual(handler.response.headers["Location"], - "http://my.host.domain%s" % stats_httpd.XML_URL_PATH) + client._http_vsn_str = 'HTTP/1.1' + client.putrequest('GET', '/') + client.putheader('Host', address) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.status, 302) + self.assertEqual(response.getheader('Location'), + "http://%s:%d%s" % (address, port, stats_httpd.XML_URL_PATH)) - # 404 NotFound - handler.path = '/path/to/foo/bar' - handler.headers = {} - handler.do_GET() - self.assertEqual(handler.response.code, 404) + # # 404 NotFound + client._http_vsn_str = 'HTTP/1.0' + client.putrequest('GET', '/path/to/foo/bar') + client.endheaders() + response = client.getresponse() + self.assertEqual(response.status, 404) + client.close() + statshttpd_server.shutdown() + + def test_do_GET_failed1(self): # failure case(connection with Stats is down) - handler.path = stats_httpd.XML_URL_PATH - push_answer(self.stats_httpd) - self.assertFalse(self.stats_httpd.cc_session._socket._closed) - self.stats_httpd.cc_session._socket._closed = True - handler.do_GET() - self.stats_httpd.cc_session._socket._closed = False - self.assertEqual(handler.response.code, 500) - self.stats_httpd.cc_session._clear_queues() + (address, port) = ('127.0.0.1', 65451) + statshttpd_server = ThreadingServerManager(MyStatsHttpd) + statshttpd = statshttpd_server.server + statshttpd.load_config({'listen_on' : [{ 'address': address, 'port' : port }]}) + statshttpd_server.run() + self.assertTrue(self.stats_server.server.running) + self.stats_server.shutdown() + time.sleep(TIMEOUT_SEC*2) + self.assertFalse(self.stats_server.server.running) + statshttpd.cc_session.set_timeout(milliseconds=TIMEOUT_SEC/1000) + client = http.client.HTTPConnection(address, port) + client.connect() - # failure case(Stats module returns err) - handler.path = stats_httpd.XML_URL_PATH - self.stats_httpd.cc_session.group_sendmsg( - { 'result': [ 1, "I have an error." ] }, "Stats") - self.assertFalse(self.stats_httpd.cc_session._socket._closed) - self.stats_httpd.cc_session._socket._closed = False - handler.do_GET() - self.assertEqual(handler.response.code, 500) - self.stats_httpd.cc_session._clear_queues() + # request XML + client.putrequest('GET', stats_httpd.XML_URL_PATH) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.status, 500) + + # request XSD + client.putrequest('GET', stats_httpd.XSD_URL_PATH) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.status, 500) + + # request XSL + client.putrequest('GET', stats_httpd.XSL_URL_PATH) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.status, 500) + + client.close() + statshttpd_server.shutdown() + + def test_do_GET_failed2(self): + # failure case(connection with Stats is down) + (address, port) = ('127.0.0.1', 65452) + statshttpd_server = ThreadingServerManager(MyStatsHttpd) + self.stats_httpd = statshttpd_server.server + self.stats_httpd.load_config({'listen_on' : [{ 'address': address, 'port' : port }]}) + statshttpd_server.run() + self.stats.mccs.set_command_handler( + lambda cmd, args: \ + isc.config.ccsession.create_answer(1, "I have an error.") + ) + time.sleep(TIMEOUT_SEC*5) + client = http.client.HTTPConnection(address, port) + client.connect() + + # request XML + client.putrequest('GET', stats_httpd.XML_URL_PATH) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.status, 500) + + # request XSD + client.putrequest('GET', stats_httpd.XSD_URL_PATH) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.status, 500) + + # request XSL + client.putrequest('GET', stats_httpd.XSL_URL_PATH) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.status, 500) + + client.close() + statshttpd_server.shutdown() def test_do_HEAD(self): - for ht in self.httpd: - self._test_do_HEAD(ht._handler) + (address, port) = ('127.0.0.1', 65453) + statshttpd_server = ThreadingServerManager(MyStatsHttpd) + self.stats_httpd = statshttpd_server.server + self.stats_httpd.load_config({'listen_on' : [{ 'address': address, 'port' : port }]}) + statshttpd_server.run() + time.sleep(TIMEOUT_SEC*5) + client = http.client.HTTPConnection(address, port) + client.connect() + client.putrequest('HEAD', stats_httpd.XML_URL_PATH) + client.endheaders() + response = client.getresponse() + self.assertEqual(response.status, 200) - def _test_do_HEAD(self, handler): - handler.path = '/path/to/foo/bar' - handler.do_HEAD() - self.assertEqual(handler.response.code, 404) + client.putrequest('HEAD', '/path/to/foo/bar') + client.endheaders() + response = client.getresponse() + self.assertEqual(response.status, 404) + client.close() + statshttpd_server.shutdown() + + def test_log_message(self): + class MyHttpHandler(stats_httpd.HttpHandler): + def __init__(self): + class _Dummy_class_(): pass + self.address_string = lambda : 'dummyhost' + self.log_date_time_string = lambda : \ + 'DD/MM/YYYY HH:MI:SS' + self.server = _Dummy_class_() + self.server.log_writer = self.log_writer + def log_writer(self, line): + self.logged_line = line + self.handler = MyHttpHandler() + self.handler.log_message("%s %d", 'ABCDEFG', 12345) + self.assertEqual(self.handler.logged_line, + "[b10-stats-httpd] dummyhost - - " + + "[DD/MM/YYYY HH:MI:SS] ABCDEFG 12345\n") class TestHttpServerError(unittest.TestCase): """Tests for HttpServerError exception""" - def test_raises(self): try: raise stats_httpd.HttpServerError('Nothing') @@ -162,17 +276,16 @@ class TestHttpServerError(unittest.TestCase): class TestHttpServer(unittest.TestCase): """Tests for HttpServer class""" + def setUp(self): + self.base = BaseModules() + + def tearDown(self): + self.base.shutdown() def test_httpserver(self): - self.stats_httpd = stats_httpd.StatsHttpd() - for ht in self.stats_httpd.httpd: - self.assertTrue(ht.server_address in self.stats_httpd.http_addrs) - self.assertEqual(ht.xml_handler, self.stats_httpd.xml_handler) - self.assertEqual(ht.xsd_handler, self.stats_httpd.xsd_handler) - self.assertEqual(ht.xsl_handler, self.stats_httpd.xsl_handler) - self.assertEqual(ht.log_writer, self.stats_httpd.write_log) - self.assertTrue(isinstance(ht._handler, stats_httpd.HttpHandler)) - self.assertTrue(isinstance(ht.socket, fake_socket.socket)) + statshttpd = stats_httpd.StatsHttpd() + self.assertEqual(type(statshttpd.httpd), list) + self.assertEqual(len(statshttpd.httpd), 0) class TestStatsHttpdError(unittest.TestCase): """Tests for StatsHttpdError exception""" @@ -187,130 +300,176 @@ class TestStatsHttpd(unittest.TestCase): """Tests for StatsHttpd class""" def setUp(self): - fake_socket._CLOSED = False - fake_socket.has_ipv6 = True + self.base = BaseModules() + self.stats_server = ThreadingServerManager(MyStats) + self.stats = self.stats_server.server + self.stats_server.run() self.stats_httpd = stats_httpd.StatsHttpd() + # checking IPv6 enabled on this platform + self.ipv6_enabled = True + try: + sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) + sock.bind(("::1",8000)) + sock.close() + except socket.error: + self.ipv6_enabled = False + def tearDown(self): self.stats_httpd.stop() + self.stats_server.shutdown() + self.base.shutdown() def test_init(self): - self.assertFalse(self.stats_httpd.mccs.get_socket()._closed) - self.assertEqual(self.stats_httpd.mccs.get_socket().fileno(), - id(self.stats_httpd.mccs.get_socket())) - for ht in self.stats_httpd.httpd: - self.assertFalse(ht.socket._closed) - self.assertEqual(ht.socket.fileno(), id(ht.socket)) - fake_socket._CLOSED = True - self.assertRaises(isc.cc.session.SessionError, - stats_httpd.StatsHttpd) - fake_socket._CLOSED = False + self.assertEqual(self.stats_httpd.running, False) + self.assertEqual(self.stats_httpd.poll_intval, 0.5) + self.assertEqual(self.stats_httpd.httpd, []) + self.assertEqual(type(self.stats_httpd.mccs), isc.config.ModuleCCSession) + self.assertEqual(type(self.stats_httpd.cc_session), isc.cc.Session) + self.assertEqual(len(self.stats_httpd.config), 2) + self.assertTrue('listen_on' in self.stats_httpd.config) + self.assertEqual(len(self.stats_httpd.config['listen_on']), 1) + self.assertTrue('address' in self.stats_httpd.config['listen_on'][0]) + self.assertTrue('port' in self.stats_httpd.config['listen_on'][0]) + self.assertTrue(('127.0.0.1', 8000) in set(self.stats_httpd.http_addrs)) + + def test_openclose_mccs(self): + statshttpd = stats_httpd.StatsHttpd() + statshttpd.close_mccs() + self.assertEqual(statshttpd.mccs, None) + statshttpd.open_mccs() + self.assertIsNotNone(statshttpd.mccs) + statshttpd.mccs = None + self.assertEqual(statshttpd.mccs, None) + self.assertEqual(statshttpd.close_mccs(), None) def test_mccs(self): - self.stats_httpd.open_mccs() + self.assertIsNotNone(self.stats_httpd.mccs.get_socket()) self.assertTrue( - isinstance(self.stats_httpd.mccs.get_socket(), fake_socket.socket)) + isinstance(self.stats_httpd.mccs.get_socket(), socket.socket)) self.assertTrue( isinstance(self.stats_httpd.cc_session, isc.cc.session.Session)) - self.assertTrue( - isinstance(self.stats_httpd.stats_module_spec, isc.config.ModuleSpec)) - for cfg in self.stats_httpd.stats_config_spec: - self.assertTrue('item_name' in cfg) - self.assertTrue(cfg['item_name'] in DUMMY_DATA) - self.assertTrue(len(self.stats_httpd.stats_config_spec), len(DUMMY_DATA)) - - def test_load_config(self): - self.stats_httpd.load_config() - self.assertTrue(('127.0.0.1', 8000) in set(self.stats_httpd.http_addrs)) + self.statistics_spec = self.stats_httpd.get_stats_spec() + for mod in DUMMY_DATA: + self.assertTrue(mod in self.statistics_spec) + for cfg in self.statistics_spec[mod]: + self.assertTrue('item_name' in cfg) + self.assertTrue(cfg['item_name'] in DUMMY_DATA[mod]) + self.assertTrue(len(self.statistics_spec[mod]), len(DUMMY_DATA[mod])) + self.stats_httpd.close_mccs() + self.assertIsNone(self.stats_httpd.mccs) def test_httpd(self): # dual stack (addresses is ipv4 and ipv6) - fake_socket.has_ipv6 = True - self.assertTrue(('127.0.0.1', 8000) in set(self.stats_httpd.http_addrs)) - self.stats_httpd.http_addrs = [ ('::1', 8000), ('127.0.0.1', 8000) ] - self.assertTrue( - stats_httpd.HttpServer.address_family in set([fake_socket.AF_INET, fake_socket.AF_INET6])) - self.stats_httpd.open_httpd() - for ht in self.stats_httpd.httpd: - self.assertTrue(isinstance(ht.socket, fake_socket.socket)) - self.stats_httpd.close_httpd() + if self.ipv6_enabled: + self.assertTrue(('127.0.0.1', 8000) in set(self.stats_httpd.http_addrs)) + self.stats_httpd.http_addrs = [ ('::1', 8000), ('127.0.0.1', 8000) ] + self.assertTrue( + stats_httpd.HttpServer.address_family in set([socket.AF_INET, socket.AF_INET6])) + self.stats_httpd.open_httpd() + for ht in self.stats_httpd.httpd: + self.assertTrue(isinstance(ht.socket, socket.socket)) + self.stats_httpd.close_httpd() # dual stack (address is ipv6) - fake_socket.has_ipv6 = True - self.stats_httpd.http_addrs = [ ('::1', 8000) ] - self.stats_httpd.open_httpd() - for ht in self.stats_httpd.httpd: - self.assertTrue(isinstance(ht.socket, fake_socket.socket)) - self.stats_httpd.close_httpd() - + if self.ipv6_enabled: + self.stats_httpd.http_addrs = [ ('::1', 8000) ] + self.stats_httpd.open_httpd() + for ht in self.stats_httpd.httpd: + self.assertTrue(isinstance(ht.socket, socket.socket)) + self.stats_httpd.close_httpd() + # dual stack (address is ipv4) - fake_socket.has_ipv6 = True - self.stats_httpd.http_addrs = [ ('127.0.0.1', 8000) ] - self.stats_httpd.open_httpd() - for ht in self.stats_httpd.httpd: - self.assertTrue(isinstance(ht.socket, fake_socket.socket)) - self.stats_httpd.close_httpd() + if self.ipv6_enabled: + self.stats_httpd.http_addrs = [ ('127.0.0.1', 8000) ] + self.stats_httpd.open_httpd() + for ht in self.stats_httpd.httpd: + self.assertTrue(isinstance(ht.socket, socket.socket)) + self.stats_httpd.close_httpd() # only-ipv4 single stack - fake_socket.has_ipv6 = False - self.stats_httpd.http_addrs = [ ('127.0.0.1', 8000) ] - self.stats_httpd.open_httpd() - for ht in self.stats_httpd.httpd: - self.assertTrue(isinstance(ht.socket, fake_socket.socket)) - self.stats_httpd.close_httpd() - + if not self.ipv6_enabled: + self.stats_httpd.http_addrs = [ ('127.0.0.1', 8000) ] + self.stats_httpd.open_httpd() + for ht in self.stats_httpd.httpd: + self.assertTrue(isinstance(ht.socket, socket.socket)) + self.stats_httpd.close_httpd() + # only-ipv4 single stack (force set ipv6 ) - fake_socket.has_ipv6 = False - self.stats_httpd.http_addrs = [ ('::1', 8000) ] - self.assertRaises(stats_httpd.HttpServerError, - self.stats_httpd.open_httpd) - + if not self.ipv6_enabled: + self.stats_httpd.http_addrs = [ ('::1', 8000) ] + self.assertRaises(stats_httpd.HttpServerError, + self.stats_httpd.open_httpd) + # hostname self.stats_httpd.http_addrs = [ ('localhost', 8000) ] self.stats_httpd.open_httpd() for ht in self.stats_httpd.httpd: - self.assertTrue(isinstance(ht.socket, fake_socket.socket)) + self.assertTrue(isinstance(ht.socket, socket.socket)) self.stats_httpd.close_httpd() - + self.stats_httpd.http_addrs = [ ('my.host.domain', 8000) ] - self.stats_httpd.open_httpd() - for ht in self.stats_httpd.httpd: - self.assertTrue(isinstance(ht.socket, fake_socket.socket)) + self.assertRaises(stats_httpd.HttpServerError, self.stats_httpd.open_httpd) + self.assertEqual(type(self.stats_httpd.httpd), list) + self.assertEqual(len(self.stats_httpd.httpd), 0) self.stats_httpd.close_httpd() # over flow of port number self.stats_httpd.http_addrs = [ ('', 80000) ] self.assertRaises(stats_httpd.HttpServerError, self.stats_httpd.open_httpd) + # negative self.stats_httpd.http_addrs = [ ('', -8000) ] self.assertRaises(stats_httpd.HttpServerError, self.stats_httpd.open_httpd) + # alphabet self.stats_httpd.http_addrs = [ ('', 'ABCDE') ] self.assertRaises(stats_httpd.HttpServerError, self.stats_httpd.open_httpd) - def test_start(self): - self.stats_httpd.cc_session.group_sendmsg( - { 'command': [ "shutdown" ] }, "StatsHttpd") - self.stats_httpd.start() - self.stats_httpd = stats_httpd.StatsHttpd() - self.assertRaises( - fake_select.error, self.stats_httpd.start) + # Address already in use + self.statshttpd_server = ThreadingServerManager(MyStatsHttpd) + self.statshttpd_server.server.load_config({'listen_on' : [{ 'address': '127.0.0.1', 'port' : 65454 }]}) + self.statshttpd_server.run() + time.sleep(TIMEOUT_SEC) + self.stats_httpd.load_config({'listen_on' : [{ 'address': '127.0.0.1', 'port' : 65454 }]}) + self.assertRaises(stats_httpd.HttpServerError, self.stats_httpd.open_httpd) + self.statshttpd_server.shutdown() - def test_stop(self): - # success case - fake_socket._CLOSED = False - self.stats_httpd.stop() + def test_running(self): self.assertFalse(self.stats_httpd.running) - self.assertIsNone(self.stats_httpd.mccs) - for ht in self.stats_httpd.httpd: - self.assertTrue(ht.socket._closed) - self.assertTrue(self.stats_httpd.cc_session._socket._closed) + self.statshttpd_server = ThreadingServerManager(MyStatsHttpd) + self.stats_httpd = self.statshttpd_server.server + self.stats_httpd.load_config({'listen_on' : [{ 'address': '127.0.0.1', 'port' : 65455 }]}) + self.statshttpd_server.run() + time.sleep(TIMEOUT_SEC*2) + self.assertTrue(self.stats_httpd.running) + self.statshttpd_server.shutdown() + self.assertFalse(self.stats_httpd.running) + # failure case - self.stats_httpd.cc_session._socket._closed = False - self.stats_httpd.open_mccs() - self.stats_httpd.cc_session._socket._closed = True - self.stats_httpd.stop() # No excetion raises - self.stats_httpd.cc_session._socket._closed = False + self.stats_httpd = stats_httpd.StatsHttpd() + self.stats_httpd.cc_session.close() + self.assertRaises( + isc.cc.session.SessionError, self.stats_httpd.start) + + def test_select_failure(self): + def raise_select_except(*args): + raise select.error('dummy error') + def raise_select_except_with_errno(*args): + raise select.error(errno.EINTR) + (address, port) = ('127.0.0.1', 65456) + stats_httpd.select.select = raise_select_except + statshttpd = stats_httpd.StatsHttpd() + statshttpd.load_config({'listen_on' : [{ 'address': address, 'port' : port }]}) + self.assertRaises(select.error, statshttpd.start) + statshttpd.stop() + stats_httpd.select.select = raise_select_except_with_errno + statshttpd_server = ThreadingServerManager(MyStatsHttpd) + statshttpd = statshttpd_server.server + statshttpd.load_config({'listen_on' : [{ 'address': address, 'port' : port }]}) + statshttpd_server.run() + time.sleep(TIMEOUT_SEC*2) + statshttpd_server.shutdown() def test_open_template(self): # successful conditions @@ -363,38 +522,40 @@ class TestStatsHttpd(unittest.TestCase): self.assertEqual( self.stats_httpd.config_handler(dict(_UNKNOWN_KEY_=None)), isc.config.ccsession.create_answer( - 1, "Unknown known config: _UNKNOWN_KEY_")) + 1, "Unknown known config: _UNKNOWN_KEY_")) + self.assertEqual( self.stats_httpd.config_handler( - dict(listen_on=[dict(address="::2",port=8000)])), + dict(listen_on=[dict(address="127.0.0.2",port=8000)])), isc.config.ccsession.create_answer(0)) self.assertTrue("listen_on" in self.stats_httpd.config) for addr in self.stats_httpd.config["listen_on"]: self.assertTrue("address" in addr) self.assertTrue("port" in addr) - self.assertTrue(addr["address"] == "::2") + self.assertTrue(addr["address"] == "127.0.0.2") self.assertTrue(addr["port"] == 8000) - self.assertEqual( - self.stats_httpd.config_handler( - dict(listen_on=[dict(address="::1",port=80)])), - isc.config.ccsession.create_answer(0)) - self.assertTrue("listen_on" in self.stats_httpd.config) - for addr in self.stats_httpd.config["listen_on"]: - self.assertTrue("address" in addr) - self.assertTrue("port" in addr) - self.assertTrue(addr["address"] == "::1") - self.assertTrue(addr["port"] == 80) + if self.ipv6_enabled: + self.assertEqual( + self.stats_httpd.config_handler( + dict(listen_on=[dict(address="::1",port=8000)])), + isc.config.ccsession.create_answer(0)) + self.assertTrue("listen_on" in self.stats_httpd.config) + for addr in self.stats_httpd.config["listen_on"]: + self.assertTrue("address" in addr) + self.assertTrue("port" in addr) + self.assertTrue(addr["address"] == "::1") + self.assertTrue(addr["port"] == 8000) self.assertEqual( self.stats_httpd.config_handler( - dict(listen_on=[dict(address="1.2.3.4",port=54321)])), + dict(listen_on=[dict(address="127.0.0.1",port=54321)])), isc.config.ccsession.create_answer(0)) self.assertTrue("listen_on" in self.stats_httpd.config) for addr in self.stats_httpd.config["listen_on"]: self.assertTrue("address" in addr) self.assertTrue("port" in addr) - self.assertTrue(addr["address"] == "1.2.3.4") + self.assertTrue(addr["address"] == "127.0.0.1") self.assertTrue(addr["port"] == 54321) (ret, arg) = isc.config.ccsession.parse_answer( self.stats_httpd.config_handler( @@ -500,8 +661,6 @@ class TestStatsHttpd(unittest.TestCase): imp.reload(stats_httpd) os.environ["B10_FROM_SOURCE"] = tmppath imp.reload(stats_httpd) - stats_httpd.socket = fake_socket - stats_httpd.select = fake_select if __name__ == "__main__": unittest.main() diff --git a/src/bin/stats/tests/b10-stats_test.py b/src/bin/stats/tests/b10-stats_test.py index 2fb4ab5e50..b013c7a8bc 100644 --- a/src/bin/stats/tests/b10-stats_test.py +++ b/src/bin/stats/tests/b10-stats_test.py @@ -13,633 +13,496 @@ # NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION # WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. -# -# Tests for the stats module -# -import os -import sys -import time import unittest +import os +import threading +import io +import time import imp -from isc.cc.session import Session, SessionError -from isc.config.ccsession import ModuleCCSession, ModuleCCSessionError -from fake_time import time, strftime, gmtime -import stats -stats.time = time -stats.strftime = strftime -stats.gmtime = gmtime -from stats import SessionSubject, CCSessionListener, get_timestamp, get_datetime -from fake_time import _TEST_TIME_SECS, _TEST_TIME_STRF -if "B10_FROM_SOURCE" in os.environ: - TEST_SPECFILE_LOCATION = os.environ["B10_FROM_SOURCE"] +\ - "/src/bin/stats/tests/testdata/stats_test.spec" -else: - TEST_SPECFILE_LOCATION = "./testdata/stats_test.spec" +import stats +import isc.cc.session +from test_utils import BaseModules, ThreadingServerManager, MyStats, send_command, TIMEOUT_SEC + +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': 'one', 'item_type': 'integer' }, + { 'item_name': 'two', 'item_type': 'integer' }, + { 'item_name': 'three', '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': 'one', 'item_type': 'integer' }, + { 'item_name': 'two', 'item_type': 'integer' }, + { 'item_name': 'three', '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' } + ] + + def test_parse_spec(self): + self.assertEqual( + stats.parse_spec(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,0,0], + 'test_map2' : { 'A' : 0, 'B' : 0, 'C' : 0 }, + 'test_none' : None }) + self.assertRaises(TypeError, stats.parse_spec, None) + self.assertRaises(KeyError, stats.parse_spec, [{'item_name':'Foo'}]) + + def test_get_timestamp(self): + self.assertEqual(stats.get_timestamp(), 1308730448.965706) + + def test_get_datetime(self): + stats.time = lambda : 1308730448.965706 + stats.gmtime = lambda : (2011, 6, 22, 8, 14, 8, 2, 173, 0) + self.assertEqual(stats.get_datetime(), '2011-06-22T08:14:08Z') + self.assertNotEqual(stats.get_datetime( + (2011, 6, 22, 8, 23, 40, 2, 173, 0)), '2011-06-22T08:14:08Z') + +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 tearDown(self): + pass + + 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): - self.session = Session() - self.subject = SessionSubject(session=self.session) - self.listener = CCSessionListener(self.subject) - self.stats_spec = self.listener.cc_session.get_module_spec().get_config_spec() - self.module_name = self.listener.cc_session.get_module_spec().get_module_name() - self.stats_data = { - 'report_time' : get_datetime(), - 'bind10.boot_time' : "1970-01-01T00:00:00Z", - 'stats.timestamp' : get_timestamp(), - 'stats.lname' : self.session.lname, - 'auth.queries.tcp': 0, - 'auth.queries.udp': 0, - "stats.boot_time": get_datetime(), - "stats.start_time": get_datetime(), - "stats.last_update_time": get_datetime() - } - # check starting - self.assertFalse(self.subject.running) - self.subject.start() - self.assertEqual(len(self.session.old_message_queue), 1) - self.assertTrue(self.subject.running) - self.assertEqual(len(self.session.message_queue), 0) - self.assertEqual(self.module_name, 'Stats') + self.base = BaseModules() + self.stats = stats.Stats() + 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") def tearDown(self): - # check closing - self.subject.stop() - self.assertFalse(self.subject.running) - self.subject.detach(self.listener) - self.listener.stop() - self.session.close() + self.base.shutdown() - def test_local_func(self): - """ - Test for local function - - """ - # test for result_ok - self.assertEqual(type(result_ok()), dict) - self.assertEqual(result_ok(), {'result': [0]}) - self.assertEqual(result_ok(1), {'result': [1]}) - self.assertEqual(result_ok(0,'OK'), {'result': [0, 'OK']}) - self.assertEqual(result_ok(1,'Not good'), {'result': [1, 'Not good']}) - self.assertEqual(result_ok(None,"It's None"), {'result': [None, "It's None"]}) - self.assertNotEqual(result_ok(), {'RESULT': [0]}) + 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) - # test for get_timestamp - self.assertEqual(get_timestamp(), _TEST_TIME_SECS) + 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 - # test for get_datetime - self.assertEqual(get_datetime(), _TEST_TIME_STRF) + def test_start(self): + statsserver = ThreadingServerManager(MyStats) + stats = statsserver.server + self.assertFalse(stats.running) + statsserver.run() + time.sleep(TIMEOUT_SEC) + self.assertTrue(stats.running) + statsserver.shutdown() + self.assertFalse(stats.running) - def test_show_command(self): - """ - Test for show command - - """ - # test show command without arg - self.session.group_sendmsg({"command": [ "show", None ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - # ignore under 0.9 seconds - self.assertEqual(result_ok(0, self.stats_data), result_data) - self.assertEqual(len(self.session.message_queue), 0) + def test_start_with_err(self): + statsd = stats.Stats() + statsd.update_statistics_data = lambda x,**y: [1] + self.assertRaises(stats.StatsError, statsd.start) - # test show command with arg - self.session.group_sendmsg({"command": [ "show", {"stats_item_name": "stats.lname"}]}, "Stats") - self.assertEqual(len(self.subject.session.message_queue), 1) - self.subject.check() - result_data = self.subject.session.get_message("Stats", None) - self.assertEqual(result_ok(0, {'stats.lname': self.stats_data['stats.lname']}), - result_data) - self.assertEqual(len(self.subject.session.message_queue), 0) + def test_config_handler(self): + self.assertEqual(self.stats.config_handler({'foo':'bar'}), + isc.config.create_answer(0)) - # test show command with arg which has wrong name - self.session.group_sendmsg({"command": [ "show", {"stats_item_name": "stats.dummy"}]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - # ignore under 0.9 seconds - self.assertEqual(result_ok(0, self.stats_data), result_data) - self.assertEqual(len(self.session.message_queue), 0) - - def test_set_command(self): - """ - Test for set command - - """ - # test set command - self.stats_data['auth.queries.udp'] = 54321 - self.assertEqual(self.stats_data['auth.queries.udp'], 54321) - self.assertEqual(self.stats_data['auth.queries.tcp'], 0) - self.session.group_sendmsg({ "command": [ - "set", { - 'stats_data': {'auth.queries.udp': 54321 } - } ] }, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # test show command - self.session.group_sendmsg({"command": [ "show", None ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, self.stats_data), result_data) - self.assertEqual(len(self.session.message_queue), 0) - - # test set command 2 - self.stats_data['auth.queries.udp'] = 0 - self.assertEqual(self.stats_data['auth.queries.udp'], 0) - self.assertEqual(self.stats_data['auth.queries.tcp'], 0) - self.session.group_sendmsg({ "command": [ "set", {'stats_data': {'auth.queries.udp': 0}} ]}, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # test show command 2 - self.session.group_sendmsg({"command": [ "show", None ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, self.stats_data), result_data) - self.assertEqual(len(self.session.message_queue), 0) - - # test set command 3 - self.stats_data['auth.queries.tcp'] = 54322 - self.assertEqual(self.stats_data['auth.queries.udp'], 0) - self.assertEqual(self.stats_data['auth.queries.tcp'], 54322) - self.session.group_sendmsg({ "command": [ - "set", { - 'stats_data': {'auth.queries.tcp': 54322 } - } ] }, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # test show command 3 - self.session.group_sendmsg({"command": [ "show", None ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, self.stats_data), result_data) - self.assertEqual(len(self.session.message_queue), 0) - - def test_remove_command(self): - """ - Test for remove command - - """ - self.session.group_sendmsg({"command": - [ "remove", {"stats_item_name": 'bind10.boot_time' }]}, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - self.assertEqual(self.stats_data.pop('bind10.boot_time'), "1970-01-01T00:00:00Z") - self.assertFalse('bind10.boot_time' in self.stats_data) - - # test show command with arg - self.session.group_sendmsg({"command": - [ "show", {"stats_item_name": 'bind10.boot_time'}]}, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertFalse('bind10.boot_time' in result_data['result'][1]) - self.assertEqual(result_ok(0, self.stats_data), result_data) - self.assertEqual(len(self.session.message_queue), 0) - - def test_reset_command(self): - """ - Test for reset command - - """ - self.session.group_sendmsg({"command": [ "reset" ] }, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # test show command - self.session.group_sendmsg({"command": [ "show" ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, self.stats_data), result_data) - self.assertEqual(len(self.session.message_queue), 0) - - def test_status_command(self): - """ - Test for status command - - """ - self.session.group_sendmsg({"command": [ "status" ] }, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(0, "I'm alive."), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - def test_unknown_command(self): - """ - Test for unknown command - - """ - self.session.group_sendmsg({"command": [ "hoge", None ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(1, "Unknown command: 'hoge'"), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - def test_shutdown_command(self): - """ - Test for shutdown command - - """ - self.session.group_sendmsg({"command": [ "shutdown", None ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.assertTrue(self.subject.running) - self.subject.check() - self.assertFalse(self.subject.running) - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - - def test_some_commands(self): - """ - Test for some commands in a row - - """ - # test set command - self.stats_data['bind10.boot_time'] = '2010-08-02T14:47:56Z' - self.assertEqual(self.stats_data['bind10.boot_time'], '2010-08-02T14:47:56Z') - self.session.group_sendmsg({ "command": [ - "set", { - 'stats_data': {'bind10.boot_time': '2010-08-02T14:47:56Z' } - }]}, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # check its value - self.session.group_sendmsg({ "command": [ - "show", { 'stats_item_name': 'bind10.boot_time' } - ] }, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, {'bind10.boot_time': '2010-08-02T14:47:56Z'}), - result_data) - self.assertEqual(result_ok(0, {'bind10.boot_time': self.stats_data['bind10.boot_time']}), - result_data) - self.assertEqual(len(self.session.message_queue), 0) - - # test set command 2nd - self.stats_data['auth.queries.udp'] = 98765 - self.assertEqual(self.stats_data['auth.queries.udp'], 98765) - self.session.group_sendmsg({ "command": [ - "set", { 'stats_data': { - 'auth.queries.udp': - self.stats_data['auth.queries.udp'] - } } - ] }, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # check its value - self.session.group_sendmsg({"command": [ - "show", {'stats_item_name': 'auth.queries.udp'} - ] }, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, {'auth.queries.udp': 98765}), - result_data) - self.assertEqual(result_ok(0, {'auth.queries.udp': self.stats_data['auth.queries.udp']}), - result_data) - self.assertEqual(len(self.session.message_queue), 0) - - # test set command 3 - self.stats_data['auth.queries.tcp'] = 4321 - self.session.group_sendmsg({"command": [ - "set", - {'stats_data': {'auth.queries.tcp': 4321 }} ]}, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # check value - self.session.group_sendmsg({"command": [ "show", {'stats_item_name': 'auth.queries.tcp'} ]}, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, {'auth.queries.tcp': 4321}), - result_data) - self.assertEqual(result_ok(0, {'auth.queries.tcp': self.stats_data['auth.queries.tcp']}), - result_data) - self.assertEqual(len(self.session.message_queue), 0) - - self.session.group_sendmsg({"command": [ "show", {'stats_item_name': 'auth.queries.udp'} ]}, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, {'auth.queries.udp': 98765}), - result_data) - self.assertEqual(result_ok(0, {'auth.queries.udp': self.stats_data['auth.queries.udp']}), - result_data) - self.assertEqual(len(self.session.message_queue), 0) - - # test set command 4 - self.stats_data['auth.queries.tcp'] = 67890 - self.session.group_sendmsg({"command": [ - "set", {'stats_data': {'auth.queries.tcp': 67890 }} ]}, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # test show command for all values - self.session.group_sendmsg({"command": [ "show", None ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, self.stats_data), result_data) - self.assertEqual(len(self.session.message_queue), 0) - - def test_some_commands2(self): - """ - Test for some commands in a row using list-type value - - """ - self.stats_data['listtype'] = [1, 2, 3] - self.assertEqual(self.stats_data['listtype'], [1, 2, 3]) - self.session.group_sendmsg({ "command": [ - "set", {'stats_data': {'listtype': [1, 2, 3] }} - ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # check its value - self.session.group_sendmsg({ "command": [ - "show", { 'stats_item_name': 'listtype'} - ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, {'listtype': [1, 2, 3]}), - result_data) - self.assertEqual(result_ok(0, {'listtype': self.stats_data['listtype']}), - result_data) - self.assertEqual(len(self.session.message_queue), 0) - - # test set list-type value - self.assertEqual(self.stats_data['listtype'], [1, 2, 3]) - self.session.group_sendmsg({"command": [ - "set", {'stats_data': {'listtype': [3, 2, 1, 0] }} - ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # check its value - self.session.group_sendmsg({ "command": [ - "show", { 'stats_item_name': 'listtype' } - ] }, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, {'listtype': [3, 2, 1, 0]}), - result_data) - self.assertEqual(len(self.session.message_queue), 0) - - def test_some_commands3(self): - """ - Test for some commands in a row using dictionary-type value - - """ - self.stats_data['dicttype'] = {"a": 1, "b": 2, "c": 3} - self.assertEqual(self.stats_data['dicttype'], {"a": 1, "b": 2, "c": 3}) - self.session.group_sendmsg({ "command": [ - "set", { - 'stats_data': {'dicttype': {"a": 1, "b": 2, "c": 3} } - }]}, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # check its value - self.session.group_sendmsg({ "command": [ "show", { 'stats_item_name': 'dicttype' } ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, {'dicttype': {"a": 1, "b": 2, "c": 3}}), - result_data) - self.assertEqual(result_ok(0, {'dicttype': self.stats_data['dicttype']}), - result_data) - self.assertEqual(len(self.session.message_queue), 0) - - # test set list-type value - self.assertEqual(self.stats_data['dicttype'], {"a": 1, "b": 2, "c": 3}) - self.session.group_sendmsg({"command": [ - "set", {'stats_data': {'dicttype': {"a": 3, "b": 2, "c": 1, "d": 0} }} ]}, - "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - self.assertEqual(len(self.session.message_queue), 0) - - # check its value - self.session.group_sendmsg({ "command": [ "show", { 'stats_item_name': 'dicttype' }]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - result_data = self.session.get_message("Stats", None) - self.assertEqual(result_ok(0, {'dicttype': {"a": 3, "b": 2, "c": 1, "d": 0} }), - result_data) - self.assertEqual(len(self.session.message_queue), 0) - - def test_config_update(self): - """ - Test for config update - - """ - # test show command without arg - self.session.group_sendmsg({"command": [ "config_update", {"x-version":999} ]}, "Stats") - self.assertEqual(len(self.session.message_queue), 1) - self.subject.check() - self.assertEqual(result_ok(), - self.session.get_message("Stats", None)) - - def test_for_boss(self): - last_queue = self.session.old_message_queue.pop() + def test_command_handler(self): + statsserver = ThreadingServerManager(MyStats) + statsserver.run() + time.sleep(TIMEOUT_SEC*4) + self.base.boss.server._started.wait() self.assertEqual( - last_queue.msg, {'command': ['getstats']}) + send_command( + 'show', 'Stats', + params={ 'owner' : 'Boss', + 'name' : 'boot_time' }), + (0, '2011-06-22T08:14:08Z')) self.assertEqual( - last_queue.env['group'], 'Boss') + send_command( + 'set', 'Stats', + params={ 'owner' : 'Boss', + 'data' : { 'boot_time' : '2012-06-22T18:24:08Z' } }), + (0, None)) + self.assertEqual( + send_command( + 'show', 'Stats', + params={ 'owner' : 'Boss', + 'name' : 'boot_time' }), + (0, '2012-06-22T18:24:08Z')) + self.assertEqual( + send_command('status', 'Stats'), + (0, "Stats is up. (PID " + str(os.getpid()) + ")")) -class TestStats2(unittest.TestCase): - - def setUp(self): - self.session = Session() - self.subject = SessionSubject(session=self.session) - self.listener = CCSessionListener(self.subject) - self.module_name = self.listener.cc_session.get_module_spec().get_module_name() - # check starting - self.assertFalse(self.subject.running) - self.subject.start() - self.assertTrue(self.subject.running) - self.assertEqual(len(self.session.message_queue), 0) - self.assertEqual(self.module_name, 'Stats') - - def tearDown(self): - # check closing - self.subject.stop() - self.assertFalse(self.subject.running) - self.subject.detach(self.listener) - self.listener.stop() - - def test_specfile(self): - """ - Test for specfile + (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'], '2012-06-22T18:24:08Z') + 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) - """ - if "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") - self.assertEqual(stats.SCHEMA_SPECFILE_LOCATION, - os.environ["B10_FROM_SOURCE"] + os.sep + \ - "src" + os.sep + "bin" + os.sep + "stats" + \ - os.sep + "stats-schema.spec") - imp.reload(stats) - # change path of SPECFILE_LOCATION - stats.SPECFILE_LOCATION = TEST_SPECFILE_LOCATION - stats.SCHEMA_SPECFILE_LOCATION = TEST_SPECFILE_LOCATION - self.assertEqual(stats.SPECFILE_LOCATION, TEST_SPECFILE_LOCATION) - self.subject = stats.SessionSubject(session=self.session) - self.session = self.subject.session - self.listener = stats.CCSessionListener(self.subject) + self.assertEqual( + send_command('__UNKNOWN__', 'Stats'), + (1, "Unknown command: '__UNKNOWN__'")) - self.assertEqual(self.listener.stats_spec, []) - self.assertEqual(self.listener.stats_data, {}) + statsserver.shutdown() - self.assertEqual(self.listener.commands_spec, [ - { - "command_name": "status", - "command_description": "identify whether stats module is alive or not", - "command_args": [] - }, - { - "command_name": "the_dummy", - "command_description": "this is for testing", - "command_args": [] - }]) + 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.parse_spec(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'], "1970-01-01T00:00:00Z") + self.assertEqual(my_statistics_data['boot_time'], "1970-01-01T00:00:00Z") + self.assertEqual(my_statistics_data['last_update_time'], "1970-01-01T00:00:00Z") + self.assertEqual(my_statistics_data['timestamp'], 0.0) + self.assertEqual(my_statistics_data['lname'], "") + my_statistics_data = stats.parse_spec(self.stats.modules['Boss'].get_statistics_spec()) + self.assertTrue('boot_time' in my_statistics_data) + self.assertEqual(my_statistics_data['boot_time'], "1970-01-01T00:00:00Z") - def test_func_initialize_data(self): - """ - Test for initialize_data function + 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) + my_statistics_data = self.stats.get_statistics_data(owner='Stats') + 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.assertIsNone(self.stats.get_statistics_data(owner='Foo')) + my_statistics_data = self.stats.get_statistics_data(owner='Stats') + self.assertTrue('boot_time' in my_statistics_data) + my_statistics_data = self.stats.get_statistics_data(owner='Stats', name='report_time') + self.assertEqual(my_statistics_data, "1970-01-01T00:00:00Z") + my_statistics_data = self.stats.get_statistics_data(owner='Stats', name='boot_time') + self.assertEqual(my_statistics_data, "1970-01-01T00:00:00Z") + my_statistics_data = self.stats.get_statistics_data(owner='Stats', name='last_update_time') + self.assertEqual(my_statistics_data, "1970-01-01T00:00:00Z") + my_statistics_data = self.stats.get_statistics_data(owner='Stats', name='timestamp') + self.assertEqual(my_statistics_data, 0.0) + my_statistics_data = self.stats.get_statistics_data(owner='Stats', name='lname') + self.assertEqual(my_statistics_data, '') + self.assertIsNone(self.stats.get_statistics_data(owner='Stats', name='Bar')) + self.assertIsNone(self.stats.get_statistics_data(owner='Foo', name='Bar')) + self.assertEqual(self.stats.get_statistics_data(name='Bar'), None) + + 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='2000-01-01T10:10:10Z') + self.assertTrue('Stats' in self.stats.statistics_data) + my_statistics_data = self.stats.statistics_data['Stats'] + self.assertEqual(my_statistics_data['last_update_time'], '2000-01-01T10:10:10Z') + 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']) + + def test_command_status(self): + self.assertEqual(self.stats.command_status(), + isc.config.create_answer( + 0, "Stats is up. (PID " + str(os.getpid()) + ")")) - """ - # prepare for sample data set - stats_spec = [ - { - "item_name": "none_sample", - "item_type": "null", - "item_default": "None" - }, - { - "item_name": "boolean_sample", - "item_type": "boolean", - "item_default": True - }, - { - "item_name": "string_sample", - "item_type": "string", - "item_default": "A something" - }, - { - "item_name": "int_sample", - "item_type": "integer", - "item_default": 9999999 - }, - { - "item_name": "real_sample", - "item_type": "real", - "item_default": 0.0009 - }, - { - "item_name": "list_sample", - "item_type": "list", - "item_default": [0, 1, 2, 3, 4], - "list_item_spec": [] - }, - { - "item_name": "map_sample", - "item_type": "map", - "item_default": {'name':'value'}, - "map_item_spec": [] - }, - { - "item_name": "other_sample", - "item_type": "__unknown__", - "item_default": "__unknown__" - } - ] - # data for comparison - stats_data = { - 'none_sample': None, - 'boolean_sample': True, - 'string_sample': 'A something', - 'int_sample': 9999999, - 'real_sample': 0.0009, - 'list_sample': [0, 1, 2, 3, 4], - 'map_sample': {'name':'value'}, - 'other_sample': '__unknown__' - } - self.assertEqual(self.listener.initialize_data(stats_spec), stats_data) + def test_command_shutdown(self): + 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, "item name is not specified")) + self.assertEqual(self.stats.command_show(owner='Foo', name='_bar_'), + isc.config.create_answer( + 1, "specified module name and/or item name are incorrect")) + self.assertEqual(self.stats.command_show(owner='Foo', name='bar'), + isc.config.create_answer( + 1, "specified module name and/or item name are incorrect")) + orig_get_timestamp = stats.get_timestamp + orig_get_datetime = stats.get_datetime + stats.get_timestamp = lambda : 1308730448.965706 + stats.get_datetime = lambda : '2011-06-22T08:14:08Z' + self.assertEqual(stats.get_timestamp(), 1308730448.965706) + self.assertEqual(stats.get_datetime(), '2011-06-22T08:14:08Z') + self.assertEqual(self.stats.command_show(owner='Stats', name='report_time'), \ + isc.config.create_answer(0, '2011-06-22T08:14:08Z')) + self.assertEqual(self.stats.statistics_data['Stats']['timestamp'], 1308730448.965706) + self.assertEqual(self.stats.statistics_data['Stats']['boot_time'], '1970-01-01T00:00:00Z') + 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) - def test_func_main(self): - # explicitly make failed - self.session.close() - stats.main(session=self.session) + 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), 2) + for item in schema: + self.assertTrue(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.assertFalse('Stats' in value) + self.assertFalse('Boss' in value) + self.assertFalse('Auth' in value) + for item in value: + 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.assertFalse('Stats' in value) + self.assertFalse('Boss' in value) + self.assertFalse('Auth' in value) + self.assertTrue(len(value) == 7) + self.assertTrue('item_name' in value) + self.assertTrue('item_type' in value) + self.assertTrue('item_optional' in value) + self.assertTrue('item_default' in value) + self.assertTrue('item_title' in value) + self.assertTrue('item_description' in value) + self.assertTrue('item_format' in value) + self.assertEqual(value['item_name'], 'report_time') + self.assertEqual(value['item_format'], 'date-time') + + self.assertEqual(self.stats.command_showschema(owner='Foo'), + isc.config.create_answer( + 1, "specified module name and/or item name are incorrect")) + self.assertEqual(self.stats.command_showschema(owner='Foo', name='bar'), + isc.config.create_answer( + 1, "specified module name and/or item name are incorrect")) + self.assertEqual(self.stats.command_showschema(owner='Stats', name='bar'), + isc.config.create_answer( + 1, "specified module name and/or item name are incorrect")) + 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 : '2011-06-22T06:12:38Z' + (rcode, value) = isc.config.ccsession.parse_answer( + self.stats.command_set(owner='Boss', + data={ 'boot_time' : '2011-06-22T13:15:04Z' })) + stats.get_datetime = orig_get_datetime + self.assertEqual(rcode, 0) + self.assertTrue(value is None) + self.assertEqual(self.stats.statistics_data['Boss']['boot_time'], + '2011-06-22T13:15:04Z') + self.assertEqual(self.stats.statistics_data['Stats']['last_update_time'], + '2011-06-22T06:12:38Z') + 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, + "specified module name and/or statistics data are incorrect:" + + " 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_osenv(self): """ @@ -652,11 +515,8 @@ class TestStats2(unittest.TestCase): os.environ["B10_FROM_SOURCE"] = path imp.reload(stats) -def result_ok(*args): - if args: - return { 'result': list(args) } - else: - return { 'result': [ 0 ] } +def test_main(): + unittest.main() if __name__ == "__main__": - unittest.main() + test_main() diff --git a/src/bin/stats/tests/test_utils.py b/src/bin/stats/tests/test_utils.py index bd23182d2c..cfffc15a35 100644 --- a/src/bin/stats/tests/test_utils.py +++ b/src/bin/stats/tests/test_utils.py @@ -42,11 +42,10 @@ def send_shutdown(module_name): return send_command("shutdown", module_name) class ThreadingServerManager: - def __init__(self, server_class, verbose): + def __init__(self, server_class): self.server_class = server_class self.server_class_name = server_class.__name__ - self.verbose = verbose - self.server = self.server_class(self.verbose) + self.server = self.server_class() self.server._thread = threading.Thread( name=self.server_class_name, target=self.server.run) self.server._thread.daemon = True @@ -60,10 +59,9 @@ class ThreadingServerManager: self.server._thread.join(TIMEOUT_SEC) class MockMsgq: - def __init__(self, verbose): - self.verbose = verbose + def __init__(self): self._started = threading.Event() - self.msgq = msgq.MsgQ(None, verbose) + self.msgq = msgq.MsgQ(None) result = self.msgq.setup() if result: sys.exit("Error on Msgq startup: %s" % result) @@ -81,7 +79,7 @@ class MockMsgq: self.msgq.shutdown() class MockCfgmgr: - def __init__(self, verbose): + def __init__(self): self._started = threading.Event() self.cfgmgr = isc.config.cfgmgr.ConfigManager( os.environ['CONFIG_TESTDATA_PATH'], "b10-config.db") @@ -127,8 +125,7 @@ class MockBoss: """ _BASETIME = (2011, 6, 22, 8, 14, 8, 2, 173, 0) - def __init__(self, verbose): - self.verbose = verbose + def __init__(self): self._started = threading.Event() self.running = False self.spec_file = io.StringIO(self.spec_str) @@ -200,8 +197,7 @@ class MockAuth: } } """ - def __init__(self, verbose): - self.verbose = verbose + def __init__(self): self._started = threading.Event() self.running = False self.spec_file = io.StringIO(self.spec_str) @@ -239,9 +235,9 @@ class MockAuth: return isc.config.create_answer(1, "Unknown Command") class MyStats(stats.Stats): - def __init__(self, verbose): + def __init__(self): self._started = threading.Event() - stats.Stats.__init__(self, verbose) + stats.Stats.__init__(self) def run(self): self._started.set() @@ -251,9 +247,9 @@ class MyStats(stats.Stats): send_shutdown("Stats") class MyStatsHttpd(stats_httpd.StatsHttpd): - def __init__(self, verbose): + def __init__(self): self._started = threading.Event() - stats_httpd.StatsHttpd.__init__(self, verbose) + stats_httpd.StatsHttpd.__init__(self) def run(self): self._started.set() @@ -263,23 +259,22 @@ class MyStatsHttpd(stats_httpd.StatsHttpd): send_shutdown("StatsHttpd") class BaseModules: - def __init__(self, verbose): - self.verbose = verbose + def __init__(self): self.class_name = BaseModules.__name__ # Change value of BIND10_MSGQ_SOCKET_FILE in environment variables os.environ['BIND10_MSGQ_SOCKET_FILE'] = tempfile.mktemp(prefix='unix_socket.') # MockMsgq - self.msgq = ThreadingServerManager(MockMsgq, self.verbose) + self.msgq = ThreadingServerManager(MockMsgq) self.msgq.run() # MockCfgmgr - self.cfgmgr = ThreadingServerManager(MockCfgmgr, self.verbose) + self.cfgmgr = ThreadingServerManager(MockCfgmgr) self.cfgmgr.run() # MockBoss - self.boss = ThreadingServerManager(MockBoss, self.verbose) + self.boss = ThreadingServerManager(MockBoss) self.boss.run() # MockAuth - self.auth = ThreadingServerManager(MockAuth, self.verbose) + self.auth = ThreadingServerManager(MockAuth) self.auth.run() def shutdown(self):