Synnefo » snf-ganeti

#!/usr/bin/python

#

# Copyright (C) 2011 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Script for testing ganeti.utils.log"""

import os

import unittest

import logging

import tempfile

import shutil

import threading

from cStringIO import StringIO

from ganeti import constants

from ganeti import errors

from ganeti import compat

from ganeti import utils

import testutils

class TestLogHandler(unittest.TestCase):

  def testNormal(self):

    tmpfile = tempfile.NamedTemporaryFile()

    handler = utils.log._ReopenableLogHandler(tmpfile.name)

    handler.setFormatter(logging.Formatter("%(asctime)s: %(message)s"))

    logger = logging.Logger("TestLogger")

    logger.addHandler(handler)

    self.assertEqual(len(logger.handlers), 1)

    logger.error("Test message ERROR")

    logger.info("Test message INFO")

    logger.removeHandler(handler)

    self.assertFalse(logger.handlers)

    handler.close()

    self.assertEqual(len(utils.ReadFile(tmpfile.name).splitlines()), 2)

  def testReopen(self):

    tmpfile = tempfile.NamedTemporaryFile()

    tmpfile2 = tempfile.NamedTemporaryFile()

    handler = utils.log._ReopenableLogHandler(tmpfile.name)

    self.assertFalse(utils.ReadFile(tmpfile.name))

    self.assertFalse(utils.ReadFile(tmpfile2.name))

    logger = logging.Logger("TestLoggerReopen")

    logger.addHandler(handler)

    for _ in range(3):

      logger.error("Test message ERROR")

    handler.flush()

    self.assertEqual(len(utils.ReadFile(tmpfile.name).splitlines()), 3)

    before_id = utils.GetFileID(tmpfile.name)

    handler.RequestReopen()

    self.assertTrue(handler._reopen)

    self.assertTrue(utils.VerifyFileID(utils.GetFileID(tmpfile.name),

                                       before_id))

    # Rename only after requesting reopen

    os.rename(tmpfile.name, tmpfile2.name)

    assert not os.path.exists(tmpfile.name)

    # Write another message, should reopen

    for _ in range(4):

      logger.info("Test message INFO")

      # Flag must be reset

      self.assertFalse(handler._reopen)

      self.assertFalse(utils.VerifyFileID(utils.GetFileID(tmpfile.name),

                                          before_id))

    logger.removeHandler(handler)

    self.assertFalse(logger.handlers)

    handler.close()

    self.assertEqual(len(utils.ReadFile(tmpfile.name).splitlines()), 4)

    self.assertEqual(len(utils.ReadFile(tmpfile2.name).splitlines()), 3)

  def testConsole(self):

    for (console, check) in [(None, False),

                             (tempfile.NamedTemporaryFile(), True),

                             (self._FailingFile(os.devnull), False)]:

      # Create a handler which will fail when handling errors

      cls = utils.log._LogErrorsToConsole(self._FailingHandler)

      # Instantiate handler with file which will fail when writing,

      # provoking a write to the console

      handler = cls(console, self._FailingFile(os.devnull))

      logger = logging.Logger("TestLogger")

      logger.addHandler(handler)

      self.assertEqual(len(logger.handlers), 1)

      # Provoke write

      logger.error("Test message ERROR")

      # Take everything apart

      logger.removeHandler(handler)

      self.assertFalse(logger.handlers)

      handler.close()

      if console and check:

        console.flush()

        # Check console output

        consout = utils.ReadFile(console.name)

        self.assertTrue("Cannot log message" in consout)

        self.assertTrue("Test message ERROR" in consout)

  class _FailingFile(file):

    def write(self, _):

      raise Exception

  class _FailingHandler(logging.StreamHandler):

    def handleError(self, _):

      raise Exception

class TestSetupLogging(unittest.TestCase):

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

  def tearDown(self):

    shutil.rmtree(self.tmpdir)

  def testSimple(self):

    logfile = utils.PathJoin(self.tmpdir, "basic.log")

    logger = logging.Logger("TestLogger")

    self.assertTrue(callable(utils.SetupLogging(logfile, "test",

                                                console_logging=False,

                                                syslog=constants.SYSLOG_NO,

                                                stderr_logging=False,

                                                multithreaded=False,

                                                root_logger=logger)))

    self.assertEqual(utils.ReadFile(logfile), "")

    logger.error("This is a test")

    # Ensure SetupLogging used custom logger

    logging.error("This message should not show up in the test log file")

    self.assertTrue(utils.ReadFile(logfile).endswith("This is a test\n"))

  def testReopen(self):

    logfile = utils.PathJoin(self.tmpdir, "reopen.log")

    logfile2 = utils.PathJoin(self.tmpdir, "reopen.log.OLD")

    logger = logging.Logger("TestLogger")

    reopen_fn = utils.SetupLogging(logfile, "test",

                                   console_logging=False,

                                   syslog=constants.SYSLOG_NO,

                                   stderr_logging=False,

                                   multithreaded=False,

                                   root_logger=logger)

    self.assertTrue(callable(reopen_fn))

    self.assertEqual(utils.ReadFile(logfile), "")

    logger.error("This is a test")

    self.assertTrue(utils.ReadFile(logfile).endswith("This is a test\n"))

    os.rename(logfile, logfile2)

    assert not os.path.exists(logfile)

    # Notify logger to reopen on the next message

    reopen_fn()

    assert not os.path.exists(logfile)

    # Provoke actual reopen

    logger.error("First message")

    self.assertTrue(utils.ReadFile(logfile).endswith("First message\n"))

    self.assertTrue(utils.ReadFile(logfile2).endswith("This is a test\n"))

class TestSetupToolLogging(unittest.TestCase):

  def test(self):

    error_name = logging.getLevelName(logging.ERROR)

    warn_name = logging.getLevelName(logging.WARNING)

    info_name = logging.getLevelName(logging.INFO)

    debug_name = logging.getLevelName(logging.DEBUG)

    for debug in [False, True]:

      for verbose in [False, True]:

        logger = logging.Logger("TestLogger")

        buf = StringIO()

        utils.SetupToolLogging(debug, verbose, _root_logger=logger, _stream=buf)

        logger.error("level=error")

        logger.warning("level=warning")

        logger.info("level=info")

        logger.debug("level=debug")

        lines = buf.getvalue().splitlines()

        self.assertTrue(compat.all(line.count(":") == 3 for line in lines))

        messages = [line.split(":", 3)[-1].strip() for line in lines]

        if debug:

          self.assertEqual(messages, [

            "%s level=error" % error_name,

            "%s level=warning" % warn_name,

            "%s level=info" % info_name,

            "%s level=debug" % debug_name,

            ])

        elif verbose:

          self.assertEqual(messages, [

            "%s level=error" % error_name,

            "%s level=warning" % warn_name,

            "%s level=info" % info_name,

            ])

        else:

          self.assertEqual(messages, [

            "level=error",

            "level=warning",

            ])

  def testThreadName(self):

    thread_name = threading.currentThread().getName()

    for enable_threadname in [False, True]:

      logger = logging.Logger("TestLogger")

      buf = StringIO()

      utils.SetupToolLogging(True, True, threadname=enable_threadname,

                             _root_logger=logger, _stream=buf)

      logger.debug("test134042376")

      lines = buf.getvalue().splitlines()

      self.assertEqual(len(lines), 1)

      if enable_threadname:

        self.assertTrue((" %s " % thread_name) in lines[0])

      else:

        self.assertTrue(thread_name not in lines[0])

if __name__ == "__main__":

  testutils.GanetiTestProgram()