[svn] commit: r2963 - in /branches/trac267/src/lib/python/isc/log: log.py tests/log_test.py
BIND 10 source code commits
bind10-changes at lists.isc.org
Fri Sep 17 08:40:40 UTC 2010
Author: chenzhengzhang
Date: Fri Sep 17 08:40:40 2010
New Revision: 2963
Log:
update code and unittest accoring to Stephen's comments
Modified:
branches/trac267/src/lib/python/isc/log/log.py
branches/trac267/src/lib/python/isc/log/tests/log_test.py
Modified: branches/trac267/src/lib/python/isc/log/log.py
==============================================================================
--- branches/trac267/src/lib/python/isc/log/log.py (original)
+++ branches/trac267/src/lib/python/isc/log/log.py Fri Sep 17 08:40:40 2010
@@ -44,19 +44,33 @@
"""RotatingFileHandler: replace RotatingFileHandler with a custom handler"""
def __init__(self, filename, mode='a', maxBytes=0, backupCount=0, encoding=None, delay=0):
- dir = os.path.split(filename)
- # Create log directory beforehand, because the underlying logging framework won't
- # create non-exsiting log directory on writing logs.
- if not (os.path.exists(dir[0])):
- os.makedirs(dir[0])
- super(NSFileLogHandler, self).__init__(filename, mode, maxBytes,
+ abs_file_name = self._get_abs_file_path(filename)
+ """Create log directory beforehand, because the underlying logging framework won't
+ create non-exsiting log directory on writing logs.
+ """
+ if not (os.path.exists(os.path.dirname(abs_file_name))):
+ os.makedirs(os.path.dirname(abs_file_name))
+ super(NSFileLogHandler, self).__init__(abs_file_name, mode, maxBytes,
backupCount, encoding, delay)
def handleError(self, record):
"""Overwrite handleError to provide more user-friendly error messages"""
if logging.raiseExceptions:
ei = sys.exc_info()
- log_err("[b10-logging] ", str(ei[1]))
+ if (ei[1]):
+ sys.stderr.write("[b10-logging] : " + str(ei[1]))
+
+ def _get_abs_file_path(self, file_name):
+ """ Get absolute file path"""
+ # For a bare filename, log_dir will be set the current directory.
+ if not os.path.dirname(file_name):
+ abs_file_dir = os.getcwd()
+ else:
+ abs_file_dir = os.path.abspath(os.path.dirname(file_name))
+ abs_file_name = os.path.join(abs_file_dir, os.path.basename(file_name))
+
+ return abs_file_name
+
def shouldRollover(self, record):
"""Rewrite RotatingFileHandler.shouldRollover.
@@ -66,23 +80,27 @@
dfn = self.baseFilename
if (self.stream) and (not os.path.exists(dfn)): #Does log file exist?
self.stream = None
- # Log directory may be deleted while bind10 running or updated with a
- # non-existing directory.
- dir = os.path.split(dfn)
- if not (os.path.exists(dir[0])): #Does log subdirectory exist?
- os.makedirs(dir[0])
+ """ Log directory may be deleted while bind10 running or updated with a
+ non-existing directory. Need to create log directory beforehand, because
+ the underlying logging framework won't create non-exsiting log directory
+ on writing logs.
+ """
+ if not (os.path.exists(os.path.dirname(dfn))): #Does log subdirectory exist?
+ os.makedirs(os.path.dirname(dfn))
self.stream = self._open()
return super(NSFileLogHandler, self).shouldRollover(record)
def update_config(self, file_name, backup_count, max_bytes):
"""Update RotatingFileHandler configuration.
+ Changes will be picked up in the next call to shouldRollover().
input:
log file name
max backup count
predetermined log file size
"""
- self.baseFilename = file_name
+ abs_file_name = self._get_abs_file_path(file_name)
+ self.baseFilename = abs_file_name
self.maxBytes = max_bytes
self.backupCount = backup_count
Modified: branches/trac267/src/lib/python/isc/log/tests/log_test.py
==============================================================================
--- branches/trac267/src/lib/python/isc/log/tests/log_test.py (original)
+++ branches/trac267/src/lib/python/isc/log/tests/log_test.py Fri Sep 17 08:40:40 2010
@@ -1,3 +1,22 @@
+# Copyright (C) 2010 Internet Systems Consortium.
+#
+# Permission to use, copy, modify, and distribute this software for any
+# purpose with or without fee is hereby granted, provided that the above
+# copyright notice and this permission notice appear in all copies.
+#
+# THE SOFTWARE IS PROVIDED "AS IS" AND INTERNET SYSTEMS CONSORTIUM
+# DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
+# INTERNET SYSTEMS CONSORTIUM BE LIABLE FOR ANY SPECIAL, DIRECT,
+# INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING
+# FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
+# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
+# WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+#
+# Tests for the python logging module
+#
+
from isc.log.log import *
import unittest
import os
@@ -20,8 +39,6 @@
self.handler = NSFileLogHandler(filename = self.FILE_LOG1.name,
maxBytes = 1024,
backupCount = 5)
- self.stderr_bak = sys.stderr
- sys.stderr = open(os.devnull, 'w')
def test_shouldRollover(self):
if(os.path.exists(self.FILE_LOG1.name)):
@@ -30,15 +47,48 @@
self.handler.shouldRollover(record)
self.assertTrue(os.path.exists(self.FILE_LOG1.name))
+ def test_get_absolute_file_path(self):
+ abs_file_name = self.handler._get_abs_file_path(self.FILE_LOG1.name)
+ self.assertEqual(abs_file_name, self.FILE_LOG1.name)
+ # test bare filename
+ file_name1 = "bind10.py"
+ abs_file_name = self.handler._get_abs_file_path(file_name1)
+ self.assertEqual(abs_file_name, os.path.join(os.getcwd(), file_name1))
+ # test relative path
+ file_name2 = "./bind10.py"
+ abs_file_name = self.handler._get_abs_file_path(file_name2)
+ self.assertEqual(abs_file_name, os.path.join(os.getcwd(), os.path.basename(file_name2)))
+
def test_update_config(self):
self.handler.update_config(self.FILE_LOG2.name, 3, 512)
self.assertEqual(self.handler.baseFilename, self.FILE_LOG2.name)
self.assertEqual(self.handler.maxBytes, 512)
self.assertEqual(self.handler.backupCount, 3)
+ # check the existence new log file.
+ # emit() will call shouldRollover() to update the log file
+ if(os.path.exists(self.FILE_LOG2.name)):
+ os.remove(self.FILE_LOG2.name)
+ record = logging.LogRecord(None, None, "", 0, "rotate file handler", (), None, None)
+ self.handler.emit(record)
+ self.assertTrue(os.path.exists(self.FILE_LOG2.name))
+
def test_handle_Error(self):
- # handle_Error() just print out error message
- pass
+ if(os.path.exists(self.FILE_LOG3.name)):
+ os.remove(self.FILE_LOG3.name)
+ # redirect error message to file
+ savederr = sys.stderr
+ errfd = open(self.FILE_LOG3.name, 'w+')
+ sys.stderr = errfd
+ record = logging.LogRecord(None, None, "", 0, "record message", (), None, None)
+ try:
+ raise ValueError("ValueError")
+ except ValueError:
+ self.handler.handleError(record)
+
+ self.assertEqual("[b10-logging] : ValueError", errfd.read())
+ sys.stderr = savederr
+ errfd.close()
def tearDown(self):
self.handler.flush()
@@ -46,9 +96,6 @@
self.FILE_LOG1.close()
self.FILE_LOG2.close()
self.FILE_LOG3.close()
- sys.stderr.flush()
- sys.stderr = self.stderr_bak
-
class TestSysLogHandler(unittest.TestCase):
def setUp(self):
@@ -59,9 +106,9 @@
self.assertEqual(sysLevel, syslog.LOG_ERR)
def test_emit(self):
- record = logging.LogRecord(None, None, "", 0, "syslog handler", (), None, None)
+ syslog_message = "bind10 syslog testing"
+ record = logging.LogRecord(None, None, "", 0, syslog_message, (), None, None)
self.handler.emit(record)
-
class TestLogging(unittest.TestCase):
@@ -87,24 +134,18 @@
self.assertNotEqual(self.file_stream_logger._stream_handler, None)
self.assertEqual(self.file_stream_logger._syslog_handler, None)
- ret = self.file_stream_logger._file_handler in self.file_stream_logger.handlers
- self.assertTrue(ret)
- ret = self.file_stream_logger._stream_handler in self.file_stream_logger.handlers
- self.assertTrue(ret)
- ret = self.file_stream_logger._syslog_handler in self.file_stream_logger.handlers
- self.assertFalse(ret)
+ self.assertIn(self.file_stream_logger._file_handler, self.file_stream_logger.handlers)
+ self.assertIn(self.file_stream_logger._stream_handler, self.file_stream_logger.handlers)
+ self.assertNotIn(self.file_stream_logger._syslog_handler, self.file_stream_logger.handlers)
logLevel = LEVELS.get('debug', logging.NOTSET)
self.assertEqual(self.file_stream_logger.getEffectiveLevel(), logLevel)
self.assertEqual(self.syslog_logger._file_handler, None)
self.assertEqual(self.syslog_logger._stream_handler, None)
self.assertNotEqual(self.syslog_logger._syslog_handler, None)
- ret = self.syslog_logger._file_handler in self.syslog_logger.handlers
- self.assertFalse(ret)
- ret = self.syslog_logger._stream_handler in self.syslog_logger.handlers
- self.assertFalse(ret)
- ret = self.syslog_logger._syslog_handler in self.syslog_logger.handlers
- self.assertTrue(ret)
+ self.assertNotIn(self.syslog_logger._file_handler, self.syslog_logger.handlers)
+ self.assertNotIn(self.syslog_logger._stream_handler, self.syslog_logger.handlers)
+ self.assertIn(self.syslog_logger._syslog_handler, self.syslog_logger.handlers)
logLevel = LEVELS.get('info', logging.NOTSET)
self.assertEqual(self.syslog_logger.getEffectiveLevel(), logLevel)
@@ -114,48 +155,40 @@
self.syslog_logger.removeHandler(self.syslog_logger._file_handler)
self.syslog_logger._add_rotate_handler('', 5, 1024)
- ret = self.syslog_logger._file_handler in self.syslog_logger.handlers
- self.assertFalse(ret)
+ self.assertNotIn(self.syslog_logger._file_handler, self.syslog_logger.handlers)
self.syslog_logger._add_rotate_handler(self.FILE_STREAM_LOG1.name, 5, 1024)
- ret = self.syslog_logger._file_handler in self.syslog_logger.handlers
- self.assertTrue(ret)
+ self.assertIn(self.syslog_logger._file_handler, self.syslog_logger.handlers)
# test IOError exception
self.syslog_logger.removeHandler(self.syslog_logger._file_handler)
log_file = self.FILE_STREAM_LOG1.name + '/logfile'
self.syslog_logger._add_rotate_handler(log_file, 5, 1024)
- ret = self.syslog_logger._file_handler in self.syslog_logger.handlers
- self.assertFalse(ret)
+ self.assertNotIn(self.syslog_logger._file_handler, self.syslog_logger.handlers)
def test_add_stream_handler(self):
if(self.file_stream_logger._stream_handler in self.file_stream_logger.handlers):
self.file_stream_logger.removeHandler(self.file_stream_logger._stream_handler)
self.file_stream_logger._add_stream_handler()
- ret = self.file_stream_logger._stream_handler in self.file_stream_logger.handlers
- self.assertTrue(ret)
+ self.assertIn(self.file_stream_logger._stream_handler, self.file_stream_logger.handlers)
def test_add_syslog_handler(self):
if(self.syslog_logger._syslog_handler in self.syslog_logger.handlers):
self.syslog_logger.removeHandler(self.syslog_logger._syslog_handler)
self.syslog_logger._add_syslog_handler()
- ret = self.syslog_logger._syslog_handler in self.syslog_logger.handlers
- self.assertTrue(ret)
+ self.assertIn(self.syslog_logger._syslog_handler, self.syslog_logger.handlers)
def test_update_rotate_handler(self):
self.file_stream_logger._update_rotate_handler(self.FILE_STREAM_LOG2.name, 4, 1024)
- ret = self.file_stream_logger._file_handler in self.file_stream_logger.handlers
- self.assertTrue(ret)
+ self.assertIn(self.file_stream_logger._file_handler, self.file_stream_logger.handlers)
self.file_stream_logger._update_rotate_handler('', 5, 1024)
- ret = self.file_stream_logger._file_handler in self.file_stream_logger.handlers
- self.assertFalse(ret)
+ self.assertNotIn(self.file_stream_logger._file_handler, self.file_stream_logger.handlers)
self.file_stream_logger._update_rotate_handler(self.FILE_STREAM_LOG1.name, 4, 1024)
- ret = self.file_stream_logger._file_handler in self.file_stream_logger.handlers
- self.assertTrue(ret)
+ self.assertIn(self.file_stream_logger._file_handler, self.file_stream_logger.handlers)
def test_get_config(self):
config_data = {'log_file' : self.FILE_STREAM_LOG1.name,
More information about the bind10-changes
mailing list