Better, simpler logging

detect_secrets/core/log.py

@@ -1,59 +1,49 @@
-#!/usr/bin/python
 import logging
 import sys
 
 
-class CustomLog(logging.getLoggerClass()):  # pragma: no cover
+def get_logger(name=None, format_string=None):
+    """
+    :type name: str
+    :param name: used for declaring log channels.
 
-    log_format_string = '[%(module)s]\t%(levelname)s\t%(message)s'
+    :type format_string: str
+    :param format_string: for custom formatting
+    """
+    logging.captureWarnings(True)
+    log = logging.getLogger(name)
 
-    # See CustomLog.enableDebug
-    debug_mode = 0
+    # Bind custom method to instance.
+    # Source: https://stackoverflow.com/a/2982
+    log.set_debug_level = _set_debug_level.__get__(log)
+    log.set_debug_level(0)
 
-    def __init__(self, debug_mode=None, formatter=None, *args, **kwargs):
-        """
-        :param name:       string; used for declaring log channels.
-        :param debug_mode: debug level for this specific logger instance.
-        :param formatter:  string; for custom formatting
-        """
-        super(CustomLog, self).__init__('', *args, **kwargs)
+    if not format_string:
+        format_string = '[%(module)s]\t%(levelname)s\t%(message)s'
 
-        if debug_mode is not None:
-            self.debug_mode = debug_mode
+    # Setting up log formats
+    log.handlers = []
+    handler = logging.StreamHandler(sys.stderr)
+    handler.setFormatter(
+        logging.Formatter(format_string),
+    )
+    log.addHandler(handler)
 
-        if formatter is None:
-            self.formatter = logging.Formatter(CustomLog.log_format_string)
-        elif isinstance(formatter, str):
-            self.formatter = logging.Formatter(formatter)
+    return log
 
-    @classmethod
-    def enableDebug(cls, verbose_level):
-        """Configure the global verbosity of logs
 
-        :param verbose_level: integer; between 0-2
-        """
-        cls.debug_mode = verbose_level
+def _set_debug_level(self, debug_level):
+    """
+    :type debug_level: int, between 0-2
+    :param debug_level: configure verbosity of log
+    """
+    mapping = {
+        0: logging.ERROR,
+        1: logging.INFO,
+        2: logging.DEBUG,
+    }
 
-    def getLogger(self, name=None):
-        log = logging.getLogger(name)
+    self.setLevel(mapping[debug_level])
 
-        debug_mode = self.debug_mode if self.debug_mode is not None else CustomLog.debug_mode
 
-        # Apply custom default options
-        log_level = logging.ERROR
-        if debug_mode == 1:
-            log_level = logging.INFO
-        elif debug_mode == 2:
-            log_level = logging.DEBUG
-
-        log.setLevel(log_level)
-
-        if self.formatter:
-            log.handlers = []
-            handler = logging.StreamHandler(sys.stderr)
-            handler.setFormatter(self.formatter)
-            log.addHandler(handler)
-
-        logging.captureWarnings(True)
-
-        return log
+log = get_logger()

detect_secrets/core/secrets_collection.py

@@ -11,14 +11,11 @@
 from unidiff.errors import UnidiffParseError
 
 from detect_secrets import VERSION
-from detect_secrets.core.log import CustomLog
+from detect_secrets.core.log import log
 from detect_secrets.core.potential_secret import PotentialSecret
 from detect_secrets.plugins.core import initialize
 
 
-CustomLogObj = CustomLog()
-
-
 class SecretsCollection(object):
 
     def __init__(self, plugins=(), exclude_regex=''):
@@ -51,7 +48,7 @@ def load_baseline_from_string(cls, string):
         try:
             return cls._load_baseline_from_dict(json.loads(string))
         except (IOError, ValueError):
-            CustomLogObj.getLogger().error('Incorrectly formatted baseline!')
+            log.error('Incorrectly formatted baseline!')
             raise
 
     @classmethod
@@ -135,7 +132,7 @@ def scan_diff(
                 'hash': last_commit_hash,
                 'repo_name': repo_name,
             }
-            CustomLogObj.getLogger().error(alert)
+            log.error(alert)
             raise
 
         if self.exclude_regex:
@@ -183,7 +180,7 @@ def scan_file(self, filename, filename_key=None):
 
             return True
         except IOError:
-            CustomLogObj.getLogger().warning("Unable to open file: %s", filename)
+            log.warning("Unable to open file: %s", filename)
             return False
 
     def get_secret(self, filename, secret, type_=None):
@@ -272,7 +269,6 @@ def _extract_secrets_from_file(self, f, filename):
         :type f:        File object
         :type filename: string
         """
-        log = CustomLogObj.getLogger()
         try:
             log.info("Checking file: %s", filename)
 

detect_secrets/main.py

@@ -8,7 +8,7 @@
 from detect_secrets import VERSION
 from detect_secrets.core import audit
 from detect_secrets.core import baseline
-from detect_secrets.core.log import CustomLog
+from detect_secrets.core.log import log
 from detect_secrets.core.usage import ParserBuilder
 from detect_secrets.plugins.core import initialize
 
@@ -24,7 +24,7 @@ def main(argv=None):
 
     args = parse_args(argv)
     if args.verbose:  # pragma: no cover
-        CustomLog.enableDebug(args.verbose)
+        log.set_debug_level(args.verbose)
 
     if args.version:    # pragma: no cover
         print(VERSION)

detect_secrets/plugins/core/initialize.py

@@ -8,10 +8,7 @@
 from ..high_entropy_strings import Base64HighEntropyString  # noqa: F401
 from ..high_entropy_strings import HexHighEntropyString     # noqa: F401
 from ..private_key import PrivateKeyDetector                # noqa: F401
-from detect_secrets.core.log import CustomLog
-
-
-_CustomLogObj = CustomLog()
+from detect_secrets.core.log import log
 
 
 def from_parser_builder(plugins_dict):
@@ -45,7 +42,7 @@ def from_plugin_classname(plugin_classname, **kwargs):
     try:
         instance = klass(**kwargs)
     except TypeError:
-        _CustomLogObj.getLogger().warning(
+        log.warning(
             'Unable to initialize plugin!',
         )
         raise

detect_secrets/pre_commit_hook.py

@@ -5,20 +5,18 @@
 import sys
 import textwrap
 
-try:
-    from functools import lru_cache
-except ImportError:
-    from functools32 import lru_cache
-
 from detect_secrets import VERSION
 from detect_secrets.core.baseline import get_secrets_not_in_baseline
 from detect_secrets.core.baseline import update_baseline_with_removed_secrets
-from detect_secrets.core.log import CustomLog
+from detect_secrets.core.log import get_logger
 from detect_secrets.core.secrets_collection import SecretsCollection
 from detect_secrets.core.usage import ParserBuilder
 from detect_secrets.plugins.core import initialize
 
 
+log = get_logger(format_string='%(message)s')
+
+
 def parse_args(argv):
     return ParserBuilder().add_pre_commit_arguments()\
         .parse_args(argv)
@@ -27,7 +25,7 @@ def parse_args(argv):
 def main(argv=None):
     args = parse_args(argv)
     if args.verbose:  # pragma: no cover
-        CustomLog.enableDebug(args.verbose)
+        log.set_debug_level(args.verbose)
 
     try:
         # If baseline is provided, we first want to make sure
@@ -101,7 +99,7 @@ def get_baseline(baseline_filename):
             baseline_version,
         )
     except ValueError:
-        _get_custom_log().error(
+        log.error(
             'The supplied baseline may be incompatible with the current\n'
             'version of detect-secrets. Please recreate your baseline to\n'
             'avoid potential mis-configurations.\n\n'
@@ -123,7 +121,7 @@ def _get_baseline_string_from_file(filename):   # pragma: no cover
             return f.read()
 
     except IOError:
-        _get_custom_log().error(
+        log.error(
             'Unable to open baseline file: %s.', filename,
         )
 
@@ -150,7 +148,7 @@ def raise_exception_if_baseline_file_is_not_up_to_date(filename):
         raise ValueError
 
     if filename.encode() in files_changed_but_not_staged:
-        _get_custom_log().error((
+        log.error((
             'Your baseline file ({}) is unstaged.\n'
             '`git add {}` to fix this.'
         ).format(
@@ -211,19 +209,12 @@ def pretty_print_diagnostics(secrets):
 
     :type secrets: SecretsCollection
     """
-    log = _get_custom_log()
-
-    _print_warning_header(log)
-    _print_secrets_found(log, secrets)
-    _print_mitigation_suggestions(log)
-
-
-@lru_cache(maxsize=1)
-def _get_custom_log():
-    return CustomLog(formatter='%(message)s').getLogger()
+    _print_warning_header()
+    _print_secrets_found(secrets)
+    _print_mitigation_suggestions()
 
 
-def _print_warning_header(log):
+def _print_warning_header():
     message = (
         'Potential secrets about to be committed to git repo! Please rectify '
         'or explicitly ignore with `pragma: whitelist secret` comment.'
@@ -233,13 +224,13 @@ def _print_warning_header(log):
     log.error('')
 
 
-def _print_secrets_found(log, secrets):
+def _print_secrets_found(secrets):
     for filename in secrets.data:
         for secret in secrets.data[filename].values():
             log.error(secret)
 
 
-def _print_mitigation_suggestions(log):
+def _print_mitigation_suggestions():
     suggestions = [
         'For information about putting your secrets in a safer place, please ask in #security',
         'Mark false positives with `# pragma: whitelist secret`',

testing/mocks.py

@@ -1,5 +1,6 @@
 """This is a collection of utility functions for easier, DRY testing."""
 import io
+from collections import defaultdict
 from collections import namedtuple
 from contextlib import contextmanager
 from subprocess import CalledProcessError
@@ -127,5 +128,33 @@ def mock_file_object(string):
 
 @contextmanager
 def mock_log(namespace):
-    with mock.patch(namespace, autospec=True) as m:
-        yield m
+    class MockLogWrapper(object):
+        """This is used to check what is being logged."""
+
+        def __init__(self):
+            self.messages = defaultdict(str)
+
+        def error(self, message, *args):
+            self.messages['error'] += (str(message) + '\n') % args
+
+        @property
+        def error_messages(self):
+            return self.messages['error']
+
+        def warning(self, message, *args):
+            self.messages['warning'] += (str(message) + '\n') % args
+
+        @property
+        def warning_messages(self):
+            return self.messages['warning']
+
+        def info(self, message, *args):
+            self.messages['info'] += (str(message) + '\n') % args
+
+        @property
+        def info_messages(self):
+            return self.messages['info']
+
+    wrapper = MockLogWrapper()
+    with mock.patch(namespace, wrapper):
+        yield wrapper

tests/core/secrets_collection_test.py

@@ -21,7 +21,7 @@
 
 @pytest.fixture
 def mock_log():
-    with mock_log_base('detect_secrets.core.secrets_collection.CustomLogObj') as m:
+    with mock_log_base('detect_secrets.core.secrets_collection.log') as m:
         yield m
 
 
@@ -58,7 +58,7 @@ def test_error_reading_file(self, mock_log):
         logic = secrets_collection_factory()
 
         assert not logic.scan_file('non_existent_file')
-        mock_log.getLogger().warning.assert_called_once()
+        mock_log.warning_messages == 'Unable to open file: non_existent_file'
 
     def test_success_single_plugin(self):
         logic = secrets_collection_factory(
@@ -103,7 +103,8 @@ def test_unicode_decode_error(self, mock_log):
 
             logic.scan_file('filename')
 
-        assert mock_log.getLogger().warning.called
+        assert mock_log.info_messages == 'Checking file: filename\n'
+        assert mock_log.warning_messages == 'filename failed to load.\n'
 
         # If the file read was successful, secret would have been caught and added.
         assert len(logic.data) == 0
@@ -277,7 +278,7 @@ def test_load_baseline_with_invalid_input(self, mock_log):
                 }),
             )
 
-        assert mock_log.getLogger().error.called
+        assert mock_log.error_messages == 'Incorrectly formatted baseline!\n'
 
     def get_baseline_dict(self, gmtime):
         # They are all the same secret, so they should all have the same secret hash.

tests/plugins/core/initialize_test.py

@@ -22,7 +22,7 @@ def test_success(self):
     def test_fails_if_not_base_plugin(self):
         with pytest.raises(TypeError):
             initialize.from_plugin_classname(
-                'CustomLog',
+                'log',
             )
 
     def test_fails_on_bad_initialization(self):