summaryrefslogtreecommitdiff
path: root/test/nagios
diff options
context:
space:
mode:
authorAzul <azul@leap.se>2014-04-15 17:17:36 +0200
committerAzul <azul@leap.se>2014-04-17 10:39:59 +0200
commitbe2971c2e615cc8a808822317d049e99f5183bdc (patch)
treeee2e77b91f661f5fa6c69f5d851eb848f38eb3b6 /test/nagios
parent44d2d031555c889b94e9738cb45740b16a4071ce (diff)
refactor: move nagios specifs to nagios_test
nagios_test.run takes a function and executes it. If it returns nothing or 0 and OK nagios message is printed. If it returns sth. else this will be printed a a warning If it raises an exception that will result in a CRITICAL report. This way we can keep the nagios things outside the test cases and just write simple functions that either return 0, a warnign or raise a meaningful exception
Diffstat (limited to 'test/nagios')
-rw-r--r--test/nagios/nagios_report.py24
-rw-r--r--test/nagios/nagios_test.py49
-rw-r--r--test/nagios/report.py19
-rwxr-xr-xtest/nagios/soledad_sync.py96
-rwxr-xr-xtest/nagios/webapp_login.py28
5 files changed, 121 insertions, 95 deletions
diff --git a/test/nagios/nagios_report.py b/test/nagios/nagios_report.py
new file mode 100644
index 0000000..13cd551
--- /dev/null
+++ b/test/nagios/nagios_report.py
@@ -0,0 +1,24 @@
+def functions_for_system(under_test):
+ """
+ returns a set of functions to use for nagios reporting:
+ >>> ok, warn, critical, unknown = functions_for_system("tested system")
+
+ each of them will print a nagios line with its argument and
+ return the exit code:
+ >>> warn("that looks strange")
+ 1 tested system - WARNING - that looks strange
+ 1
+ """
+ def report_function(code):
+ return lambda message : report(under_test, code, message)
+ return map(report_function, [0,1,2,3])
+
+def report(system, code, message):
+ codes = {0: 'OK', 1: 'WARNING', 2: 'CRITICAL', 3: 'UNKNOWN'}
+ print "%d %s - %s - %s" % \
+ (code, system, codes[code], message)
+ return code
+
+if __name__ == "__main__":
+ import doctest
+ doctest.testmod()
diff --git a/test/nagios/nagios_test.py b/test/nagios/nagios_test.py
new file mode 100644
index 0000000..3eb8d55
--- /dev/null
+++ b/test/nagios/nagios_test.py
@@ -0,0 +1,49 @@
+import __main__ as main
+import os
+import sys
+import nagios_report
+
+def run(test):
+ """
+ run takes a function and tries it out.
+ If it returns nothing or 0 everything is fine and run prints an OK message
+ with the function name.
+ >>> def this_works_fine(): return
+ >>> run(this_works_fine)
+ 0 nagios_test.py - OK - this_works_fine
+ 0
+ >>> def this_also_works_fine(): return 0
+ >>> run(this_also_works_fine)
+ 0 nagios_test.py - OK - this_also_works_fine
+ 0
+
+ If the function returns something else it will be printed as a warning.
+ >>> run(lambda : "this is a warning")
+ 1 nagios_test.py - WARNING - this is a warning
+ 1
+
+ Errors raised will result in a CRITICAL nagios string.
+ >>> def failure(): raise Exception("something went wrong")
+ >>> run(failure)
+ 2 nagios_test.py - CRITICAL - something went wrong
+ 2
+ """
+ try:
+ name = os.path.basename(main.__file__)
+ except AttributeError:
+ name = sys.argv[0]
+ ok, warn, fail, unknown = nagios_report.functions_for_system(name)
+ try:
+ warning = test()
+ if warning and warning != 0:
+ code = warn(warning)
+ else:
+ code = ok(test.__name__)
+ except Exception as exc:
+ code = fail(exc.message or str(exc))
+ return code
+
+
+if __name__ == "__main__":
+ import doctest
+ doctest.testmod()
diff --git a/test/nagios/report.py b/test/nagios/report.py
deleted file mode 100644
index d2720a5..0000000
--- a/test/nagios/report.py
+++ /dev/null
@@ -1,19 +0,0 @@
-system = 'undefined'
-
-def report(code, message):
- codes = {0: 'OK', 1: 'WARNING', 2: 'CRITICAL', 3: 'UNKNOWN'}
- print "%d %s - %s - %s" % \
- (code, system, codes[code], message)
- exit(code)
-
-def fail(message):
- report(2, message)
-
-def warn(message):
- report(1, message)
-
-def ok(message):
- report(0, message)
-
-def unknown(message):
- report(3, message)
diff --git a/test/nagios/soledad_sync.py b/test/nagios/soledad_sync.py
index a3c2d5a..9f51fd1 100755
--- a/test/nagios/soledad_sync.py
+++ b/test/nagios/soledad_sync.py
@@ -32,14 +32,6 @@ HTTPSyncTarget.set_token_credentials = set_token_credentials
HTTPSyncTarget._sign_request = _sign_request
-def fail(reason):
- print '2 soledad_sync - CRITICAL - ' + reason
- exit(2)
-
-# monkey patch webapp_login's fail function to report as soledad
-webapp_login.fail = fail
-
-
# The following function could fetch all info needed to sync using soledad.
# Despite that, we won't use all that info because we are instead faking a
# Soledad sync by using U1DB slightly modified syncing capabilities. Part of
@@ -47,58 +39,42 @@ webapp_login.fail = fail
# to actually use the Soledad client in the future.
def get_soledad_info(config, tempdir):
- # get login and get user info
- user = config['user']
- api = config['api']
- usr = srp.User( user['username'], user['password'], srp.SHA256, srp.NG_1024 )
- try:
+ # get login and get user info
+ user = config['user']
+ api = config['api']
+ usr = srp.User( user['username'], user['password'], srp.SHA256, srp.NG_1024 )
auth = webapp_login.authenticate(api, usr)
- except requests.exceptions.ConnectionError:
- fail('no connection to server')
- # get soledad server url
- service_url = 'https://%s:%d/%d/config/soledad-service.json' % \
- (api['domain'], api['port'], api['version'])
- soledad_hosts = requests.get(service_url).json()['hosts']
- host = soledad_hosts.keys()[0]
- server_url = 'https://%s:%d/user-%s' % \
- (soledad_hosts[host]['hostname'], soledad_hosts[host]['port'],
- auth['id'])
- # get provider ca certificate
- #ca_cert = requests.get('https://127.0.0.1/ca.crt', verify=False).text
- #cert_file = os.path.join(tempdir, 'ca.crt')
- cert_file = None # not used for now
- #with open(cert_file, 'w') as f:
- # f.write(ca_cert)
- return auth['id'], user['password'], server_url, cert_file, auth['token']
-
-
-def run_tests():
- tempdir = tempfile.mkdtemp()
- uuid, password, server_url, cert_file, token = \
- get_soledad_info(webapp_login.read_config(), tempdir)
- exc = None
- try:
- # in the future, we can replace the following by an actual Soledad
- # client sync, if needed
- db = u1db.open(os.path.join(tempdir, '%s.db' % uuid), True)
- creds = {'token': {'uuid': uuid, 'token': token}}
- db.sync(server_url, creds=creds, autocreate=False)
- except Exception as e:
- exc = e
- shutil.rmtree(tempdir)
- exit(report(exc))
-
-
-def report(exc):
- if exc is None:
- print '0 soledad_sync - OK - can sync soledad fine'
- return 0
- if isinstance(exc, u1db.errors.U1DBError):
- print '2 soledad_sync - CRITICAL - ' + exc.message
- else:
- print '2 soledad_sync - CRITICAL - ' + str(exc)
- return 2
-
+ # get soledad server url
+ service_url = 'https://%s:%d/%d/config/soledad-service.json' % \
+ (api['domain'], api['port'], api['version'])
+ soledad_hosts = requests.get(service_url).json()['hosts']
+ host = soledad_hosts.keys()[0]
+ server_url = 'https://%s:%d/user-%s' % \
+ (soledad_hosts[host]['hostname'], soledad_hosts[host]['port'],
+ auth['id'])
+ # get provider ca certificate
+ #ca_cert = requests.get('https://127.0.0.1/ca.crt', verify=False).text
+ #cert_file = os.path.join(tempdir, 'ca.crt')
+ cert_file = None # not used for now
+ #with open(cert_file, 'w') as f:
+ # f.write(ca_cert)
+ return auth['id'], user['password'], server_url, cert_file, auth['token']
+
+
+def can_sync_soledad_fine():
+ tempdir = tempfile.mkdtemp()
+ try:
+ uuid, password, server_url, cert_file, token = \
+ get_soledad_info(webapp_login.read_config(), tempdir)
+ # in the future, we can replace the following by an actual Soledad
+ # client sync, if needed
+ db = u1db.open(os.path.join(tempdir, '%s.db' % uuid), True)
+ creds = {'token': {'uuid': uuid, 'token': token}}
+ db.sync(server_url, creds=creds, autocreate=False)
+ finally:
+ shutil.rmtree(tempdir)
if __name__ == '__main__':
- run_tests()
+ import nagios_test
+ exit_code = nagios_test.run(can_sync_soledad_fine)
+ exit(exit_code)
diff --git a/test/nagios/webapp_login.py b/test/nagios/webapp_login.py
index a7e3473..6d06438 100755
--- a/test/nagios/webapp_login.py
+++ b/test/nagios/webapp_login.py
@@ -9,40 +9,34 @@ import random
import srp._pysrp as srp
import binascii
import yaml
-import report
safe_unhexlify = lambda x: binascii.unhexlify(x) if (
len(x) % 2 == 0) else binascii.unhexlify('0' + x)
-report.system = 'webapp login'
-
def read_config():
with open("/etc/leap/hiera.yaml", 'r') as stream:
config = yaml.load(stream)
user = config['webapp']['nagios_test_user']
if 'username' not in user:
- report.fail('nagios test user lacks username')
+ raise Exception('nagios test user lacks username')
if 'password' not in user:
- report.fail('nagios test user lacks password')
+ raise Exception('nagios test user lacks password')
api = config['api']
api['version'] = config['webapp']['api_version']
return {'api': api, 'user': user}
-def run_tests(config):
+def login_successfully(config=None):
+ config = config or read_config()
user = config['user']
api = config['api']
usr = srp.User(user['username'], user['password'], srp.SHA256, srp.NG_1024)
- try:
- auth = authenticate(api, usr)
- except requests.exceptions.ConnectionError:
- report.fail('no connection to server')
+ auth = authenticate(api, usr)
if ('errors' in auth):
- report.fail('srp password auth failed')
+ raise Exception('srp password auth failed')
usr.verify_session(safe_unhexlify(auth["M2"]))
- if usr.authenticated():
- report.ok('can login to webapp fine')
- report.warn('failed to verify webapp server')
+ if not usr.authenticated():
+ return 'failed to verify webapp server'
def authenticate(api, usr):
api_url = "https://{domain}:{port}/{version}".format(**api)
@@ -55,7 +49,7 @@ def authenticate(api, usr):
response = session.post(api_url + '/sessions', data=params, verify=False)
init = response.json()
if ('errors' in init):
- report.fail('test user not found')
+ raise Exception('test user not found')
M = usr.process_challenge(
safe_unhexlify(init['salt']), safe_unhexlify(init['B']))
response = session.put(api_url + '/sessions/' + uname, verify=False,
@@ -63,4 +57,6 @@ def authenticate(api, usr):
return response.json()
if __name__ == '__main__':
- run_tests(read_config())
+ import nagios_test
+ exit_code = nagios_test.run(login_successfully)
+ exit(exit_code)