Module: check_mk
Branch: master
Commit: fdcaa633ddbb94b280de88e1fa6b6146223acf9c
URL:
http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=fdcaa633ddbb94…
Author: Lars Michelsen <lm(a)mathias-kettner.de>
Date: Thu May 24 20:30:08 2018 +0200
Add integration test to verify whole standard mail notification chain
Change-Id: I14f60033439759c7eb776d1e6a0a8accbc71dc9f
---
tests/integration/notifications/test_simple_rbn.py | 50 ++++++++++++++++++++++
.../test_unreachable_notifications.py | 1 +
tests/testlib/__init__.py | 46 ++++++++++++++++++++
3 files changed, 97 insertions(+)
diff --git a/tests/integration/notifications/test_simple_rbn.py
b/tests/integration/notifications/test_simple_rbn.py
new file mode 100644
index 0000000..6546b75
--- /dev/null
+++ b/tests/integration/notifications/test_simple_rbn.py
@@ -0,0 +1,50 @@
+import pytest
+import time
+
+from testlib import web, WatchLog
+
+(a)pytest.fixture()
+def test_config(web):
+ users = {
+ "hh": {
+ "alias": "Harry Hirsch",
+ "password": "1234",
+ "email": u"%s@localhost"% web.site.id,
+ 'contactgroups': ['all'],
+ },
+ }
+
+ expected_users = set(["cmkadmin", "automation"] + users.keys())
+ response = web.add_htpasswd_users(users)
+ all_users = web.get_all_users()
+ assert not expected_users - set(all_users.keys())
+
+ # Notify
+ web.add_host("notify-test", attributes={
+ "ipaddress": "127.0.0.1",
+ })
+ web.activate_changes()
+
+ yield
+
+ web.delete_host("notify-test")
+ web.delete_htpasswd_users(users.keys())
+ web.activate_changes()
+
+
+(a)pytest.mark.parametrize("core"core", [ "nagios", "cmc" ])
+def test_simple_rbn_notification(test_config, site, core):
+ site.set_config("CORE", core, with_restart=True)
+
+ # Open the log file and scan to end
+ l = WatchLog(site, "var/log/notify.log")
+
+ # Set object down to trigger a notification
+ site.send_host_check_result("notify-test", 1, "FAKE DOWN",
expected_state=1)
+
+ # Now check for appearing log lines - one after the other
+ l.check_logged("Got raw notification (notify-test)")
+ l.check_logged("notifying hh via mail")
+ l.check_logged("Creating spoolfile:")
+ l.check_logged("(notify-test) for local delivery")
+ l.check_logged("Output: Spooled mail to local mail transmission agent")
diff --git a/tests/integration/notifications/test_unreachable_notifications.py
b/tests/integration/notifications/test_unreachable_notifications.py
index 42c40e9..13f4ff8 100644
--- a/tests/integration/notifications/test_unreachable_notifications.py
+++ b/tests/integration/notifications/test_unreachable_notifications.py
@@ -108,6 +108,7 @@ def initial_state(site, scenario):
time.sleep(1) # TODO: Add check for rotation
+# TODO: Use testlib.WatchLog()
class HistoryLog(object):
def __init__(self, site, core):
self._site = site
diff --git a/tests/testlib/__init__.py b/tests/testlib/__init__.py
index 42bbd2f..01097d1 100644
--- a/tests/testlib/__init__.py
+++ b/tests/testlib/__init__.py
@@ -1681,6 +1681,52 @@ class CMKEventConsoleStatus(object):
+class WatchLog(object):
+ """Small helper for integration tests: Watch a sites log
file"""
+ def __init__(self, site, rel_path):
+ self._site = site
+ self._rel_path = rel_path
+ self._log = self._open_log()
+ self._buf = []
+
+
+ def _log_path(self):
+ return self._rel_path
+
+
+ def _open_log(self):
+ if not self._site.file_exists(self._log_path()):
+ self._site.write_file(self._log_path(), "")
+
+ fobj = open(self._site.path(self._log_path()), "r")
+ fobj.seek(0, 2) # go to end of file
+ return fobj
+
+
+ def check_logged(self, match_for, timeout=5):
+ if not self._check_for_line(match_for, timeout):
+ raise Exception("Did not find %r in %s after %d seconds" %
+ (match_for, self._log_path(), timeout))
+
+
+ def check_not_logged(self, match_for, timeout=5):
+ if self._check_for_line(match_for, timeout):
+ raise Exception("Found %r in %s after %d seconds" %
+ (match_for, self._log_path(), timeout))
+
+
+ def _check_for_line(self, match_for, timeout):
+ timeout_at = time.time() + timeout
+ while time.time() < timeout_at:
+ #print "read till timeout %0.2f sec left" % (timeout_at -
time.time())
+ line = self._log.readline()
+ sys.stdout.write(line)
+ if match_for in line:
+ return True
+ time.sleep(0.1)
+
+
+
@pytest.fixture(scope="module")
def web(site):