Module: check_mk
Branch: master
Commit: b2caeff85832f56d75e2546dfbc00ac96b59d39b
URL:
http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=b2caeff85832f5…
Author: Lars Michelsen <lm(a)mathias-kettner.de>
Date: Mon Dec 3 11:16:35 2018 +0100
Refactored HistoryLog to use testlib.WatchLog implementation
Change-Id: Ib45e00c46b462ec171b5959d95b1ee802ed32a5a
---
.../test_unreachable_notifications.py | 50 +++-------------------
tests/testlib/__init__.py | 10 +++--
2 files changed, 13 insertions(+), 47 deletions(-)
diff --git a/tests/integration/notifications/test_unreachable_notifications.py
b/tests/integration/notifications/test_unreachable_notifications.py
index a153ee9..200a404 100644
--- a/tests/integration/notifications/test_unreachable_notifications.py
+++ b/tests/integration/notifications/test_unreachable_notifications.py
@@ -5,7 +5,7 @@ import pytest
import time
import os
import sys
-from testlib import web
+from testlib import web, WatchLog
STATE_UP = 0
STATE_DOWN = 1
@@ -108,57 +108,19 @@ def initial_state(site, scenario):
time.sleep(1) # TODO: Add check for rotation
-# TODO: Use testlib.WatchLog()
-class HistoryLog(object):
+class HistoryLog(WatchLog):
def __init__(self, site, core):
- self._site = site
- self._core = core
- self._log = self._open_log()
- self._buf = []
+ super(HistoryLog, self).__init__(site, self._history_log_path(core),
default_timeout=10)
-
- def _log_path(self):
- if self._core == "cmc":
+ def _history_log_path(self, core):
+ if core == "cmc":
return "var/check_mk/core/history"
- elif self._core == "nagios":
+ elif core == "nagios":
return "var/nagios/nagios.log"
else:
raise NotImplementedError()
- 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=10):
- 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=10):
- 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)
-
- return False
-
def _send_child_down(scenario, site, log):
# - Set child down, expect DOWN notification
diff --git a/tests/testlib/__init__.py b/tests/testlib/__init__.py
index 529b6fe..39f078c 100644
--- a/tests/testlib/__init__.py
+++ b/tests/testlib/__init__.py
@@ -1742,10 +1742,11 @@ class CMKEventConsoleStatus(object):
class WatchLog(object):
"""Small helper for integration tests: Watch a sites log
file"""
- def __init__(self, site, rel_path):
+ def __init__(self, site, rel_path, default_timeout=5):
self._site = site
self._rel_path = rel_path
self._log = self._open_log()
+ self._default_timeout = default_timeout
self._buf = []
def _log_path(self):
@@ -1759,17 +1760,20 @@ class WatchLog(object):
fobj.seek(0, 2) # go to end of file
return fobj
- def check_logged(self, match_for, timeout=5):
+ def check_logged(self, match_for, timeout=None):
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):
+ def check_not_logged(self, match_for, timeout=None):
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):
+ if timeout is None:
+ timeout = self._default_timeout
+
timeout_at = time.time() + timeout
while time.time() < timeout_at:
#print "read till timeout %0.2f sec left" % (timeout_at -
time.time())