Skip several test-cases of systemd backend, if journal seems to be not available (e. g. no rights to read journal);

Closes gh-2100
pull/2090/head^2
sebres 2018-04-03 12:39:37 +02:00
parent fd0471927d
commit 50d7c649ba
1 changed files with 21 additions and 14 deletions

View File

@ -1145,6 +1145,7 @@ def get_monitor_failures_journal_testcase(Filter_): # pragma: systemd no cover
def setUp(self):
"""Call before every test case."""
super(MonitorJournalFailures, self).setUp()
self._runtimeJournal = None
self.test_file = os.path.join(TEST_FILES_DIR, "testcase-journal.log")
self.jail = DummyJail()
self.filter = None
@ -1156,6 +1157,7 @@ def get_monitor_failures_journal_testcase(Filter_): # pragma: systemd no cover
'TEST_FIELD': "1", 'TEST_UUID': self.test_uuid}
def _initFilter(self, **kwargs):
self._getRuntimeJournal() # check journal available
self.filter = Filter_(self.jail, **kwargs)
self.filter.addJournalMatch([
"SYSLOG_IDENTIFIER=fail2ban-testcases",
@ -1176,21 +1178,26 @@ def get_monitor_failures_journal_testcase(Filter_): # pragma: systemd no cover
def _getRuntimeJournal(self):
"""Retrieve current system journal path
If none found, None will be returned
If not found, SkipTest exception will be raised.
"""
# Depending on the system, it could be found under /run or /var/log (e.g. Debian)
# which are pointed by different systemd-path variables. We will
# check one at at time until the first hit
for systemd_var in 'system-runtime-logs', 'system-state-logs':
tmp = Utils.executeCmd(
'find "$(systemd-path %s)" -name system.journal' % systemd_var,
timeout=10, shell=True, output=True
)
self.assertTrue(tmp)
out = str(tmp[1].decode('utf-8')).split('\n')[0]
if out:
return out
# we can cache it:
if self._runtimeJournal is None:
# Depending on the system, it could be found under /run or /var/log (e.g. Debian)
# which are pointed by different systemd-path variables. We will
# check one at at time until the first hit
for systemd_var in 'system-runtime-logs', 'system-state-logs':
tmp = Utils.executeCmd(
'find "$(systemd-path %s)" -name system.journal' % systemd_var,
timeout=10, shell=True, output=True
)
self.assertTrue(tmp)
out = str(tmp[1].decode('utf-8')).split('\n')[0]
if out: break
self._runtimeJournal = out
if self._runtimeJournal:
return self._runtimeJournal
raise unittest.SkipTest('systemd journal seems to be not available (e. g. no rights to read)')
def testJournalFilesArg(self):
# retrieve current system journal path
jrnlfile = self._getRuntimeJournal()