summaryrefslogtreecommitdiff
path: root/test/nagios/support
diff options
context:
space:
mode:
authorazul <azul@riseup.net>2014-04-18 14:34:32 +0200
committerazul <azul@riseup.net>2014-04-18 14:34:32 +0200
commit8078cf0f853310cfb0d0681db45effb96850efc9 (patch)
tree86fb496091a772cefd0e07d1558f6b7fe0636963 /test/nagios/support
parent3513ad74f950b113af1ba1e3d06bc6a55c48fde5 (diff)
parentd639e0a48599b30777b80c2809ded1efb3a6d926 (diff)
Merge pull request #147 from azul/test/nagios-test-signup0.5.1-rc
Test/nagios test signup
Diffstat (limited to 'test/nagios/support')
-rw-r--r--test/nagios/support/__init__.py0
-rw-r--r--test/nagios/support/api.py39
-rw-r--r--test/nagios/support/config.py14
-rw-r--r--test/nagios/support/nagios_report.py24
-rw-r--r--test/nagios/support/nagios_test.py49
-rw-r--r--test/nagios/support/user.py64
6 files changed, 190 insertions, 0 deletions
diff --git a/test/nagios/support/__init__.py b/test/nagios/support/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/test/nagios/support/__init__.py
diff --git a/test/nagios/support/api.py b/test/nagios/support/api.py
new file mode 100644
index 0000000..ec1af99
--- /dev/null
+++ b/test/nagios/support/api.py
@@ -0,0 +1,39 @@
+import requests
+import json
+
+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 self.parse_json(response)
+
+ def post(self, path, **args):
+ response = self.session.post(self.api_url(path),
+ verify=self.verify,
+ **args)
+ return self.parse_json(response)
+
+ def put(self, path, **args):
+ response = self.session.put(self.api_url(path),
+ verify=self.verify,
+ **args)
+ return self.parse_json(response)
+
+ def parse_json(self, response):
+ try:
+ return response.json()
+ except TypeError:
+ return response.json # older versions of requests
+
diff --git a/test/nagios/support/config.py b/test/nagios/support/config.py
new file mode 100644
index 0000000..afb4464
--- /dev/null
+++ b/test/nagios/support/config.py
@@ -0,0 +1,14 @@
+import yaml
+
+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']
+
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/support/user.py b/test/nagios/support/user.py
new file mode 100644
index 0000000..9bf1d0a
--- /dev/null
+++ b/test/nagios/support/user.py
@@ -0,0 +1,64 @@
+import srp._pysrp as srp
+import binascii
+import string
+import random
+
+safe_unhexlify = lambda x: binascii.unhexlify(x) if (
+ len(x) % 2 == 0) else binascii.unhexlify('0' + x)
+
+# let's have some random name and password
+def id_generator(size=6, chars=string.ascii_lowercase + string.digits):
+ return ''.join(random.choice(chars) for x in range(size))
+
+class User():
+ def __init__(self, config = None):
+ if config and config.user:
+ self.username = config.user["username"]
+ self.password = config.user["password"]
+ else:
+ self.username = 'test_' + id_generator()
+ self.password = id_generator() + id_generator()
+ self.srp_user = srp.User(self.username, self.password, srp.SHA256, srp.NG_1024)
+
+ def signup(self, api):
+ salt, vkey = srp.create_salted_verification_key( self.username, self.password, srp.SHA256, srp.NG_1024 )
+ user_params = {
+ 'user[login]': self.username,
+ 'user[password_verifier]': binascii.hexlify(vkey),
+ 'user[password_salt]': binascii.hexlify(salt)
+ }
+ return api.post('users.json', data = user_params)
+
+ 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')
+ return auth
+
+ 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.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()
+