summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--test/nagios/nagios_report.py24
-rw-r--r--test/nagios/nagios_test.py49
-rw-r--r--test/nagios/support/__init__.py0
-rw-r--r--test/nagios/support/api.py33
-rw-r--r--test/nagios/support/config.py14
-rw-r--r--test/nagios/support/user.py43
-rwxr-xr-xtest/nagios/webapp_login.py93
7 files changed, 94 insertions, 162 deletions
diff --git a/test/nagios/nagios_report.py b/test/nagios/nagios_report.py
deleted file mode 100644
index 13cd551..0000000
--- a/test/nagios/nagios_report.py
+++ /dev/null
@@ -1,24 +0,0 @@
-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
deleted file mode 100644
index 3eb8d55..0000000
--- a/test/nagios/nagios_test.py
+++ /dev/null
@@ -1,49 +0,0 @@
-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/__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..3b6a90f
--- /dev/null
+++ b/test/nagios/support/api.py
@@ -0,0 +1,33 @@
+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 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()
+
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/user.py b/test/nagios/support/user.py
new file mode 100644
index 0000000..8e49c4b
--- /dev/null
+++ b/test/nagios/support/user.py
@@ -0,0 +1,43 @@
+import srp._pysrp as srp
+import binascii
+
+safe_unhexlify = lambda x: binascii.unhexlify(x) if (
+ len(x) % 2 == 0) else binascii.unhexlify('0' + x)
+
+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()
+
diff --git a/test/nagios/webapp_login.py b/test/nagios/webapp_login.py
index 7e2efd7..4e78836 100755
--- a/test/nagios/webapp_login.py
+++ b/test/nagios/webapp_login.py
@@ -2,96 +2,11 @@
# Test Authentication with the webapp API works.
-import requests
-import json
import string
import random
-import srp._pysrp as srp
-import binascii
-import yaml
-
-safe_unhexlify = lambda x: binascii.unhexlify(x) if (
- len(x) % 2 == 0) else binascii.unhexlify('0' + x)
-
-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()
-
+from support.api import Api
+from support.config import Config
+from support.user import User
def login_successfully():
config = Config()
@@ -100,6 +15,6 @@ def login_successfully():
user.login(api)
if __name__ == '__main__':
- import nagios_test
+ from support import nagios_test
exit_code = nagios_test.run(login_successfully)
exit(exit_code)