Commit c7c26219 authored by Łukasz Nowak's avatar Łukasz Nowak Committed by Łukasz Nowak

promise/plugin: Generally improve check_url_available

Improvements:

 * flake8ize test_check_url_available and check_url_available
 * cover more of check_url_available
parent cd0fb8f8
from zope import interface as zope_interface
from slapos.grid.promise import interface
from slapos.grid.promise.generic import GenericPromise
import os
import requests
class RunPromise(GenericPromise):
zope_interface.implements(interface.IPromise)
......@@ -36,7 +37,8 @@ class RunPromise(GenericPromise):
cert = None
try:
result = requests.get(url, verify=verify, allow_redirects=True, timeout=timeout, cert=cert)
result = requests.get(
url, verify=verify, allow_redirects=True, timeout=timeout, cert=cert)
except requests.ConnectionError as e:
self.logger.error(
"ERROR connection not possible while accessing %r" % (url, ))
......@@ -46,18 +48,18 @@ class RunPromise(GenericPromise):
return
http_code = result.status_code
check_secure = self.getConfig('check-secure')
check_secure = int(self.getConfig('check-secure', 0))
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 == 401 and check_secure == 1:
self.logger.info("%r is protected (returned %s)." % (url, http_code))
elif http_code != expected_http_code:
self.logger.error("%s is not available (returned %s, expected %s)." % (
self.logger.error("%r is not available (returned %s, expected %s)." % (
url, http_code, expected_http_code))
else:
self.logger.info("%s: URL is available" % http_code)
self.logger.info("%r is available" % (url,))
def anomaly(self):
return self._test(result_count=3, failure_amount=3)
##############################################################################
#
# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
# Copyright (c) 2019 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
......@@ -25,12 +25,52 @@
#
##############################################################################
from slapos.test.promise.plugin import TestPromisePluginMixin
from slapos.grid.promise import PromiseError
import os
from slapos.test.promise.plugin import TestPromisePluginMixin
import BaseHTTPServer
import json
import multiprocessing
import time
import unittest
SLAPOS_TEST_IPV4 = '127.0.0.1'
SLAPOS_TEST_IPV4_PORT = 57965
HTTPS_ENDPOINT = "http://%s:%s/" % (SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV4_PORT)
class TestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
def do_GET(self):
timeout = int(self.headers.dict.get('timeout', '0'))
time.sleep(timeout)
response = int(self.path.split('/')[-1])
self.send_response(response)
self.send_header("Content-type", "application/json")
self.end_headers()
response = {
'Path': self.path,
}
self.wfile.write(json.dumps(response, indent=2))
class TestCheckUrlAvailable(TestPromisePluginMixin):
@classmethod
def setUpClass(cls):
server = BaseHTTPServer.HTTPServer(
(SLAPOS_TEST_IPV4, SLAPOS_TEST_IPV4_PORT),
TestHandler)
cls.server_process = multiprocessing.Process(
target=server.serve_forever)
cls.server_process.start()
@classmethod
def tearDownClass(cls):
cls.server_process.terminate()
cls.server_process.join()
def setUp(self):
TestPromisePluginMixin.setUp(self)
self.promise_name = "check-url-available.py"
......@@ -59,7 +99,10 @@ extra_config_dict = {
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "ERROR: Invalid URL u'https://': No host supplied")
self.assertEqual(
result['result']['message'],
"ERROR: Invalid URL u'https://': No host supplied"
)
def test_check_url_malformed(self):
content = self.base_content % {
......@@ -73,7 +116,10 @@ extra_config_dict = {
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "ERROR: Invalid URL '': No schema supplied. Perhaps you meant http://?")
self.assertEqual(
result['result']['message'],
"ERROR: Invalid URL '': No schema supplied. Perhaps you meant http://?"
)
def test_check_url_site_off(self):
content = content = self.base_content % {
......@@ -87,7 +133,64 @@ extra_config_dict = {
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], True)
self.assertEqual(result['result']['message'], "ERROR connection not possible while accessing 'https://localhost:56789/site'")
self.assertEqual(
result['result']['message'],
"ERROR connection not possible while accessing "
"'https://localhost:56789/site'"
)
def test_check_200(self):
url = HTTPS_ENDPOINT + '200'
content = content = self.base_content % {
'url': url,
'timeout': 10,
'check_secure': 0
}
self.writePromise(self.promise_name, content)
self.configureLauncher()
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], False)
self.assertEqual(
result['result']['message'],
"%r is available" % (url,)
)
def test_check_401(self):
url = HTTPS_ENDPOINT + '401'
content = content = self.base_content % {
'url': 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'],
"%r is not available (returned 401, expected 200)." % (url,)
)
def test_check_401_secure(self):
url = HTTPS_ENDPOINT + '401'
content = content = self.base_content % {
'url': url,
'timeout': 10,
'check_secure': 1
}
self.writePromise(self.promise_name, content)
self.configureLauncher()
self.launcher.run()
result = self.getPromiseResult(self.promise_name)
self.assertEqual(result['result']['failed'], False)
self.assertEqual(
result['result']['message'],
"%r is protected (returned 401)." % (url,)
)
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