server/hosts/monitors/monitors_util_unittest.py

#!/usr/bin/python

import fcntl, os, signal, subprocess, StringIO
import tempfile, textwrap, time, unittest
import monitors_util


def InlineStringIO(text):
    return StringIO.StringIO(textwrap.dedent(text).strip())


class WriteLoglineTestCase(unittest.TestCase):
    def setUp(self):
        self.time_tuple = (2008, 10, 31, 18, 58, 17, 4, 305, 1)
        self.format = '[%Y-%m-%d %H:%M:%S]'
        self.formatted_time_tuple = '[2008-10-31 18:58:17]'
        self.msg = 'testing testing'

        # Stub out time.localtime()
        self.orig_localtime = time.localtime
        time.localtime = lambda: self.time_tuple


    def tearDown(self):
        time.localtime = self.orig_localtime


    def test_prepend_timestamp(self):
        timestamped = monitors_util.prepend_timestamp(
            self.msg, self.format)
        self.assertEquals(
            '%s\t%s' % (self.formatted_time_tuple, self.msg), timestamped)


    def test_write_logline_with_timestamp(self):
        logfile = StringIO.StringIO()
        monitors_util.write_logline(logfile, self.msg, self.format)
        logfile.seek(0)
        written = logfile.read()
        self.assertEquals(
            '%s\t%s\n' % (self.formatted_time_tuple, self.msg), written)


    def test_write_logline_without_timestamp(self):
        logfile = StringIO.StringIO()
        monitors_util.write_logline(logfile, self.msg)
        logfile.seek(0)
        written = logfile.read()
        self.assertEquals(
            '%s\n' % self.msg, written)


class AlertHooksTestCase(unittest.TestCase):
    def setUp(self):
        self.msg_template = 'alert yay %s haha %s'
        self.params = ('foo', 'bar')
        self.epoch_seconds = 1225501829.9300611
        # Stub out time.time
        self.orig_time = time.time
        time.time = lambda: self.epoch_seconds


    def tearDown(self):
        time.time = self.orig_time


    def test_make_alert(self):
        warnfile = StringIO.StringIO()
        alert = monitors_util.make_alert(warnfile, "MSGTYPE",
                                         self.msg_template)
        alert(*self.params)
        warnfile.seek(0)
        written = warnfile.read()
        ts = str(int(self.epoch_seconds))
        expected = '%s\tMSGTYPE\t%s\n' % (ts, self.msg_template % self.params)
        self.assertEquals(expected, written)


    def test_build_alert_hooks(self):
        warnfile = StringIO.StringIO()
        patterns_file = InlineStringIO("""
            BUG
            ^.*Kernel panic ?(.*)
            machine panic'd (%s)

            BUG
            ^.*Oops ?(.*)
            machine Oops'd (%s)
            """)
        hooks = monitors_util.build_alert_hooks(patterns_file, warnfile)
        self.assertEquals(len(hooks), 2)


class ProcessInputTestCase(unittest.TestCase):
    def test_process_input_simple(self):
        input = InlineStringIO("""
            woo yay
            this is a line
            booya
            """)
        logfile = StringIO.StringIO()
        monitors_util.process_input(input, logfile)
        input.seek(0)
        logfile.seek(0)

        self.assertEquals(
            '%s\n%s\n' % (input.read(), monitors_util.TERM_MSG),
            logfile.read())


class FollowFilesTestCase(unittest.TestCase):
    def setUp(self):
        self.logfile_dirpath = tempfile.mkdtemp()
        self.logfile_path = os.path.join(self.logfile_dirpath, 'messages')
        self.firstline = 'bip\n'
        self.lastline_seen = 'wooo\n'
        self.line_after_lastline_seen = 'yeah\n'
        self.lastline = 'pow\n'

        self.logfile = open(self.logfile_path, 'w')
        self.logfile.write(self.firstline)
        self.logfile.write(self.lastline_seen)
        self.logfile.write(self.line_after_lastline_seen)  # 3
        self.logfile.write('man\n')   # 2
        self.logfile.write(self.lastline)   # 1
        self.logfile.close()

        self.lastlines_dirpath = tempfile.mkdtemp()
        monitors_util.write_lastlines_file(
            self.lastlines_dirpath, self.logfile_path, self.lastline_seen)


    def test_lookup_lastlines(self):
        reverse_lineno = monitors_util.lookup_lastlines(
            self.lastlines_dirpath, self.logfile_path)
        self.assertEquals(reverse_lineno, 3)


    def test_nonblocking(self):
        po = subprocess.Popen('echo', stdout=subprocess.PIPE)
        flags = fcntl.fcntl(po.stdout, fcntl.F_GETFL)
        self.assertEquals(flags, 0)
        monitors_util.nonblocking(po.stdout)
        flags = fcntl.fcntl(po.stdout, fcntl.F_GETFL)
        self.assertEquals(flags, 2048)
        po.wait()


    def test_follow_files_nostate(self):
        follow_paths = [self.logfile_path]
        lastlines_dirpath = tempfile.mkdtemp()
        procs, pipes = monitors_util.launch_tails(
            follow_paths, lastlines_dirpath)
        lines, bad_pipes = monitors_util.poll_tail_pipes(
            pipes, lastlines_dirpath)
        first_shouldmatch = '[%s]\t%s' % (
            self.logfile_path, self.lastline)
        self.assertEquals(lines[0], first_shouldmatch)
        monitors_util.snuff(procs.values())


    def test_follow_files(self):
        follow_paths = [self.logfile_path]
        procs, pipes = monitors_util.launch_tails(
            follow_paths, self.lastlines_dirpath)
        lines, bad_pipes = monitors_util.poll_tail_pipes(
            pipes, self.lastlines_dirpath)
        first_shouldmatch = '[%s]\t%s' % (
            self.logfile_path, self.line_after_lastline_seen)
        self.assertEquals(lines[0], first_shouldmatch)
        monitors_util.snuff(procs.values())
        last_shouldmatch = '[%s]\t%s' % (self.logfile_path, self.lastline)
        self.assertEquals(lines[-1], last_shouldmatch)


if __name__ == '__main__':
    unittest.main()