184 lines
5.5 KiB
Python
184 lines
5.5 KiB
Python
'''
|
|
manages logging events of the CryptoBox
|
|
'''
|
|
import sys
|
|
import os
|
|
import unittest
|
|
|
|
class CryptoBoxLogger:
|
|
'''
|
|
handles logging events and prints them e.g. to a logfile
|
|
'''
|
|
|
|
DebugLevels = {"debug":0, "info":3, "warn":6, "error":9}
|
|
DebugFacilities = {"file":0}
|
|
|
|
def __init__(self, level, facility, destination=None):
|
|
"""create a CryptoBoxLogger object and connect it to an output facility
|
|
|
|
level: an integer (0..9) or string ('debug', 'info', 'warn' or 'error')
|
|
facility: for now only 'file'
|
|
destination: e.g. the name of the logfile or syslog facility
|
|
|
|
"""
|
|
try:
|
|
try:
|
|
facility = int(facility)
|
|
except Exception:
|
|
try:
|
|
facility = self.DebugFacilities[facility]
|
|
except KeyError:
|
|
raise "LoggerError"
|
|
if (facility != 0): raise "LoggerError"
|
|
except "LoggerError":
|
|
errorMsg = "Invalid debug facility: %s" % facility
|
|
sys.stderr.write(errorMsg + "\n")
|
|
raise "LoggerError", errorMsg
|
|
try:
|
|
try:
|
|
level = int(level)
|
|
except Exception:
|
|
try:
|
|
level = self.DebugLevels[level]
|
|
except KeyError:
|
|
raise "LoggerError"
|
|
if (level < 0) or (level > 9): raise "LoggerError"
|
|
except "LoggerError":
|
|
errorMsg = "Invalid debug level: %s" % level
|
|
sys.stderr.write(errorMsg + "\n")
|
|
raise "LoggerError", errorMsg
|
|
self.debug_level = level
|
|
if facility == self.DebugFacilities["file"]:
|
|
self.logFunc = self.message2file
|
|
if destination is not None:
|
|
self.logFile = destination
|
|
else:
|
|
self.logFile = '/var/log/cryptobox.log'
|
|
try:
|
|
fsock = open(self.logFile, "a")
|
|
fsock.close()
|
|
except IOError:
|
|
errorMsg ="Unable to open logfile (%s) for writing." % (self.logFile,)
|
|
sys.stderr.write(errorMsg + "\n")
|
|
raise "LoggerError", errorMsg
|
|
|
|
else:
|
|
errorMsg = "Invalid logging facility: %d." % (facility, )
|
|
sys.stderr.write(errorMsg + "\n")
|
|
raise "LoggerError", errorMsg
|
|
|
|
|
|
def printMessage(self, msg_level, text):
|
|
if msg_level is None: msg_level = self.DebugLevels["debug"]
|
|
"convert debuglevel from string to int, if necessary"
|
|
try:
|
|
msg_level = int(msg_level)
|
|
except ValueError:
|
|
try:
|
|
msg_level = self.DebugLevels[msg_level]
|
|
except KeyError:
|
|
errorMsg = "Invalid debug level: %s" % msg_level
|
|
sys.stderr.write(errorMsg + "\n")
|
|
raise "LoggerError", errorMsg
|
|
if (msg_level < 0) or (msg_level > 9):
|
|
errorMsg = "Invalid debug level: %s" % msg_level
|
|
sys.stderr.write(errorMsg + "\n")
|
|
raise "LoggerError", errorMsg
|
|
if text is None:
|
|
errorMsg = "Empty debug message - this is not allowed"
|
|
sys.stderr.write(errorMsg + "\n")
|
|
raise "LoggerError", errorMsg
|
|
if msg_level >= self.debug_level:
|
|
self.logFunc("[CryptoBox] - %s\n" % (text, ))
|
|
|
|
|
|
def message2file(self, text):
|
|
try:
|
|
log_sock = open(self.logFile, "a")
|
|
try:
|
|
log_sock.writelines(text)
|
|
log_sock.close()
|
|
return
|
|
except IOError:
|
|
errorMsg = "Unable to write messages to logfile (%s)." % (self.logFile, )
|
|
sys.stderr.write(errorMsg + "\n")
|
|
raise "LoggerError", errorMsg
|
|
except IOError:
|
|
errorMsg = "Unable to open logfile (%s) for writing." % (self.logFile, )
|
|
sys.stderr.write(errorMsg + "\n")
|
|
raise "LoggerError", errorMsg
|
|
|
|
|
|
# ********************* test class **********************
|
|
|
|
class CryptoBoxLoggerTest(unittest.TestCase):
|
|
|
|
logFile = "/tmp/cbox-test.log"
|
|
|
|
def setUp(self):
|
|
if os.path.exists(self.logFile): os.remove(self.logFile)
|
|
|
|
def tearDown(self):
|
|
if os.path.exists(self.logFile): os.remove(self.logFile)
|
|
|
|
def testInit(self):
|
|
try:
|
|
CryptoBoxLogger(0, 0)
|
|
except "LoggerError":
|
|
CryptoBoxLogger(0, 0, self.logFile)
|
|
os.remove(self.logFile)
|
|
CryptoBoxLogger("info", "file", self.logFile)
|
|
self.assertRaises("LoggerError", CryptoBoxLogger, "invalid", 0, self.logFile)
|
|
self.assertRaises("LoggerError", CryptoBoxLogger, 0, "invalid", self.logFile)
|
|
self.assertRaises("LoggerError", CryptoBoxLogger, 10, 0, self.logFile)
|
|
self.assertRaises("LoggerError", CryptoBoxLogger, -1, 0, self.logFile)
|
|
self.assertRaises("LoggerError", CryptoBoxLogger, 0, 10, self.logFile)
|
|
self.assertRaises("LoggerError", CryptoBoxLogger, 0, -1, self.logFile)
|
|
self.assertRaises("LoggerError", CryptoBoxLogger, None, 0, self.logFile)
|
|
self.assertRaises("LoggerError", CryptoBoxLogger, 0, None, self.logFile)
|
|
self.assertRaises("LoggerError", CryptoBoxLogger, 0, 0, "/no/existing/path")
|
|
os.remove(self.logFile)
|
|
|
|
def testMessage(self):
|
|
cb = CryptoBoxLogger(3, 0, self.logFile)
|
|
content1 = self.readFile()
|
|
self.assertEquals(content1, "")
|
|
cb.printMessage(3, "Ausgabe")
|
|
content2 = self.readFile()
|
|
self.assertNotEqual(content1, content2)
|
|
self.assertRaises("LoggerError", cb.printMessage, 10, "Ausgabe")
|
|
self.assertEquals(content2, self.readFile())
|
|
self.assertRaises("LoggerError", cb.printMessage, -1, "Ausgabe")
|
|
self.assertEquals(content2, self.readFile())
|
|
self.assertRaises("LoggerError", cb.printMessage, "invalid", "Ausgabe")
|
|
self.assertEquals(content2, self.readFile())
|
|
self.assertRaises("LoggerError", cb.printMessage, 3, None)
|
|
self.assertEquals(content2, self.readFile())
|
|
cb.printMessage(2, "Ausgabe")
|
|
self.assertEquals(content2, self.readFile())
|
|
cb.printMessage(4, "Ausgabe")
|
|
self.assertNotEqual(content2, self.readFile())
|
|
os.remove(self.logFile)
|
|
|
|
def readFile(self):
|
|
fd = None
|
|
try:
|
|
fd = open(self.logFile, "r")
|
|
text = fd.read()
|
|
fd.close()
|
|
except IOError:
|
|
if fd is not None: fd.close()
|
|
text = None
|
|
return text
|
|
|
|
|
|
# *************** unit testing *********************
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
devnull = open(os.devnull, "w")
|
|
sys.stderr = devnull
|
|
except IOError:
|
|
pass
|
|
unittest.main()
|
|
|