nagios test: refactor webapp_login with classes

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