Capture warnings into log-helper

oletools/common/log_helper/_json_formatter.py

@@ -5,25 +5,52 @@
 class JsonFormatter(logging.Formatter):
     """
     Format every message to be logged as a JSON object
+
+    Uses the standard :py:class:`logging.Formatter` with standard arguments
+    to do the actual formatting, could save and use a user-supplied formatter
+    instead.
     """
     _is_first_line = True
 
     def __init__(self, other_logger_has_first_line=False):
         if other_logger_has_first_line:
             self._is_first_line = False
+        self.msg_formatter = logging.Formatter()    # could adjust this
 
     def format(self, record):
         """
         Since we don't buffer messages, we always prepend messages with a comma to make
         the output JSON-compatible. The only exception is when printing the first line,
         so we need to keep track of it.
 
-        We assume that all input comes from the OletoolsLoggerAdapter which
-        ensures that there is a `type` field in the record. Otherwise will have
-        to add a try-except around the access to `record.type`.
+        The actual conversion from :py:class:`logging.LogRecord` to a text message
+        (i.e. %-formatting, adding exception information, etc.) is delegated to the
+        standard :py:class:`logging.Formatter.
+
+        The dumped json structure contains fields `msg` with the formatted message,
+        `level` with the log-level of the message and `type`, which is created by
+        :py:class:`oletools.common.log_helper.OletoolsLoggerAdapter` or added here
+        (for input from e.g. captured warnings, third-party libraries)
         """
-        json_dict = dict(msg=record.msg.replace('\n', ' '), level=record.levelname)
-        json_dict['type'] = record.type
+        json_dict = dict(msg='', level='', type='')
+        try:
+            msg = self.msg_formatter.format(record)
+            json_dict['msg'] = msg.replace('\n', ' ')
+            json_dict['level'] = record.levelname
+            json_dict['type'] = record.type
+        except AttributeError:     # most probably: record has no "type" field
+            if record.name == 'py.warnings':   # this is from python's warning-capture logger
+                json_dict['type'] = 'warning'
+            else:
+                json_dict['type'] = 'msg'      # message of unknown origin
+        except Exception as exc:
+            try:
+                json_dict['msg'] = "Ignore {0} when formatting '{1}': {2}".format(type(exc), record.msg, exc)
+            except Exception as exc2:
+                json_dict['msg'] = 'Caught {0} in logging'.format(str(exc2))
+            json_dict['type'] = 'log-warning'
+            json_dict['level'] = 'warning'
+
         formatted_message = '    ' + json.dumps(json_dict)
 
         if self._is_first_line:

oletools/common/log_helper/_logger_adapter.py

@@ -7,6 +7,7 @@ class OletoolsLoggerAdapter(logging.LoggerAdapter):
     Adapter class for all loggers returned by the logging module.
     """
     _json_enabled = None
+    _is_warn_logger = False   # this is always False
 
     def print_str(self, message, **kwargs):
         """
@@ -44,7 +45,10 @@ def process(self, msg, kwargs):
             kwargs['extra']['type'] = kwargs['type']
             del kwargs['type']    # downstream loggers cannot deal with this
         if 'type' not in kwargs['extra']:
-            kwargs['extra']['type'] = 'msg'   # type will be added to LogRecord
+            if self._is_warn_logger:
+                kwargs['extra']['type'] = 'warning'    # this will add field
+            else:
+                kwargs['extra']['type'] = 'msg'        # 'type' to LogRecord
         return msg, kwargs
 
     def set_json_enabled_function(self, json_enabled):
@@ -53,6 +57,12 @@ def set_json_enabled_function(self, json_enabled):
         """
         self._json_enabled = json_enabled
 
+    def set_warnings_logger(self):
+        """Make this the logger for warnings"""
+        # create a object attribute that shadows the class attribute which is
+        # always False
+        self._is_warn_logger = True
+
     def level(self):
         """Return current level of logger."""
         return self.logger.level

oletools/common/log_helper/log_helper.py

@@ -152,6 +152,11 @@ def enable_logging(self, use_json=False, level='warning', log_format=DEFAULT_MES
         self._use_json = use_json
         sys.excepthook = self._get_except_hook(sys.excepthook)
 
+        # make sure warnings do not mess up our output
+        logging.captureWarnings(True)
+        warn_logger = self.get_or_create_silent_logger('py.warnings')
+        warn_logger.set_warnings_logger()
+
         # since there could be loggers already created we go through all of them
         # and set their levels to 0 so they will use the root logger's level
         for name in self._all_names:
@@ -174,6 +179,7 @@ def end_logging(self):
 
         # end logging
         self._all_names = set()
+        logging.captureWarnings(False)
         logging.shutdown()
 
         # end json list

tests/common/log_helper/log_helper_test_imported.py

@@ -4,14 +4,18 @@
 """
 
 from oletools.common.log_helper import log_helper
+import warnings
 
 DEBUG_MESSAGE = 'imported: debug log'
 INFO_MESSAGE = 'imported: info log'
 WARNING_MESSAGE = 'imported: warning log'
 ERROR_MESSAGE = 'imported: error log'
 CRITICAL_MESSAGE = 'imported: critical log'
 RESULT_MESSAGE = 'imported: result log'
+
 RESULT_TYPE = 'imported: result'
+ACTUAL_WARNING = 'Feature XYZ provided by this module might be deprecated at '\
+    'some point in the future ... or not'
 
 logger = log_helper.get_or_create_silent_logger('test_imported')
 
@@ -27,3 +31,7 @@ def log():
     logger.error(ERROR_MESSAGE)
     logger.critical(CRITICAL_MESSAGE)
     logger.info(RESULT_MESSAGE, type=RESULT_TYPE)
+
+
+def warn():
+    warnings.warn(ACTUAL_WARNING)

tests/common/log_helper/log_helper_test_main.py

@@ -2,6 +2,7 @@
 
 import sys
 import logging
+import warnings
 from tests.common.log_helper import log_helper_test_imported
 from oletools.common.log_helper import log_helper
 
@@ -11,7 +12,9 @@
 ERROR_MESSAGE = 'main: error log'
 CRITICAL_MESSAGE = 'main: critical log'
 RESULT_MESSAGE = 'main: result log'
+
 RESULT_TYPE = 'main: result'
+ACTUAL_WARNING = 'Warnings can pop up anywhere, have to be prepared!'
 
 logger = log_helper.get_or_create_silent_logger('test_main')
 
@@ -24,7 +27,8 @@ def enable_logging():
 
 def main(args):
     """
-    Try to cover possible logging scenarios. For each scenario covered, here's the expected args and outcome:
+    Try to cover possible logging scenarios. For each scenario covered, here's
+    the expected args and outcome:
     - Log without enabling: ['<level>']
         * logging when being imported - should never print
     - Log as JSON without enabling: ['as-json', '<level>']
@@ -35,13 +39,18 @@ def main(args):
         * logging as JSON when being run as script - should log messages as JSON
     - Enable, log as JSON and throw: ['enable', 'as-json', 'throw', '<level>']
         * should produce JSON-compatible output, even after an unhandled exception
+    - Enable, log as JSON and warn: ['enable', 'as-json', 'warn', '<level>']
+        * should produce JSON-compatible output, even after a warning
     """
 
     # the level should always be the last argument passed
     level = args[-1]
     use_json = 'as-json' in args
     throw = 'throw' in args
     percent_autoformat = '%-autoformat' in args
+    warn = 'warn' in args
+    exc_info = 'exc-info' in args
+    wrong_log_args = 'wrong-log-args' in args
 
     log_helper_test_imported.logger.setLevel(logging.ERROR)
 
@@ -53,6 +62,22 @@ def main(args):
     if throw:
         raise Exception('An exception occurred before ending the logging')
 
+    if warn:
+        warnings.warn(ACTUAL_WARNING)
+        log_helper_test_imported.warn()
+
+    if exc_info:
+        try:
+            raise Exception('This is an exception')
+        except Exception:
+            logger.exception('Caught exception')  # has exc_info=True
+
+    if wrong_log_args:
+        logger.info('Opening file /dangerous/file/with-%s-in-name')
+        logger.info('The result is %f')
+        logger.info('No result', 1.23)
+        logger.info('The result is %f', 'bla')
+
     log_helper.end_logging()
 
 

tests/common/log_helper/test_log_helper.py

@@ -16,8 +16,7 @@
 from tests.test_utils import PROJECT_ROOT
 
 # test file we use as "main" module
-TEST_FILE = relpath(join(dirname(abspath(__file__)), 'log_helper_test_main.py'),
-                    PROJECT_ROOT)
+TEST_FILE = join(dirname(abspath(__file__)), 'log_helper_test_main.py')
 
 # test file simulating a third party main module that only imports oletools
 TEST_FILE_3RD_PARTY = relpath(join(dirname(abspath(__file__)),
@@ -26,6 +25,8 @@
 
 PYTHON_EXECUTABLE = sys.executable
 
+PERCENT_FORMAT_OUTPUT = 'The answer is 47.'
+
 
 class TestLogHelper(unittest.TestCase):
     def test_it_doesnt_log_when_not_enabled(self):
@@ -113,7 +114,7 @@ def test_logs_type_in_json(self):
     def test_percent_autoformat(self):
         """Test that auto-formatting of log strings with `%` works."""
         output = self._run_test(['enable', '%-autoformat', 'info'])
-        self.assertIn('The answer is 47.', output)
+        self.assertIn(PERCENT_FORMAT_OUTPUT, output)
 
     def test_json_correct_on_exceptions(self):
         """
@@ -141,6 +142,93 @@ def test_import_by_third_party_enabled(self):
         self.assertIn('INFO:test_main:main: info log', output)
         self.assertIn('INFO:test_imported:imported: info log', output)
 
+    def test_json_correct_on_warnings(self):
+        """
+        Test that even on warnings our JSON is always correct
+        """
+        output = self._run_test(['enable', 'as-json', 'warn', 'warning'])
+        expected_messages = [
+            log_helper_test_main.WARNING_MESSAGE,
+            log_helper_test_main.ERROR_MESSAGE,
+            log_helper_test_main.CRITICAL_MESSAGE,
+            log_helper_test_imported.WARNING_MESSAGE,
+            log_helper_test_imported.ERROR_MESSAGE,
+            log_helper_test_imported.CRITICAL_MESSAGE,
+        ]
+
+        for msg in expected_messages:
+            self.assertIn(msg, output)
+
+        # last two entries of output should be warnings
+        jout = json.loads(output)
+        self.assertEqual(jout[-2]['level'], 'WARNING')
+        self.assertEqual(jout[-1]['level'], 'WARNING')
+        self.assertEqual(jout[-2]['type'], 'warning')
+        self.assertEqual(jout[-1]['type'], 'warning')
+        self.assertIn(log_helper_test_main.ACTUAL_WARNING, jout[-2]['msg'])
+        self.assertIn(log_helper_test_imported.ACTUAL_WARNING, jout[-1]['msg'])
+
+    def test_warnings(self):
+        """Check that warnings are captured and printed correctly"""
+        output = self._run_test(['enable', 'warn', 'warning'])
+
+        # find out which line contains the call to warnings.warn:
+        warnings_line = None
+        with open(TEST_FILE, 'rt') as reader:
+            for line_idx, line in enumerate(reader):
+                if 'warnings.warn' in line:
+                    warnings_line = line_idx + 1
+                    break
+        self.assertNotEqual(warnings_line, None)
+
+        imported_file = join(dirname(abspath(__file__)),
+                             'log_helper_test_imported.py')
+        imported_line = None
+        with open(imported_file, 'rt') as reader:
+            for line_idx, line in enumerate(reader):
+                if 'warnings.warn' in line:
+                    imported_line = line_idx + 1
+                    break
+        self.assertNotEqual(imported_line, None)
+
+        expect = '\n'.join([
+            'WARNING  ' + log_helper_test_main.WARNING_MESSAGE,
+            'ERROR    ' + log_helper_test_main.ERROR_MESSAGE,
+            'CRITICAL ' + log_helper_test_main.CRITICAL_MESSAGE,
+            'WARNING  ' + log_helper_test_imported.WARNING_MESSAGE,
+            'ERROR    ' + log_helper_test_imported.ERROR_MESSAGE,
+            'CRITICAL ' + log_helper_test_imported.CRITICAL_MESSAGE,
+            'WARNING  {0}:{1}: UserWarning: {2}'
+                .format(TEST_FILE, warnings_line, log_helper_test_main.ACTUAL_WARNING),
+            '  warnings.warn(ACTUAL_WARNING)',   # warnings include source line
+            '',
+            'WARNING  {0}:{1}: UserWarning: {2}'
+                .format(imported_file, imported_line, log_helper_test_imported.ACTUAL_WARNING),
+            '  warnings.warn(ACTUAL_WARNING)',   # warnings include source line
+        ])
+        self.assertEqual(output.strip(), expect)
+
+    def test_json_percent_formatting(self):
+        """Test that json-output has formatting args included in output."""
+        output = self._run_test(['enable', 'as-json', '%-autoformat', 'info'])
+        json.loads(output)    # check that this does not raise, so json is valid
+        self.assertIn(PERCENT_FORMAT_OUTPUT, output)
+
+    def test_json_exception_formatting(self):
+        """Test that json-output has formatted exception info in output"""
+        output = self._run_test(['enable', 'as-json', 'exc-info', 'info'])
+        json.loads(output)    # check that this does not raise, so json is valid
+        self.assertIn('Caught exception', output)      # actual log message
+        self.assertIn('This is an exception', output)    # message of caught exception
+        self.assertIn('Traceback (most recent call last)', output)    # start of trace
+        self.assertIn(TEST_FILE.replace('\\', '\\\\'), output)        # part of trace
+
+    def test_json_wrong_args(self):
+        """Test that too many or missing args do not raise exceptions inside logger"""
+        output = self._run_test(['enable', 'as-json', 'wrong-log-args', 'info'])
+        json.loads(output)    # check that this does not raise, so json is valid
+        # do not care about actual contents of output
+
     def _assert_json_messages(self, output, messages):
         try:
             json_data = json.loads(output)
@@ -160,8 +248,10 @@ def _run_test(self, args, should_succeed=True, run_third_party=False):
         we might get errors or false positives between sequential tests runs)
 
         When arg `run_third_party` is `True`, we do not run the `TEST_FILE` as
-        main moduel but the `TEST_FILE_3RD_PARTY` and return contents of
+        main module but the `TEST_FILE_3RD_PARTY` and return contents of
         `stderr` instead of `stdout`.
+
+        TODO: use tests.utils.call_and_capture
         """
         all_args = [PYTHON_EXECUTABLE, ]
         if run_third_party: