[checkmk-commits] Add integration test to verify whole standard mail notification chain

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 &lt;lm(a)mathias-kettner.de&gt; 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(&quot;core&quot;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):