[checkmk-commits] add unit test for mk_logwatch.aix

Module: check_mk Branch: master Commit: e32db48e9fb4072aeda39b66dbe02a271a3bf004 URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=e32db48e9fb407… Author: Moritz Kiemer &lt;mo(a)mathias-kettner.de&gt; Date: Mon Dec 3 07:50:31 2018 +0100 add unit test for mk_logwatch.aix Change-Id: Id2394a33342abf60163757440b2b96fcd5bb0666 --- tests/unit/plugins/test_mk_logwatch_aix.py | 93 ++++++++++++++++++++++++++++++ 1 file changed, 93 insertions(+) diff --git a/tests/unit/plugins/test_mk_logwatch_aix.py b/tests/unit/plugins/test_mk_logwatch_aix.py new file mode 100644 index 0000000..2840f37 --- /dev/null +++ b/tests/unit/plugins/test_mk_logwatch_aix.py @@ -0,0 +1,93 @@ +import os +import subprocess +import pytest +from testlib import cmk_path + +PLUGIN = os.path.join(cmk_path(), 'agents', 'plugins', 'mk_logwatch.aix') + +ERRPT_OUTPUT = [ + 'IDENTIFIER TIMESTAMP T C RESOURCE_NAME DESCRIPTION', + '8650BE3F 0820122810 I H ent2 ETHERCHANNEL RECOVERY', + 'F3846E13 0820122510 P H ent2 ETHERCHANNEL FAILOVER', + '8650BE3F 0820104410 I H ent2 ETHERCHANNEL RECOVERY', + 'F3846E13 0820093810 P H ent2 ETHERCHANNEL FAILOVER', + '8650BE3F 0820090910 I H ent2 ETHERCHANNEL RECOVERY', +] + + +def _prepare_mock_errpt(tmpdir, errpt_output): + errpt_name = str(tmpdir.join('errpt')) + errpt_script = ''.join(['#!/bin/sh\n'] + ['echo "%s"\n' % line for line in errpt_output]) + with open(errpt_name, 'w') as errpt_file: + errpt_file.write(errpt_script) + os.chmod(errpt_name, 0777) + + +def _get_env(tmpdir): + env = os.environ.copy() + env.update({"PATH": '%s:%s' % (tmpdir, os.getenv("PATH")), "MK_VARDIR": "%s" % tmpdir}) + return env + + +class StateFile(object): + def __init__(self, tmpdir): + super(StateFile, self).__init__() + self.name = str(tmpdir.join('mk_logwatch_aix.last_reported')) + + def set(self, state): + if state is None: + try: + os.remove(self.name) + except OSError: + pass + return + with open(self.name, 'w') as statefile: + statefile.write("%s\n" % state) + + def content(self): + try: + content = open(self.name).read() + # ends with newline + assert content[-1] == '\n' + return content[:-1] + except IOError: + return None + + +def _format_expected(lines): + added_prefix = ['C %s\n' % line for line in lines] + added_header = ['<<<logwatch>>>\n', '[[[errorlog]]]\n'] + added_prefix + return ''.join(added_header) + + +(a)pytest.mark.parametrize( + "errpt_output,last_reported,expectations", + [ + ( + ERRPT_OUTPUT, + [None, "", ERRPT_OUTPUT[3], ERRPT_OUTPUT[1], "something else entirely"], + [ERRPT_OUTPUT[1:], ERRPT_OUTPUT[1:], ERRPT_OUTPUT[1:3], [], ERRPT_OUTPUT[1:]], + ), + ( + ERRPT_OUTPUT[:1], # no output, just header + [None, "", "what ever"], + [[], [], []], + ), + ]) +def test_mk_logwatch_aix(tmpdir, errpt_output, last_reported, expectations): + + _prepare_mock_errpt(tmpdir, errpt_output) + env = _get_env(tmpdir) + statefile = StateFile(tmpdir) + + for state, expected in zip(last_reported, expectations): + + statefile.set(state) + + output = subprocess.check_output(PLUGIN, env=env) + + assert output == _format_expected(expected) + if len(errpt_output) > 1: # we should have updated state file + assert statefile.content() == errpt_output[1] + else: # it should not have changed + assert statefile.content() == state