Commit 98bd30c9 authored by Alain Takoudjou's avatar Alain Takoudjou

migrates old promises used in caddy-frontend SR to new format

First start of promise migration to new python promises format. Promises migrated here are
used in caddy-frontend and monitor stack.
parent 47b0ebc0
......@@ -57,17 +57,15 @@ setup(name=name,
'requests',
'jsonschema',
'zc.buildout',
'pycurl',
] + additional_install_requires,
extras_require = {
'lampconfigure': ["mysqlclient"], #needed for MySQL Database access
'zodbpack': ['ZODB3'], # needed to play with ZODB
'flask_auth' : ["Flask-Auth"],
'networkbench' : ['pycurl'],
'check_web_page_http_cache_hit' : ['pycurl'], # needed for check_web_page_http_cache_hit module
},
tests_require = [
'mock',
'pycurl'
],
zip_safe=False, # proxy depends on Flask, which has issues with
# accessing templates
......
from zope import interface as zope_interface
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise, TestResult
import re
import time
import os
class RunPromise(GenericPromise):
zope_interface.implements(interface.IPromise)
def __init__(self, config):
GenericPromise.__init__(self, config)
# set periodicity to run the promise twice per day
self.custom_frequency = int(self.getConfig('frequency', 720))
self.setPeriodicity(self.custom_frequency)
def sense(self):
"""
Check if http log file contain errors
"""
log_file = self.getConfig('log-file')
maximum_delay = int(self.getConfig('maximum-delay', 0))
if not log_file:
raise ValueError("log file was not set in promise parameters.")
regex = re.compile("^(\[[^\]]+\]) (\[[^\]]+\]) (.*)$")
error_amount = 0
no_route_error = 0
network_is_unreachable = 0
timeout = 0
parsing_failure = 0
if not os.path.exists(log_file):
# file don't exist, nothing to check
self.logger.info("OK")
return
with open(log_file) as f:
f.seek(0, 2)
block_end_byte = f.tell()
f.seek(-min(block_end_byte, 4096), 1)
data = f.read()
for line in reversed(data.splitlines()):
m = regex.match(line)
if m is None:
continue
dt, level, msg = m.groups()
try:
try:
t = time.strptime(dt[1:-1], "%a %b %d %H:%M:%S %Y")
except ValueError:
# Fail to parser for the first time, try a different output.
t = time.strptime(dt[1:-1], "%a %b %d %H:%M:%S.%f %Y")
except ValueError:
# Probably it fail to parse
if parsing_failure < 3:
# Accept failure 2 times, as the line can be actually
# cut on the middle.
parsing_failure += 1
continue
raise
if maximum_delay and (time.time()-time.mktime(t)) > maximum_delay:
# no result in the latest hour
break
if level != "[error]":
continue
# Classify the types of errors
if "(113)No route to host" in msg:
no_route_error += 1
elif "(101)Network is unreachable" in msg:
network_is_unreachable += 1
elif "(110)Connection timed out" in msg:
timeout += 1
error_amount += 1
if error_amount:
self.logger.error("ERROR=%s (NOROUTE=%s, UNREACHABLENET=%s, TIMEOUT=%s)" % (
error_amount, no_route_error, network_is_unreachable, timeout))
else:
self.logger.info("OK")
def anomaly(self):
# only check the result of the two latest sense call
return self._test(result_count=2, failure_amount=2, latest_minute=self.custom_frequency*3)
def test(self):
return TestResult(message="")
from zope import interface as zope_interface
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise, TestResult
import re
import time
from slapos.networkbench.ping import ping, ping6
class RunPromise(GenericPromise):
zope_interface.implements(interface.IPromise)
def __init__(self, config):
GenericPromise.__init__(self, config)
# set periodicity to run the promise twice per day
self.custom_frequency = int(self.getConfig('frequency', 720))
self.setPeriodicity(self.custom_frequency)
def sense(self):
"""
Check if there ICMP packets lost on given address
"""
# Address to ping to
address = self.getConfig('address')
if not address:
raise ValueError("'address' was not set in promise parameters.")
# Force use ipv4 protocol ?
ipv4 = self.getConfig('ipv4') in ('True', 'true', '1')
count = int(self.getConfig('count', 10))
threshold = int(self.getConfig('threshold', 0))
if threshold < 0:
raise ValueError("'threshold' value should be greater than 0.")
if ipv4:
result = ping(address, count=count)
else:
result = ping6(address, count=count)
message = "%s host=%s code=%s, result=%s, packet_lost_ratio=%s msg=%s" % result
packet_lost_ratio = int(result[4])
if packet_lost_ratio == -1 or packet_lost_ratio > threshold:
# Packet lost occurred
self.logger.error(message)
else:
self.logger.info(message)
def anomaly(self):
# only check the result of the two latest sense call
return self._test(result_count=2, failure_amount=2, latest_minute=self.custom_frequency*3)
def test(self):
return TestResult(message="")
from zope import interface as zope_interface
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise, TestResult
import re
import time
from slapos.networkbench.ping import ping, ping6
class RunPromise(GenericPromise):
zope_interface.implements(interface.IPromise)
def __init__(self, config):
GenericPromise.__init__(self, config)
# set periodicity to run the promise twice per day
self.custom_frequency = int(self.getConfig('frequency', 720))
self.setPeriodicity(self.custom_frequency)
def sense(self):
"""
Check re6st optimal status
"""
# promise ipv6 and ipv4 address to compare.
ipv4 = self.getConfig('ipv4')
ipv6 = self.getConfig('ipv6')
count = int(self.getConfig('count', 10))
if not ipv4:
raise ValueError("'ipv4' was not set in promise parameters.")
if not ipv6:
raise ValueError("'ipv6' was not set in promise parameters.")
result_ipv4 = ping(ipv4, count=count)
result_ipv6 = ping6(ipv6, count=count)
# push into to the log file
self.logger.info("%s host=%s code=%s, result=%s, packet_lost_ratio=%s msg=%s" % result_ipv4)
self.logger.info("%s host=%s code=%s, result=%s, packet_lost_ratio=%s msg=%s" % result_ipv6)
if result_ipv4[3] == "failed" and result_ipv6[3] != "failed":
# IPv4 is unreacheable
self.logger.info("OK: IPv4 unreachable, IPv6 reachable")
return
if result_ipv6[3] == "failed":
# IPv6 is unreacheable
self.logger.error("FAILED: IPv4 reachable, IPv6 unreachable")
return
latency4 = float(result_ipv4[3])
latency6 = float(result_ipv6[3])
# We can consider that at worst 1ms is added to
# ipv4 response, due the usage of openvpn.
acceptable_delay = int(self.getConfig('acceptable-delay', 1))
# We can consider that we accept a certain increase
# on latency, if we are on a bit congested link.
# So 10% is reseonable enough.
acceptable_lost = int(self.getConfig('acceptable-lost', 0.10))
# Increase latency with the value.
latency4 += acceptable_delay + latency4 * acceptable_lost
if latency4 < latency6:
self.logger.error("FAIL %s (latency4) > %s (latence6)" % (latency4, latency6))
else:
# Compare if both has Same working rate
self.logger.info("OK: IPv4 reachable, IPv6 reachable")
def anomaly(self):
# only check the result of the two latest sense call
return self._test(result_count=2, failure_amount=2, latest_minute=self.custom_frequency*3)
def test(self):
return TestResult(message="")
from zope import interface as zope_interface
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise
import os
import pycurl
class RunPromise(GenericPromise):
zope_interface.implements(interface.IPromise)
def __init__(self, config):
GenericPromise.__init__(self, config)
# SR can set custom periodicity
self.setPeriodicity(float(self.getConfig('frequency', 2)))
def sense(self):
"""
Check if frontend URL is available
"""
url = self.getConfig('url')
timeout = int(self.getConfig('timeout', 20))
expected_http_code = int(self.getConfig('http_code', '200'))
curl = pycurl.Curl()
curl.setopt(pycurl.URL, url)
curl.setopt(pycurl.TIMEOUT, timeout)
curl.setopt(pycurl.FOLLOWLOCATION, True)
curl.setopt(pycurl.SSL_VERIFYPEER, False)
curl.setopt(pycurl.SSL_VERIFYHOST, False)
curl.setopt(pycurl.WRITEFUNCTION, lambda x: None)
ca_cert_file = self.getConfig('ca-cert-file')
cert_file = self.getConfig('cert-file')
key_file = self.getConfig('key-file')
if key_file and cert_file and ca_cert_file:
# set certificate configuration
curl.setopt(curl.CAINFO, ca_cert_file)
curl.setopt(curl.SSLCERT, cert_file)
curl.setopt(curl.SSLKEY, key_file)
try:
curl.perform()
except pycurl.error, e:
code, message = e
self.logger.error("%s: %s" % (code, message))
return
http_code = curl.getinfo(pycurl.HTTP_CODE)
check_secure = self.getConfig('check-secure')
curl.close()
if http_code == 0:
self.logger.error("%s is not available (server not reachable)." % url)
elif http_code == 401 and check_secure == "1":
self.logger.info("%s is protected (returned %s)." % (url, http_code))
elif http_code != expected_http_code:
self.logger.error("%s is not available (returned %s, expected %s)." % (
url, http_code, expected_http_code))
else:
self.logger.info("%s: URL is available" % http_code)
def anomaly(self):
return self._test(result_count=3, failure_amount=3)
from zope import interface as zope_interface
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise
try:
import subprocess32 as subprocess
except ImportError:
import subprocess
class RunPromise(GenericPromise):
zope_interface.implements(interface.IPromise)
def __init__(self, config):
GenericPromise.__init__(self, config)
self.setPeriodicity(minute=int(self.getConfig('frequency', 5)))
def sense(self):
"""
Check trafficserver cache availability
"""
traffic_line = self.getConfig('wrapper-path')
process = subprocess.Popen(
[traffic_line, '-r', 'proxy.node.cache.percent_free'],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
result = process.communicate()[0].strip()
if process.returncode == 0:
self.logger.info("OK")
else:
self.logger.error("Cache not available, availability: %s" % result)
def anomaly(self):
"""
There is an anomaly if last 3 senses were bad.
"""
return self._anomaly(result_count=3, failure_amount=3)
\ No newline at end of file
from zope import interface as zope_interface
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise
try:
import subprocess32 as subprocess
except ImportError:
import subprocess
class RunPromise(GenericPromise):
zope_interface.implements(interface.IPromise)
def __init__(self, config):
GenericPromise.__init__(self, config)
# check configuration every 5 minutes (only for anomaly)
self.setPeriodicity(minute=int(self.getConfig('frequency', 5)))
def sense(self):
"""
Run frontend validatation script
"""
validate_script = self.getConfig('verification-script')
if not validate_script:
raise ValueError("'verification-script' was not set in promise parameters.")
process = subprocess.Popen(
[validate_script],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
message = process.communicate()[0]
if process.returncode == 0:
self.logger.info("OK")
else:
self.logger.error("%s" % message)
def anomaly(self):
return self._anomaly(result_count=1, failure_amount=1)
##############################################################################
#
# Copyright (c) 2018 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.
#
##############################################################################
from slapos.test.promise.plugin import TestPromisePluginMixin
import os
import time
from slapos.test.promise import data
from slapos.grid.promise import PromiseError
class TestCheckErrorOnHttpLog(TestPromisePluginMixin):
def get_time(self, sec):
return time.strftime("%a %b %d %H:%M:%S %Y", time.localtime(time.time()-sec))
def _update_logs(self):
log_file_list = [
"apache_error_log",
"infoonly_error_log",
"timeout_error_log",
"unreachable_error_log"]
i = 600
for log_file in log_file_list:
new = ""
old = ""
with open(self.base_path + "/" + log_file) as f:
for line in f:
new += line.replace("DATETIME", self.get_time(i))
old += line.replace("DATETIME", self.get_time(i+3600))
i -= 1
with open(self.base_path + "/SOFTINST-0_" + log_file, "w") as f:
f.write(old)
f.write(new)
def setUp(self):
TestPromisePluginMixin.setUp(self)
log_folder = os.path.join(self.partition_dir, 'var/log')
os.makedirs(log_folder)
self.base_path = "/".join(data.__file__.split("/")[:-1])
self._update_logs()
self.promise_name = "check-error-on-apache-log.py"
self.promise_pyc = os.path.join(self.plugin_dir, self.promise_name + "c")
self.base_content = """from slapos.promise.plugin.check_error_on_http_log import RunPromise
extra_config_dict = {
'log-file': '%(log_file)s',
'maximum-delay': '%(maximum_delay)s',
}
"""
def tearDown(self):
TestPromisePluginMixin.tearDown(self)
def test_no_error(self):
self.base_path = "/".join(data.__file__.split("/")[:-1])
content = self.base_content % {
'log_file': self.base_path + "/SOFTINST-0_infoonly_error_log",
'maximum_delay': 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, enable_anomaly=True)
self.launcher.run()
# run a second time to add more results
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], False)
self.assertEqual(result['result']['message'], "OK")
content = self.base_content % {
'log_file': self.base_path + "/SOFTINST-0_infoonly_error_log",
'maximum_delay': 3600
}
self.writePromise(self.promise_name, content)
# remove previous pyc, so modified promise file will be used
if os.path.exists(self.promise_pyc):
os.unlink(self.promise_pyc)
# Ignore periodicity of the promise
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], False)
self.assertEqual(result['result']['message'], "OK")
def test_error(self):
self.base_path = "/".join(data.__file__.split("/")[:-1])
content = self.base_content % {
'log_file': self.base_path + "/SOFTINST-0_apache_error_log",
'maximum_delay': 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, enable_anomaly=True)
self.launcher.run()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "ERROR=2 (NOROUTE=2, UNREACHABLENET=0, TIMEOUT=0)")
content = self.base_content % {
'log_file': self.base_path + "/SOFTINST-0_apache_error_log",
'maximum_delay': 3600
}
self.writePromise(self.promise_name, content)
# remove previous pyc, so modified promise file will be used
if os.path.exists(self.promise_pyc):
os.unlink(self.promise_pyc)
# Ignore periodicity of the promise
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "ERROR=1 (NOROUTE=1, UNREACHABLENET=0, TIMEOUT=0)")
def test_error_timeout(self):
self.base_path = "/".join(data.__file__.split("/")[:-1])
content = self.base_content % {
'log_file': self.base_path + "/SOFTINST-0_timeout_error_log",
'maximum_delay': 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, enable_anomaly=True)
self.launcher.run()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "ERROR=4 (NOROUTE=0, UNREACHABLENET=0, TIMEOUT=4)")
content = self.base_content % {
'log_file': self.base_path + "/SOFTINST-0_timeout_error_log",
'maximum_delay': 3600
}
self.writePromise(self.promise_name, content)
# remove previous pyc, so modified promise file will be used
if os.path.exists(self.promise_pyc):
os.unlink(self.promise_pyc)
# Ignore periodicity of the promise
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "ERROR=2 (NOROUTE=0, UNREACHABLENET=0, TIMEOUT=2)")
def test_error_unreacheabler(self):
self.base_path = "/".join(data.__file__.split("/")[:-1])
content = self.base_content % {
'log_file': self.base_path + "/SOFTINST-0_unreachable_error_log",
'maximum_delay': 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, enable_anomaly=True)
self.launcher.run()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "ERROR=11 (NOROUTE=0, UNREACHABLENET=11, TIMEOUT=0)")
content = self.base_content % {
'log_file': self.base_path + "/SOFTINST-0_unreachable_error_log",
'maximum_delay': 3600
}
self.writePromise(self.promise_name, content)
# remove previous pyc, so modified promise file will be used
if os.path.exists(self.promise_pyc):
os.unlink(self.promise_pyc)
# Ignore periodicity of the promise
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "ERROR=11 (NOROUTE=0, UNREACHABLENET=11, TIMEOUT=0)")
if __name__ == '__main__':
unittest.main()
##############################################################################
#
# Copyright (c) 2018 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.
#
##############################################################################
from slapos.test.promise.plugin import TestPromisePluginMixin
from slapos.grid.promise import PromiseError
import os
import os.path
import socket
class TestCheckICMPPacketLost(TestPromisePluginMixin):
def setUp(self):
TestPromisePluginMixin.setUp(self)
log_folder = os.path.join(self.partition_dir, 'var/log')
os.makedirs(log_folder)
self.promise_name = "check-icmp-packet-lost.py"
self.base_content = """from slapos.promise.plugin.check_icmp_packet_lost import RunPromise
extra_config_dict = {
'ipv4': '%(ipv4)s',
'count': '%(count)s',
'address': '%(address)s',
'threshold': '%(threshold)s'
}
"""
def tearDown(self):
TestPromisePluginMixin.tearDown(self)
def test_localhost(self):
content = self.base_content % {
'address': "localhost",
'count': 5,
"ipv4": True,
"threshold": 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, timeout=20, enable_anomaly=True)
self.launcher.run()
# run a second time to add more results
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], False)
self.assertTrue('packet_lost_ratio=0' in result['result']['message'])
def test_error(self):
content = self.base_content % {
'address': "couscous",
'count': 5,
"ipv4": True,
"threshold": 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, timeout=20, enable_anomaly=True)
self.launcher.run()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertTrue('packet_lost_ratio=-1' in result['result']['message'])
def test_localhost6_with_ping6(self):
content = self.base_content % {
'address': "::1",
'count': 5,
"ipv4": False,
"threshold": 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, timeout=20, enable_anomaly=True)
self.launcher.run()
# run a second time to add more results
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], False)
self.assertTrue('packet_lost_ratio=0' in result['result']['message'])
def test_localhost6_with_ping4(self):
content = self.base_content % {
'address': "::1",
'count': 5,
"ipv4": True,
"threshold": 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, timeout=20, enable_anomaly=True)
self.launcher.run()
# run a second time to add more results
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], False)
self.assertTrue('packet_lost_ratio=0' in result['result']['message'])
def test_error6(self):
content = self.base_content % {
'address': "couscous",
'count': 5,
"ipv4": False,
"threshold": 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, timeout=20, enable_anomaly=True)
self.launcher.run()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertTrue('packet_lost_ratio=-1' in result['result']['message'])
def test_packet_lost_less_than_threshold(self):
content = self.base_content % {
'address': "10.2.3.4",
'count': 5,
"ipv4": True,
"threshold": 110
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, timeout=20, enable_anomaly=True)
self.launcher.run()
# run a second time to add more results
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], False)
self.assertTrue('packet_lost_ratio=100' in result['result']['message'])
if __name__ == '__main__':
unittest.main()
##############################################################################
#
# Copyright (c) 2018 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.
#
##############################################################################
from slapos.test.promise.plugin import TestPromisePluginMixin
from slapos.grid.promise import PromiseError
import os
class TestCheckRe6stOptimalStatus(TestPromisePluginMixin):
def setUp(self):
TestPromisePluginMixin.setUp(self)
self.promise_name = "check-icmp-packet-lost.py"
self.base_content = """from slapos.promise.plugin.check_re6st_optimal_status import RunPromise
extra_config_dict = {
'ipv4': '%(ipv4)s',
'count': '%(count)s',
'ipv6': '%(ipv6)s',
}
"""
def tearDown(self):
TestPromisePluginMixin.tearDown(self)
def test_ipv6_is_faster(self):
content = self.base_content % {
'ipv4': "8.8.8.8",
'ipv6': '::1',
'count': 5,
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, timeout=20, enable_anomaly=True)
self.launcher.run()
# run a second time to add more results
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
last_message = result['result']['message'].split('\n')[-1]
self.assertEqual(result['result']['failed'], False)
self.assertEqual(last_message, "OK: IPv4 reachable, IPv6 reachable")
def test_ipv4_is_faster(self):
content = self.base_content % {
'ipv4': "127.0.0.1",
'ipv6': '2001:67c:1254::1',
'count': 5,
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, timeout=20, enable_anomaly=True)
self.launcher.run()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
last_message = result['result']['message'].split('\n')[-1]
self.assertEqual(result['result']['failed'], True)
self.assertTrue(last_message.startswith('FAIL'))
def test_ipv4_unreacheable_and_ipv6_ok(self):
content = self.base_content % {
'ipv4': "couscous",
'ipv6': '::1',
'count': 5,
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, timeout=20, enable_anomaly=True)
self.launcher.run()
# run a second time to add more results
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
last_message = result['result']['message'].split('\n')[-1]
self.assertEqual(result['result']['failed'], False)
self.assertEqual(last_message, "OK: IPv4 unreachable, IPv6 reachable")
def test_ipv6_fail(self):
content = self.base_content % {
'ipv4': "127.0.0.1",
'ipv6': 'couscous',
'count': 5,
}
self.writePromise(self.promise_name, content)
self.configureLauncher(force=True, timeout=20, enable_anomaly=True)
self.launcher.run()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
last_message = result['result']['message'].split('\n')[-1]
self.assertEqual(result['result']['failed'], True)
self.assertEqual(last_message, "FAILED: IPv4 reachable, IPv6 unreachable")
if __name__ == '__main__':
unittest.main()
##############################################################################
#
# Copyright (c) 2018 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.
#
##############################################################################
from slapos.test.promise.plugin import TestPromisePluginMixin
from slapos.grid.promise import PromiseError
import os
class TestCheckUrlAvailable(TestPromisePluginMixin):
def setUp(self):
TestPromisePluginMixin.setUp(self)
self.promise_name = "check-url-available.py"
self.base_content = """from slapos.promise.plugin.check_url_available import RunPromise
extra_config_dict = {
'url': '%(url)s',
'timeout': %(timeout)s,
'check-secure': %(check_secure)s,
}
"""
def tearDown(self):
TestPromisePluginMixin.tearDown(self)
def test_check_url_bad(self):
content = self.base_content % {
'url': 'https://',
'timeout': 10,
'check_secure': 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "3: Bad URL")
def test_check_url_malformed(self):
content = self.base_content % {
'url': '',
'timeout': 10,
'check_secure': 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "3: <url> malformed")
def test_check_url_site_off(self):
content = content = self.base_content % {
'url': 'https://localhost:56789/site',
'timeout': 10,
'check_secure': 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher()
with self.assertRaises(PromiseError):
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "7: Failed to connect to localhost port 56789: Connection refused")
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