Commit 154f61d5 authored by Rafael Monnerat's avatar Rafael Monnerat

slapos.networkbench: Refactor and include tests

  Use json as configuration format
  Accept remote configuration file (url to a json)
  Introduce API to assert reponse codes and result for http and dns
  Add Unit tests.
parent f907713d
......@@ -47,6 +47,7 @@ setup(name=name,
'netifaces',
'erp5.util',
'PyRSS2Gen',
'dnspython',
] + additional_install_requires,
extras_require = {
'lampconfigure': ["mysqlclient"], #needed for MySQL Database access
......
networkbench
============
import socket
import time
import dns.resolver
def resolve(name, expected_list=None):
""" Resolve name using standard system name resolution.
"""
begin = time.time()
try:
ip_list = [i.to_text() for i in dns.resolver.query(name, "A")]
resolution = 200
status = "OK"
except dns.resolver.NXDOMAIN:
resolution = 600
status = "Cannot resolve the hostname"
ip_list = []
resolving_time = time.time() - begin
# Out put is:
# TEST IDENTIFIER, NAME, RESOLUTION (200 or 600), Time for resolve,
# status ("OK" or "Cannot resolve the hostname"), Resolved IP.
if expected_list is not None and set(expected_list) != set(ip_list):
status = "UNEXPECTED"
ip_list = "%s (expected) != %s (found)" % (expected_list, ip_list)
return ('DNS', name, resolution, resolving_time, status, ip_list)
import sys
import pycurl
from StringIO import StringIO
def get_curl(buffer, url):
curl = pycurl.Curl()
curl.setopt(curl.URL, url)
curl.setopt(curl.CONNECTTIMEOUT, 10)
curl.setopt(curl.TIMEOUT, 30)
curl.setopt(curl.WRITEDATA, buffer)
curl.setopt(curl.SSL_VERIFYPEER, False)
curl.setopt(curl.SSL_VERIFYHOST, False)
result = "OK"
try:
curl.perform()
except:
import traceback
traceback.print_exc(file=sys.stderr)
sys.stderr.flush()
result = "FAIL"
return curl, result
def request(url, expected_dict):
buffer = StringIO()
curl, result = get_curl(buffer, url)
body = buffer.getvalue()
rendering_time = "%s;%s;%s;%s;%s" % \
(curl.getinfo(curl.NAMELOOKUP_TIME),
curl.getinfo(curl.CONNECT_TIME),
curl.getinfo(curl.PRETRANSFER_TIME),
curl.getinfo(curl.STARTTRANSFER_TIME),
curl.getinfo(curl.TOTAL_TIME))
response_code = curl.getinfo(pycurl.HTTP_CODE)
expected_response = expected_dict.get("expected_response", None)
if expected_response is not None and \
expected_response != response_code:
result = "UNEXPECTED (%s != %s)" % (expected_response, response_code)
expected_text = expected_dict.get("expected_text", None)
if expected_text is not None and \
str(expected_text) not in str(body):
result = "UNEXPECTED (%s not in page content)" % (expected_text)
curl.close()
info_list = ('GET', url, response_code, rendering_time, result)
return info_list
import subprocess
import re
# rtt min/avg/max/mdev = 1.102/1.493/2.203/0.438 ms
ping_re = re.compile(
".*"
"(?P<min>[\d\.]+)/"
"(?P<avg>[\d\.]+)/"
"(?P<max>[\d\.]+)/"
"(?P<mdev>[\d\.]+) ms"
)
date_reg_exp = re.compile('\d{4}[-/]\d{2}[-/]\d{2}')
def ping(host, timeout=10, protocol="4"):
if protocol == '4':
ping_bin = 'ping'
test_title = 'PING'
elif protocol == '6':
ping_bin = 'ping6'
test_title = 'PING6'
proc = subprocess.Popen((ping_bin, '-c', '10', '-w', str(timeout), host),
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = proc.communicate()
if 'Network is unreachable' in err:
return (test_title, host, 600, 'failed', 100, "Network is unreachable")
try:
packet_loss_line, summary_line = (out.splitlines() or [''])[-2:]
except:
return (test_title, host, 600, 'failed', -1, "Fail to parser ping output")
m = ping_re.match(summary_line)
match = re.search('(\d*)% packet loss', packet_loss_line)
packet_lost_ratio = match.group(1)
info_list = (test_title, host, 600, 'failed', packet_lost_ratio, "Cannot ping host")
if packet_lost_ratio != 0:
if m:
info_list = (test_title, host, 200, m.group('avg'), packet_lost_ratio,
'min %(min)s max %(max)s avg %(avg)s' % m.groupdict())
else:
info_list = (test_title, host, 600, 'failed', packet_lost_ratio,
"You have package Lost")
return info_list
def ping6(host, timeout=10):
return ping(host, timeout=10, protocol='6')
##############################################################################
#
# Copyright (c) 2015 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import unittest
import os.path
from slapos.networkbench import dnsbench
from slapos.networkbench.ping import ping, ping6
from slapos.networkbench.http import request
DNS_EXPECTED_LIST = ["85.118.38.162", "176.31.129.213"]
class TestDNSBench(unittest.TestCase):
def test_dnsbench_ok(self):
""" Test dns resolution, this test may fail if
not ran from Europe.
# Update domain later.
"""
info = dnsbench.resolve(
"www.erp5.com", DNS_EXPECTED_LIST)
self.assertEquals(info[0], 'DNS')
self.assertEquals(info[1], 'www.erp5.com')
self.assertEquals(info[2], 200)
self.assertTrue(info[3] < 1)
self.assertEquals(info[4], 'OK')
self.assertEquals(set(info[5]), set([u'85.118.38.162', u'176.31.129.213']))
def test_dnsbench_fail(self):
""" Test dns failure resolution
"""
info = dnsbench.resolve(
"thisdomaindontexist.erp5.com")
self.assertEquals(info[0], 'DNS')
self.assertEquals(info[1], 'thisdomaindontexist.erp5.com')
self.assertEquals(info[2], 600)
self.assertTrue(info[3] < 1)
self.assertEquals(info[4], 'Cannot resolve the hostname')
self.assertEquals(info[5], [])
def test_dnsbench_unexpected(self):
""" Test dns unexpected resolution
"""
info = dnsbench.resolve(
"www.erp5.com", [DNS_EXPECTED_LIST[0]])
self.assertEquals(info[0], 'DNS')
self.assertEquals(info[1], 'www.erp5.com')
self.assertEquals(info[2], 200)
self.assertTrue(info[3] < 1)
self.assertEquals(info[4], 'UNEXPECTED')
self.assertTrue(info[5].startswith("['85.118.38.162'] (expected) != "))
class TestPing(unittest.TestCase):
def test_ping_ok(self):
info = ping("localhost")
self.assertEquals(info[0], 'PING')
self.assertEquals(info[1], 'localhost')
self.assertEquals(info[2], 200)
self.assertTrue(float(info[3]) < 0.2)
self.assertEquals(info[4], '0')
self.assertTrue(info[5].startswith("min"))
def test_ping_fail(self):
info = ping("couscous")
self.assertEquals(info[0], 'PING')
self.assertEquals(info[1], 'couscous')
self.assertEquals(info[2], 600)
self.assertEquals(info[3], 'failed')
self.assertEquals(info[4], -1)
self.assertEquals(info[5], 'Fail to parser ping output')
def test_ping6_ok(self):
info = ping6("localhost")
self.assertEquals(info[0], 'PING6')
self.assertEquals(info[1], 'localhost')
self.assertEquals(info[2], 200)
self.assertTrue(float(info[3]) < 0.2)
self.assertEquals(info[4], '0')
self.assertTrue(info[5].startswith("min"))
def test_ping6_fail(self):
info = ping6("couscous")
self.assertEquals(info[0], 'PING6')
self.assertEquals(info[1], 'couscous')
self.assertEquals(info[2], 600)
self.assertEquals(info[3], 'failed')
self.assertEquals(info[4], -1)
self.assertEquals(info[5], 'Fail to parser ping output')
class TestHTTPBench(unittest.TestCase):
def test_request_ok(self):
""" This test is way to badly written as it depends on
www.erp5.com for now, please replace it
"""
info = request("https://www.erp5.com", {})
self.assertEquals(info[0], 'GET')
self.assertEquals(info[1], 'https://www.erp5.com')
self.assertEquals(info[2], 200)
self.assertEquals(len(info[3].split(';')), 5 )
self.assertEquals(info[4], "OK")
def test_request_expected_response(self):
""" This test is way to badly written as it depends on
www.erp5.com for now, please replace it
"""
info = request("https://www.erp5.com", {"expected_response": 200})
self.assertEquals(info[0], 'GET')
self.assertEquals(info[1], 'https://www.erp5.com')
self.assertEquals(info[2], 200)
self.assertEquals(len(info[3].split(';')), 5 )
self.assertEquals(info[4], "OK")
def test_request_expected_redirection(self):
""" This test is way to badly written as it depends on
www.erp5.com for now, please replace it
"""
info = request("http://www.erp5.com", {"expected_response": 302})
self.assertEquals(info[0], 'GET')
self.assertEquals(info[1], 'http://www.erp5.com')
self.assertEquals(info[2], 302)
self.assertEquals(len(info[3].split(';')), 5 )
self.assertEquals(info[4], "OK")
def test_request_expected_text(self):
""" This test is way to badly written as it depends on
www.erp5.com for now, please replace it
"""
info = request("https://www.erp5.com", {"expected_text": "ERP5"})
self.assertEquals(info[0], 'GET')
self.assertEquals(info[1], 'https://www.erp5.com')
self.assertEquals(info[2], 200)
self.assertEquals(len(info[3].split(';')), 5 )
self.assertEquals(info[4], "OK")
def test_request_fail(self):
""" Test unreachable URL
"""
info = request("http://thisurldontexist.erp5.com", {})
self.assertEquals(info[0], 'GET')
self.assertEquals(info[1], 'http://thisurldontexist.erp5.com')
self.assertEquals(info[2], 0)
self.assertEquals(len(info[3].split(';')), 5 )
self.assertEquals(info[4], "FAIL")
def test_request_unexpected_response(self):
""" This test is way to badly written as it depends on
www.erp5.com for now, please replace it
"""
info = request("http://www.erp5.com", {"expected_response": 200})
self.assertEquals(info[0], 'GET')
self.assertEquals(info[1], 'http://www.erp5.com')
self.assertEquals(info[2], 302)
self.assertEquals(len(info[3].split(';')), 5 )
self.assertEquals(info[4], "UNEXPECTED (200 != 302)")
def test_request_unexpected_text(self):
""" This test is way to badly written as it depends on
www.erp5.com for now, please replace it.
"""
info = request("https://www.erp5.com", {"expected_text": "COUSCOUS"})
self.assertEquals(info[0], 'GET')
self.assertEquals(info[1], 'https://www.erp5.com')
self.assertEquals(info[2], 200)
self.assertEquals(len(info[3].split(';')), 5 )
self.assertEquals(info[4], "UNEXPECTED (COUSCOUS not in page content)")
#def request(url, expected_dict):
#
# rendering_time = "%s;%s;%s;%s;%s" % \
# (curl.getinfo(curl.NAMELOOKUP_TIME),
# curl.getinfo(curl.CONNECT_TIME),
# curl.getinfo(curl.PRETRANSFER_TIME),
# curl.getinfo(curl.STARTTRANSFER_TIME),
# curl.getinfo(curl.TOTAL_TIME))
#
# response_code = curl.getinfo(pycurl.HTTP_CODE)
#
# expected_response = expected_dict.get("expected_response", None)
# if expected_response is not None and \
# expected_response != response_code:
# result = "UNEXPECTED (%s != %s)" % (expected_response, response_code)
#
# expected_text = expected_dict.get("expected_text", None)
# if expected_text is not None and \
# str(expected_text) not in str(body):
# result = "UNEXPECTED (%s not in page content)" % (expected_text)
#
#
# info_list = ('GET', url, response_code, rendering_time, result)
#
# return info_list
#
if __name__ == '__main__':
unittest.main()
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment