log_helper: fix tests after refactoring code

Samir Aguiar · Christian Herdtweck
1 parent d3624f1d
Showing 3 changed files with 81 additions and 175 deletions
tests/util/log_helper/log_helper_test_imported.py
tests/util/log_helper/log_helper_test_main.py
tests/util/log_helper/test_log_helper.py
@@ -12,7 +12,7 @@ WARNING_MESSAGE = &#39;imported: warning log&#39;
 ERROR_MESSAGE = 'imported: error log'
 CRITICAL_MESSAGE = 'imported: critical log'
-logger = log_helper.get_or_create_logger('test_imported', logging.ERROR)
+logger = log_helper.get_or_create_silent_logger('test_imported', logging.ERROR)
 def log():
 """ Test log_helpers """
 import sys
-import logging
 from tests.util.log_helper import log_helper_test_imported
 from oletools.util.log_helper import log_helper
@@ -11,90 +10,48 @@ WARNING_MESSAGE = &#39;main: warning log&#39;
 ERROR_MESSAGE = 'main: error log'
 CRITICAL_MESSAGE = 'main: critical log'
-logger = log_helper.get_or_create_logger('test_main')
+logger = log_helper.get_or_create_silent_logger('test_main')
-def init_logging_and_log(cmd_line_args=None):
-    args = cmd_line_args if cmd_line_args else sys.argv
+def init_logging_and_log(args):
+    """
+    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>']
+        * logging as JSON when being imported - should never print
+    - Enable and log: ['enable', '<level>']
+        * logging when being run as script - should log messages
+    - Enable and log as JSON: ['as-json', 'enable', '<level>']
+        * 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
+    """
-    if 'silent' in args:
-        return _log_silently()
-    elif 'dictionary' in args:
-        return _log_dictionary(args)
-    elif 'current_level' in args:
-        _enable_logging()
-        return _log_at_current_level()
-    elif 'default' in args:
-        _enable_logging()
-        return _log(logger)
+    # the level should always be the last argument passed
+    level = args[-1]
+    use_json = 'as-json' in args
+    throw = 'throw' in args
-    use_json = '-j' in args
-    throw_exception = 'throw' in args
+    if 'enable' in args:
+        log_helper.enable_logging(use_json, level, stream=sys.stdout)
-    level = _parse_log_level(args)
+    _log()
-    _enable_logging(use_json, level)
-    _log(logger)
-    log_helper_test_imported.log()
-
-    if throw_exception:
+    if throw:
         raise Exception('An exception occurred before ending the logging')
-    _end_logging()
-
-
-def _parse_log_level(args):
-    if 'debug' in args:
-        return 'debug'
-    elif 'info' in args:
-        return 'info'
-    elif 'warning' in args:
-        return 'warning'
-    elif 'error' in args:
-        return 'error'
-    else:
-        return 'critical'
-
-
-def _log_dictionary(args):
-    level = _parse_log_level(args)
-    _enable_logging(True, level)
-
-    logger.log_at_current_level({
-        'msg': DEBUG_MESSAGE
-    })
     log_helper.end_logging()
-def _enable_logging(use_json=False, level='warning'):
-    log_helper.enable_logging(use_json, level, stream=sys.stdout)
-
-
-def _log_at_current_level():
-    logger.log_at_current_level(DEBUG_MESSAGE)
-
-
-def _log_silently():
-    silent_logger = log_helper.get_or_create_silent_logger('test_main_silent', logging.DEBUG - 1)
-    _log(silent_logger)
-
-
-def _log(current_logger):
-    current_logger.debug(DEBUG_MESSAGE)
-    current_logger.info(INFO_MESSAGE)
-    current_logger.warning(WARNING_MESSAGE)
-    current_logger.error(ERROR_MESSAGE)
-    current_logger.critical(CRITICAL_MESSAGE)
-
-
-def _end_logging():
-    log_helper.end_logging()
+def _log():
+    logger.debug(DEBUG_MESSAGE)
+    logger.info(INFO_MESSAGE)
+    logger.warning(WARNING_MESSAGE)
+    logger.error(ERROR_MESSAGE)
+    logger.critical(CRITICAL_MESSAGE)
+    log_helper_test_imported.log()
 if __name__ == '__main__':
-    # since we are using subprocess, add delimiters so we can easily extract
-    # the output that matters (when debugging tests we get extra output,
-    # so we need to ignore it)
-    print('<#')
-    init_logging_and_log(sys.argv)
-    print('#>')
+    init_logging_and_log(sys.argv[1:])
@@ -5,22 +5,18 @@ Check if it handles imported modules correctly
 and that the default silent logger won't log when nothing is enabled
 """
-from __future__ import print_function
-
 import unittest
 import sys
 import json
-import re
+import subprocess
 from tests.util.log_helper import log_helper_test_main
 from tests.util.log_helper import log_helper_test_imported
 from os.path import dirname, join, relpath, abspath
-from subprocess import check_output, STDOUT, CalledProcessError
 # this is the common base of "tests" and "oletools" dirs
-ROOT_DIRECTORY = dirname(dirname(dirname(dirname(abspath(__file__)))))
+ROOT_DIRECTORY = abspath(join(__file__, '..', '..', '..', '..'))
 TEST_FILE = relpath(join(dirname(__file__), 'log_helper_test_main.py'), ROOT_DIRECTORY)
 PYTHON_EXECUTABLE = sys.executable
-REGEX = re.compile('<#(.*)(:?#>|Traceback)', re.MULTILINE | re.DOTALL)
 MAIN_LOG_MESSAGES = [
     log_helper_test_main.DEBUG_MESSAGE,
@@ -32,31 +28,21 @@ MAIN_LOG_MESSAGES = [
 class TestLogHelper(unittest.TestCase):
-    def test_default_logging_to_stderr(self):
-        """
-        Basic test for simple logging
-        """
-        output = self._run_test(['default'])
-        self.assertIn(log_helper_test_main.WARNING_MESSAGE, output)
+    def test_it_doesnt_log_when_not_enabled(self):
+        output = self._run_test(['debug'])
+        self.assertTrue(len(output) == 0)
-    def test_logging_silently(self):
-        """
-        Test that nothing will be logged when logging is not enabled
-        and we are using a silent logger (uses the NullHandler)
-        """
-        output = self._run_test(['silent'])
+    def test_it_doesnt_log_json_when_not_enabled(self):
+        output = self._run_test(['as-json', 'debug'])
         self.assertTrue(len(output) == 0)
-    def test_setting_level_in_main_module(self):
-        """
-        Make sure that the level set in the main module is kept when
-        logging from imported modules.
-        """
-        output = self._run_test(['debug'])
+    def test_logs_when_enabled(self):
+        output = self._run_test(['enable', 'warning'])
-        expected_messages = MAIN_LOG_MESSAGES + [
-            log_helper_test_imported.DEBUG_MESSAGE,
-            log_helper_test_imported.INFO_MESSAGE,
+        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
@@ -65,97 +51,60 @@ class TestLogHelper(unittest.TestCase):
         for msg in expected_messages:
             self.assertIn(msg, output)
-    def test_logging_at_current_level(self):
-        """
-        Test that logging at current level will always print a message
-        """
-        output = self._run_test(['current_level'])
-        self.assertIn(log_helper_test_main.DEBUG_MESSAGE, output)
-
-    def test_logging_as_json(self):
-        """
-        Basic test for json logging
-        """
-        output = self._run_test(['critical', '-j'])
+    def test_logs_json_when_enabled(self):
+        output = self._run_test(['enable', 'as-json', 'critical'])
-        try:
-            json_data = json.loads(output)
-            self._assert_json_messages(json_data, [
-                log_helper_test_main.CRITICAL_MESSAGE,
-                log_helper_test_imported.CRITICAL_MESSAGE
-            ])
-        except ValueError:
-            self.fail('Invalid json:\n' + output)
-        self.assertNotEqual(len(json_data), 0, msg='Output was empty')
-
-    def test_logging_dictionary_as_json(self):
-        """
-        Test support for passing a dictionary to the logger
-        and have it logged as JSON
-        """
-        output = self._run_test(['dictionary'])
-
-        try:
-            json_data = json.loads(output)
-            self._assert_json_messages(json_data, [
-                log_helper_test_main.DEBUG_MESSAGE
-            ])
-        except ValueError:
-            self.fail('Invalid json:\n' + output)
-        self.assertNotEqual(len(json_data), 0, msg='Output was empty')
+        self._assert_json_messages(output, [
+            log_helper_test_main.CRITICAL_MESSAGE,
+            log_helper_test_imported.CRITICAL_MESSAGE
+        ])
     def test_json_correct_on_exceptions(self):
         """
         Test that even on unhandled exceptions our JSON is always correct
         """
-        output = self._run_test(['critical', 'throw', '-j'], True)
+        output = self._run_test(['enable', 'as-json', 'throw', 'critical'], False)
+        self._assert_json_messages(output, [
+            log_helper_test_main.CRITICAL_MESSAGE,
+            log_helper_test_imported.CRITICAL_MESSAGE
+        ])
+    def _assert_json_messages(self, output, messages):
         try:
             json_data = json.loads(output)
-            self._assert_json_messages(json_data, [
-                log_helper_test_main.CRITICAL_MESSAGE,
-                log_helper_test_imported.CRITICAL_MESSAGE
-            ])
+            self.assertEquals(len(json_data), len(messages))
+
+            for i in range(len(messages)):
+                self.assertEquals(messages[i], json_data[i]['msg'])
         except ValueError:
             self.fail('Invalid json:\n' + output)
-        self.assertNotEqual(len(json_data), 0, msg='Output was empty')
-    def _assert_json_messages(self, json_data, messages):
-        self.assertEquals(len(json_data), len(messages))
-
-        for i in range(len(messages)):
-            self.assertEquals(messages[i], json_data[i]['msg'])
+        self.assertNotEqual(len(json_data), 0, msg='Output was empty')
-    @staticmethod
-    def _run_test(args, ignore_exceptions=False):
+    def _run_test(self, args, should_succeed=True):
         """
         Use subprocess to better simulate the real scenario and avoid
         logging conflicts when running multiple tests (since logging depends on singletons,
         we might get errors or false positives between sequential tests runs)
         """
-        try:
-            output = check_output(
-                [PYTHON_EXECUTABLE, TEST_FILE] + args,
-                shell=False,
-                cwd=ROOT_DIRECTORY,
-                stderr=STDOUT,
-                universal_newlines=True
-            )
-
-            if not isinstance(output, str):
-                output = output.decode('utf-8')
-        except CalledProcessError as ex:
-            if ignore_exceptions:
-                output = ex.output
-            else:
-                # we want tests to fail if an exception occur
-                print('Caught unexpected error. Print output and re-raise')
-                for line in ex.output.splitlines():
-                    print('error output: {}'.format(line.rstrip()))
-                print('(If you have errors about imports, try unsetting PYTHONPATH)')
-                raise ex
-
-        return REGEX.search(output).group(1).strip()
+        child = subprocess.Popen(
+            [PYTHON_EXECUTABLE, TEST_FILE] + args,
+            shell=False,
+            env={'PYTHONPATH': ROOT_DIRECTORY},
+            universal_newlines=True,
+            cwd=ROOT_DIRECTORY,
+            stdin=None,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE
+        )
+        (output, output_err) = child.communicate()
+
+        if not isinstance(output, str):
+            output = output.decode('utf-8')
+
+        self.assertEquals(child.returncode == 0, should_succeed)
+
+        return output.strip()
 # just in case somebody calls this file as a script