Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions tabcmd/execution/logger_config.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import logging
import logging.handlers
import os
import sys

path = os.path.dirname(os.path.abspath(__file__))

Expand Down Expand Up @@ -36,6 +37,10 @@ def logToRoot(message, *args, **kwargs):
setattr(logging, method_name, logToRoot)


def _below_warning(record: logging.LogRecord) -> bool:
return record.levelno < logging.WARNING


def add_trace_level():
trace_level: int = logging.DEBUG - 5
add_log_level("TRACE", trace_level)
Expand Down Expand Up @@ -63,10 +68,18 @@ def configure_log(name: str, logging_level_input: str):
if not any(
isinstance(h, logging.StreamHandler) and not isinstance(h, logging.FileHandler) for h in named_logger.handlers
):
console = logging.StreamHandler()
console.setLevel(logging_level)
console.setFormatter(logging.Formatter(log_format))
named_logger.addHandler(console)
# INFO and below → stdout so PowerShell doesn't treat normal output as errors.
# WARNING and above → stderr so real errors are still distinguishable.
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setLevel(logging_level)
stdout_handler.addFilter(_below_warning)
stdout_handler.setFormatter(logging.Formatter(log_format))
named_logger.addHandler(stdout_handler)

stderr_handler = logging.StreamHandler(sys.stderr)
stderr_handler.setLevel(logging.WARNING)
stderr_handler.setFormatter(logging.Formatter(FORMATS[logging.ERROR]))
named_logger.addHandler(stderr_handler)

return named_logger

Expand Down
97 changes: 91 additions & 6 deletions tests/commands/test_logger_config.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,26 @@
import io
import logging
import logging.handlers
import sys
import unittest
from unittest import mock

from tabcmd.execution.logger_config import configure_log


def _clear_handlers():
root = logging.getLogger()
for handler in root.handlers[:]:
handler.close()
root.removeHandler(handler)
for name in list(logging.Logger.manager.loggerDict):
if name.startswith("test_"):
logging.getLogger(name).handlers.clear()


class LoggerConfigTests(unittest.TestCase):
def setUp(self):
root = logging.getLogger()
for handler in root.handlers[:]:
handler.close()
root.removeHandler(handler)
_clear_handlers()

def test_rotating_file_handler_present(self):
configure_log("test_rotate", "DEBUG")
Expand All @@ -27,17 +37,92 @@ def test_rotating_file_handler_settings(self):
self.assertEqual(handler.maxBytes, 1_000_000)
self.assertEqual(handler.backupCount, 5)

def test_no_duplicate_handlers_on_repeated_calls(self):
def test_no_duplicate_file_handlers_on_repeated_calls(self):
configure_log("test_dup", "DEBUG")
configure_log("test_dup", "DEBUG")
root_handlers = logging.getLogger().handlers
rotating = [h for h in root_handlers if isinstance(h, logging.handlers.RotatingFileHandler)]
self.assertEqual(len(rotating), 1, "Repeated configure_log calls must not add duplicate handlers")
self.assertEqual(len(rotating), 1, "Repeated configure_log calls must not add duplicate file handlers")

def test_no_duplicate_console_handlers_on_repeated_calls(self):
fake_out, fake_err = io.StringIO(), io.StringIO()
with mock.patch("sys.stdout", fake_out), mock.patch("sys.stderr", fake_err):
configure_log("test_dup_console", "DEBUG")
configure_log("test_dup_console", "DEBUG")
named = logging.getLogger("test_dup_console")
stream_handlers = [
h for h in named.handlers if isinstance(h, logging.StreamHandler) and not isinstance(h, logging.FileHandler)
]
self.assertEqual(len(stream_handlers), 2, "Repeated calls must not add more than 2 console handlers")

def test_root_logger_level_set(self):
configure_log("test_level", "WARNING")
self.assertEqual(logging.getLogger().level, logging.WARNING)


class StdoutStderrSplitTests(unittest.TestCase):
"""INFO goes to stdout; WARNING+ goes to stderr."""

def setUp(self):
_clear_handlers()

def _make_logger(self, name):
fake_out, fake_err = io.StringIO(), io.StringIO()
with mock.patch("sys.stdout", fake_out), mock.patch("sys.stderr", fake_err):
logger = configure_log(name, "DEBUG")
logger.propagate = False
return logger, fake_out, fake_err

def test_two_stream_handlers_attached(self):
logger, _, _ = self._make_logger("test_split_two_handlers")
stream_handlers = [
h
for h in logger.handlers
if isinstance(h, logging.StreamHandler) and not isinstance(h, logging.FileHandler)
]
self.assertEqual(len(stream_handlers), 2)

def test_stdout_and_stderr_handlers_bound_to_correct_streams(self):
fake_out, fake_err = io.StringIO(), io.StringIO()
with mock.patch("sys.stdout", fake_out), mock.patch("sys.stderr", fake_err):
logger = configure_log("test_split_streams", "DEBUG")
stream_handlers = [
h
for h in logger.handlers
if isinstance(h, logging.StreamHandler) and not isinstance(h, logging.FileHandler)
]
streams = {h.stream for h in stream_handlers}
self.assertIn(fake_out, streams)
self.assertIn(fake_err, streams)

def test_info_goes_to_stdout_not_stderr(self):
logger, fake_out, fake_err = self._make_logger("test_split_info")
logger.info("hello stdout")
self.assertIn("hello stdout", fake_out.getvalue())
self.assertNotIn("hello stdout", fake_err.getvalue())

def test_warning_goes_to_stderr_not_stdout(self):
logger, fake_out, fake_err = self._make_logger("test_split_warning")
logger.warning("a warning")
self.assertNotIn("a warning", fake_out.getvalue())
self.assertIn("a warning", fake_err.getvalue())

def test_error_goes_to_stderr_not_stdout(self):
logger, fake_out, fake_err = self._make_logger("test_split_error")
logger.error("an error")
self.assertNotIn("an error", fake_out.getvalue())
self.assertIn("an error", fake_err.getvalue())

def test_no_double_print_from_sibling_loggers(self):
"""Two distinct leaf loggers sharing the same streams must not double-print."""
fake_out, fake_err = io.StringIO(), io.StringIO()
with mock.patch("sys.stdout", fake_out), mock.patch("sys.stderr", fake_err):
logger_a = configure_log("test_split_sibling_a", "DEBUG")
configure_log("test_split_sibling_b", "DEBUG")
logger_a.propagate = False
logger_a.info("msg from a")
self.assertEqual(fake_out.getvalue().count("msg from a"), 1, "Message should appear exactly once")


if __name__ == "__main__":
unittest.main()
Loading