Merge pull request #147 from azul/test/nagios-test-signup0.5.1-rc

Test/nagios test signup

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()
+