diff options
| -rw-r--r-- | test/nagios/nagios_report.py | 24 | ||||
| -rw-r--r-- | test/nagios/nagios_test.py | 49 | ||||
| -rw-r--r-- | test/nagios/report.py | 19 | ||||
| -rwxr-xr-x | test/nagios/soledad_sync.py | 96 | ||||
| -rwxr-xr-x | test/nagios/webapp_login.py | 28 |
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) |