mirror of
https://gitlab.isc.org/isc-projects/kea
synced 2025-08-30 13:37:55 +00:00
[trac930] refurbish the unittests for new stats module, new stats
httpd module and new mockups and utilities in test_utils.py
This commit is contained in:
Naoki Kambe
parent
ed5311a26b
commit
1a8c86ea25
@ -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()
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@ -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):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user