Commit dbebf741 authored by Trevor Cappallo's avatar Trevor Cappallo
Browse files

exit handler test

parent b4f70c4b
......@@ -18,17 +18,23 @@ import json
import OperationalMail
def __exit_handler(logger, addresses, critical_addresses):
def __exit_handler(logger, notify):
"""Email given addresses if there are any buffered errors."""
if logger.has_critical():
addresses = list(set([a for a in itertools.chain(addresses, critical_addresses)]))
if not logger.has_errors():'No errors found. Email not sent')
for level in 'ERROR', 'CRITICAL':
if level not in notify:
notify[level] = []
if logger.has_errors():
addresses = notify['ERROR']
if logger.has_critical():
addresses = addresses.extend(notify['CRITICAL'])
addresses = list(set(addresses))
if not addresses:
logging.warning('No recipients found. Email not sent')
else:'No errors found. Email not sent')
class OperationalLogger:
......@@ -67,6 +73,7 @@ class OperationalLogger:
date_format_console = self.DEFAULT_DATE_FORMAT_CONSOLE
if from_address is None:
from_address = self.DEFAULT_FROM_ADDRESS
self.__from_address = from_address
logging.getLogger(logger_name).handlers = []
logging.getLogger('_sub').handlers = []
......@@ -126,8 +133,10 @@ class OperationalLogger:
"""Return whether any CRITICAL-level log statements in buffer."""
return len(self.__critical_buffer.getvalue()) > 0
def email_log(self, addresses, from_address):
def email_log(self, addresses, from_address=None):
"""Send email to addresses with current buffered errors."""
if from_address is None:
from_address = self.__from_address"Sending log email to %s", addresses)
message = "<br />\n".join([__class__.__colorize(line) for line in self.__error_buffer.getvalue().split('\n')])
OperationalMail.send_status(subject='Errors detected {}:{}'.format(self.__server_name,
......@@ -222,8 +231,8 @@ def setup(config_file=None, logger_name=None, log_filename=None, log_root_dir=No'Log writing to: %s', os.path.join(log_directory, log_filename))'Process id {}, parent process id {}'.format(os.getpid(), os.getppid()))'-' * 60)
if notify:
atexit.register(__exit_handler, logger, notify)
#if notify:
atexit.register(__exit_handler, logger, config['notify'])
def long_log(logging_level=logging.INFO, message=''):
......@@ -9,6 +9,8 @@ import os
import os.path
import logging
import shutil
import atexit
import StringIO
import OperationalLogger as OpLog
import OperationalMail as OpMail
......@@ -59,6 +61,13 @@ class TestAll(unittest.TestCase):
ret = OpLog.log_shell(['diff', self.file_a, self.file_a])
assert ret == 0
def test_exit_handler_no_rcpts(self):
logging.error("exit handler test no rcpts")
s = StringIO.StringIO()
temp_buffer = logging.StreamHandler(s)
assert "No recipients found. Email not sent" in s.getvalue()
if __name__ == '__main__':
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment