logutil: add multi / single threaded formats and test cases

ohmu_common_py/logutil.py

@@ -15,30 +15,43 @@
     daemon = None
 
 
-LOG_FORMAT = "%(asctime)s\t%(name)s\t%(threadName)s\t%(levelname)s\t%(message)s"
+LOG_FORMAT_BASIC = "%(asctime)s\t%(name)s\t%(levelname)s\t%(message)s"
+LOG_FORMAT_BASIC_MT = "%(asctime)s\t%(name)s\t%(threadName)s\t%(levelname)s\t%(message)s"
 LOG_FORMAT_SHORT = "%(levelname)s\t%(message)s"
-LOG_FORMAT_SYSLOG = "%(name)s %(threadName)s %(levelname)s: %(message)s"
+LOG_FORMAT_SYSLOG = "%(name)s %(levelname)s: %(message)s"
+LOG_FORMAT_SYSLOG_MT = "%(name)s %(threadName)s %(levelname)s: %(message)s"
 
+_multi_threaded = False
 
-def set_syslog_handler(address, facility, logger):
+
+def set_syslog_handler(address, facility, logger, multi_threaded=None):
+    if multi_threaded is None:
+        multi_threaded = _multi_threaded
     syslog_handler = logging.handlers.SysLogHandler(address=address, facility=facility)
     logger.addHandler(syslog_handler)
-    formatter = logging.Formatter(LOG_FORMAT_SYSLOG)
+    formatter = logging.Formatter(LOG_FORMAT_SYSLOG_MT if multi_threaded else LOG_FORMAT_SYSLOG)
     syslog_handler.setFormatter(formatter)
     return syslog_handler
 
 
-def configure_logging(level=logging.DEBUG, short_log=False):
+def configure_logging(level=logging.DEBUG, short_log=False, multi_threaded=False):
+    global _multi_threaded  # pylint: disable=global-statement
+    _multi_threaded = multi_threaded
+
     # Are we running under systemd?
     if os.getenv("NOTIFY_SOCKET"):
-        logging.basicConfig(level=level, format=LOG_FORMAT_SYSLOG)
+        format_string = LOG_FORMAT_SYSLOG_MT if multi_threaded else LOG_FORMAT_SYSLOG
         if not daemon:
             print(
                 "WARNING: Running under systemd but python-systemd not available, "
                 "systemd won't see our notifications"
             )
+    elif short_log:
+        format_string = LOG_FORMAT_SHORT
     else:
-        logging.basicConfig(level=level, format=LOG_FORMAT_SHORT if short_log else LOG_FORMAT)
+        format_string = LOG_FORMAT_BASIC_MT if multi_threaded else LOG_FORMAT_BASIC
+
+    logging.basicConfig(level=level, format=format_string)
 
 
 def notify_systemd(status):

sync.py

@@ -15,6 +15,7 @@
     "ohmu_common_py/logutil.py",
     "ohmu_common_py/pgutil.py",
     "ohmu_common_py/statsd.py",
+    "test/test_logutil.py",
     "test/test_pgutil.py",
     "version.py",
 ]

test/conftest.py

@@ -5,7 +5,7 @@
 See LICENSE for details
 """
 
-from ohmu_common_py import logutil
+from ohmu_common_py import logutil, version  # noqa pylint: disable=unused-import
 
 
 logutil.configure_logging()

test/test_logutil.py

@@ -0,0 +1,78 @@
+"""
+ohmu_common_py - logutil tests
+
+Copyright (c) 2016 Ohmu Ltd
+See LICENSE for details
+"""
+# pylint: disable=protected-access
+from ohmu_common_py import logutil
+import logging
+import logging.handlers
+import os
+import pytest
+
+
+@pytest.yield_fixture
+def reset_logging():
+    """reset root log handler to the default value after running the test"""
+    notify_socket = os.environ.pop("NOTIFY_SOCKET", None)
+    roothandlers = logging.getLogger().handlers
+    oldhandlers = roothandlers.copy()
+    roothandlers.clear()
+    try:
+        yield
+    finally:
+        roothandlers.clear()
+        roothandlers.extend(oldhandlers)
+        if notify_socket is not None:
+            os.environ["NOTIFY_SOCKET"] = notify_socket
+
+
+def test_configure_logging(reset_logging):  # pylint: disable=redefined-outer-name,unused-argument
+    roothandlers = logging.getLogger().handlers
+    roothandlers.clear()
+    logutil.configure_logging(short_log=True)
+    assert "(name)" not in roothandlers[0].formatter._fmt
+
+    roothandlers.clear()
+    logutil.configure_logging(short_log=False)
+    assert "(name)" in roothandlers[0].formatter._fmt
+    assert "(threadName)" not in roothandlers[0].formatter._fmt
+
+    roothandlers.clear()
+    logutil.configure_logging(multi_threaded=True)
+    assert "(threadName)" in roothandlers[0].formatter._fmt
+    assert "(asctime)" in roothandlers[0].formatter._fmt
+
+    os.environ["NOTIFY_SOCKET"] = "/dev/null"
+    roothandlers.clear()
+    logutil.configure_logging(multi_threaded=True)
+    assert "(threadName)" in roothandlers[0].formatter._fmt
+    assert "(asctime)" not in roothandlers[0].formatter._fmt
+
+
+def test_syslog_handler(reset_logging):  # pylint: disable=redefined-outer-name,unused-argument
+    rootlogger = logging.getLogger()
+    roothandlers = rootlogger.handlers
+    roothandlers.clear()
+    logutil.configure_logging()
+    logutil.set_syslog_handler("/dev/log", "local2", rootlogger)
+    assert len(roothandlers) == 2
+
+    assert isinstance(roothandlers[0], logging.StreamHandler)
+    assert "(asctime)" in roothandlers[0].formatter._fmt
+    assert "(threadName)" not in roothandlers[0].formatter._fmt
+
+    assert isinstance(roothandlers[1], logging.handlers.SysLogHandler)
+    assert "(asctime)" not in roothandlers[1].formatter._fmt
+    assert "(threadName)" not in roothandlers[1].formatter._fmt
+
+    roothandlers.clear()
+    logutil.configure_logging(multi_threaded=True)
+    logutil.set_syslog_handler("/dev/log", "local2", rootlogger)
+    assert len(roothandlers) == 2
+
+    assert isinstance(roothandlers[0], logging.StreamHandler)
+    assert "(threadName)" in roothandlers[0].formatter._fmt
+    assert isinstance(roothandlers[1], logging.handlers.SysLogHandler)
+    assert "(threadName)" in roothandlers[1].formatter._fmt