summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
Diffstat (limited to 'test')
-rw-r--r--test/nagios/support/nagios_report.py24
-rw-r--r--test/nagios/support/nagios_test.py49
-rwxr-xr-xtest/nagios/webapp_login.py127
3 files changed, 158 insertions, 42 deletions
diff --git a/test/nagios/support/nagios_report.py b/test/nagios/support/nagios_report.py
new file mode 100644
index 0000000..13cd551
--- /dev/null
+++ b/test/nagios/support/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/support/nagios_test.py b/test/nagios/support/nagios_test.py
new file mode 100644
index 0000000..3eb8d55
--- /dev/null
+++ b/test/nagios/support/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/webapp_login.py b/test/nagios/webapp_login.py
index 6d06438..7e2efd7 100755
--- a/test/nagios/webapp_login.py
+++ b/test/nagios/webapp_login.py
@@ -13,48 +13,91 @@ import yaml
safe_unhexlify = lambda x: binascii.unhexlify(x) if (
len(x) % 2 == 0) else binascii.unhexlify('0' + x)
-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:
- raise Exception('nagios test user lacks username')
- if 'password' not in user:
- raise Exception('nagios test user lacks password')
- api = config['api']
- api['version'] = config['webapp']['api_version']
- return {'api': api, 'user': user}
-
-
-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)
- auth = authenticate(api, usr)
- if ('errors' in auth):
- raise Exception('srp password auth failed')
- usr.verify_session(safe_unhexlify(auth["M2"]))
- if not usr.authenticated():
- return 'failed to verify webapp server'
-
-def authenticate(api, usr):
- api_url = "https://{domain}:{port}/{version}".format(**api)
- session = requests.session()
- uname, A = usr.start_authentication()
- params = {
- 'login': uname,
- 'A': binascii.hexlify(A)
- }
- response = session.post(api_url + '/sessions', data=params, verify=False)
- init = response.json()
- if ('errors' in init):
- 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,
- data={'client_auth': binascii.hexlify(M)})
- return response.json()
+class Config():
+ def __init__(self, filename="/etc/leap/hiera.yaml"):
+ with open("/etc/leap/hiera.yaml", 'r') as stream:
+ config = yaml.load(stream)
+ self.user = config['webapp']['nagios_test_user']
+ if 'username' not in self.user:
+ raise Exception('nagios test user lacks username')
+ if 'password' not in self.user:
+ raise Exception('nagios test user lacks password')
+ self.api = config['api']
+ self.api['version'] = config['webapp']['api_version']
+
+class Api():
+ def __init__(self, config, verify=True):
+ self.config = config.api
+ self.session = requests.session()
+ self.verify = verify
+
+ def api_url(self, path):
+ return self.api_root() + path
+
+ def api_root(self):
+ return "https://{domain}:{port}/{version}/".format(**self.config)
+
+ def get(self, path, **args):
+ response = self.session.get(self.api_url(path),
+ verify=self.verify,
+ **args)
+ return response.json()
+
+ def post(self, path, **args):
+ response = self.session.post(self.api_url(path),
+ verify=self.verify,
+ **args)
+ return response.json()
+
+ def put(self, path, **args):
+ response = self.session.put(self.api_url(path),
+ verify=self.verify,
+ **args)
+ return response.json()
+
+class User():
+ def __init__(self, config):
+ self.config = config.user
+ self.srp_user = srp.User(self.config['username'], self.config['password'], srp.SHA256, srp.NG_1024)
+
+ def login(self, api):
+ init=self.init_authentication(api)
+ if ('errors' in init):
+ raise Exception('test user not found')
+ auth=self.authenticate(api, init)
+ if ('errors' in auth):
+ raise Exception('srp password auth failed')
+ self.verify_server(auth)
+ if not self.is_authenticated():
+ raise Exception('user is not authenticated')
+
+ def init_authentication(self, api):
+ uname, A = self.srp_user.start_authentication()
+ params = {
+ 'login': uname,
+ 'A': binascii.hexlify(A)
+ }
+ return api.post('sessions', data=params)
+
+ def authenticate(self, api, init):
+ M = self.srp_user.process_challenge(
+ safe_unhexlify(init['salt']), safe_unhexlify(init['B']))
+ auth = api.put('sessions/' + self.config["username"],
+ data={'client_auth': binascii.hexlify(M)})
+ return auth
+
+ def verify_server(self, auth):
+ self.srp_user.verify_session(safe_unhexlify(auth["M2"]))
+
+ def is_authenticated(self):
+ return self.srp_user.authenticated()
+
+
+def login_successfully():
+ config = Config()
+ user = User(config)
+ api = Api(config, verify=False)
+ user.login(api)
if __name__ == '__main__':
import nagios_test