mirror of
https://github.com/certbot/certbot.git
synced 2026-01-23 07:20:55 +03:00
@@ -24,6 +24,7 @@ from acme import jose
|
||||
import letsencrypt
|
||||
|
||||
from letsencrypt import account
|
||||
from letsencrypt import colored_logging
|
||||
from letsencrypt import configuration
|
||||
from letsencrypt import constants
|
||||
from letsencrypt import client
|
||||
@@ -786,7 +787,7 @@ def _setup_logging(args):
|
||||
level = -args.verbose_count * 10
|
||||
fmt = "%(asctime)s:%(levelname)s:%(name)s:%(message)s"
|
||||
if args.text_mode:
|
||||
handler = logging.StreamHandler()
|
||||
handler = colored_logging.StreamHandler()
|
||||
handler.setFormatter(logging.Formatter(fmt))
|
||||
else:
|
||||
handler = log.DialogHandler()
|
||||
|
||||
40
letsencrypt/colored_logging.py
Normal file
40
letsencrypt/colored_logging.py
Normal file
@@ -0,0 +1,40 @@
|
||||
"""A formatter and StreamHandler for colorizing logging output."""
|
||||
import logging
|
||||
import sys
|
||||
|
||||
from letsencrypt import le_util
|
||||
|
||||
|
||||
class StreamHandler(logging.StreamHandler):
|
||||
"""Sends colored logging output to a stream.
|
||||
|
||||
If the specified stream is not a tty, the class works like the
|
||||
standard logging.StreamHandler. Default red_level is logging.WARNING.
|
||||
|
||||
:ivar bool colored: True if output should be colored
|
||||
:ivar bool red_level: The level at which to output
|
||||
|
||||
"""
|
||||
|
||||
def __init__(self, stream=None):
|
||||
super(StreamHandler, self).__init__(stream)
|
||||
self.colored = (sys.stderr.isatty() if stream is None else
|
||||
stream.isatty())
|
||||
self.red_level = logging.WARNING
|
||||
|
||||
def format(self, record):
|
||||
"""Formats the string representation of record.
|
||||
|
||||
:param logging.LogRecord record: Record to be formatted
|
||||
|
||||
:returns: Formatted, string representation of record
|
||||
:rtype: str
|
||||
|
||||
"""
|
||||
output = super(StreamHandler, self).format(record)
|
||||
if self.colored and record.levelno >= self.red_level:
|
||||
return ''.join((le_util.ANSI_SGR_RED,
|
||||
output,
|
||||
le_util.ANSI_SGR_RESET))
|
||||
else:
|
||||
return output
|
||||
@@ -18,6 +18,15 @@ Key = collections.namedtuple("Key", "file pem")
|
||||
CSR = collections.namedtuple("CSR", "file data form")
|
||||
|
||||
|
||||
# ANSI SGR escape codes
|
||||
# Formats text as bold or with increased intensity
|
||||
ANSI_SGR_BOLD = '\033[1m'
|
||||
# Colors text red
|
||||
ANSI_SGR_RED = "\033[31m"
|
||||
# Resets output format
|
||||
ANSI_SGR_RESET = "\033[0m"
|
||||
|
||||
|
||||
def run_script(params):
|
||||
"""Run the script with the given params.
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ import textwrap
|
||||
import zope.interface
|
||||
|
||||
from letsencrypt import interfaces
|
||||
from letsencrypt import le_util
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -30,8 +31,6 @@ class Reporter(object):
|
||||
LOW_PRIORITY = 2
|
||||
"""Low priority constant. See `add_message`."""
|
||||
|
||||
_RESET = '\033[0m'
|
||||
_BOLD = '\033[1m'
|
||||
_msg_type = collections.namedtuple('ReporterMsg', 'priority text on_crash')
|
||||
|
||||
def __init__(self):
|
||||
@@ -76,7 +75,7 @@ class Reporter(object):
|
||||
no_exception = sys.exc_info()[0] is None
|
||||
bold_on = sys.stdout.isatty()
|
||||
if bold_on:
|
||||
print self._BOLD
|
||||
print le_util.ANSI_SGR_BOLD
|
||||
print 'IMPORTANT NOTES:'
|
||||
first_wrapper = textwrap.TextWrapper(
|
||||
initial_indent=' - ', subsequent_indent=(' ' * 3))
|
||||
@@ -87,7 +86,7 @@ class Reporter(object):
|
||||
msg = self.messages.get()
|
||||
if no_exception or msg.on_crash:
|
||||
if bold_on and msg.priority > self.HIGH_PRIORITY:
|
||||
sys.stdout.write(self._RESET)
|
||||
sys.stdout.write(le_util.ANSI_SGR_RESET)
|
||||
bold_on = False
|
||||
lines = msg.text.splitlines()
|
||||
print first_wrapper.fill(lines[0])
|
||||
@@ -95,4 +94,4 @@ class Reporter(object):
|
||||
print "\n".join(
|
||||
next_wrapper.fill(line) for line in lines[1:])
|
||||
if bold_on:
|
||||
sys.stdout.write(self._RESET)
|
||||
sys.stdout.write(le_util.ANSI_SGR_RESET)
|
||||
|
||||
40
letsencrypt/tests/colored_logging_test.py
Normal file
40
letsencrypt/tests/colored_logging_test.py
Normal file
@@ -0,0 +1,40 @@
|
||||
"""Tests for letsencrypt.colored_logging."""
|
||||
import logging
|
||||
import StringIO
|
||||
import unittest
|
||||
|
||||
from letsencrypt import le_util
|
||||
|
||||
|
||||
class StreamHandlerTest(unittest.TestCase):
|
||||
"""Tests for letsencrypt.colored_logging."""
|
||||
|
||||
def setUp(self):
|
||||
from letsencrypt import colored_logging
|
||||
|
||||
self.stream = StringIO.StringIO()
|
||||
self.stream.isatty = lambda: True
|
||||
self.handler = colored_logging.StreamHandler(self.stream)
|
||||
|
||||
self.logger = logging.getLogger()
|
||||
self.logger.setLevel(logging.DEBUG)
|
||||
self.logger.addHandler(self.handler)
|
||||
|
||||
def test_format(self):
|
||||
msg = 'I did a thing'
|
||||
self.logger.debug(msg)
|
||||
self.assertEqual(self.stream.getvalue(), '{0}\n'.format(msg))
|
||||
|
||||
def test_format_and_red_level(self):
|
||||
msg = 'I did another thing'
|
||||
self.handler.red_level = logging.DEBUG
|
||||
self.logger.debug(msg)
|
||||
|
||||
self.assertEqual(self.stream.getvalue(),
|
||||
'{0}{1}{2}\n'.format(le_util.ANSI_SGR_RED,
|
||||
msg,
|
||||
le_util.ANSI_SGR_RESET))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main() # pragma: no cover
|
||||
Reference in New Issue
Block a user