test(logging): add unit/functional tests for buffering handler and logger configuration

Author: acascell

tests/functional/logger/required_dependencies/test_logger_utils.py

@@ -6,11 +6,14 @@
 import random
 import string
 from enum import Enum
+from unittest.mock import patch
 
 import pytest
 
 from aws_lambda_powertools import Logger
 from aws_lambda_powertools.logging import formatter, utils
+from aws_lambda_powertools.logging.buffer import LoggerBufferConfig
+from aws_lambda_powertools.logging.buffer.handler import BufferingHandler
 
 
 @pytest.fixture
@@ -315,3 +318,84 @@ def test_copy_config_to_ext_loggers_but_preserve_log_levels(stdout, logger, log_
     # THEN external loggers log levels should be preserved
     assert logger_1.level != powertools_logger.log_level
     assert logger_2.level != powertools_logger.log_level
+
+
+def test_copy_config_with_buffering_uses_buffering_handler(stdout, logger, log_level):
+    # GIVEN an external logger and Powertools logger with buffer_config
+    ext_logger = logger()
+    buffer_config = LoggerBufferConfig(max_bytes=10240)
+    powertools_logger = Logger(
+        service=service_name(),
+        level=log_level.INFO.value,
+        stream=stdout,
+        buffer_config=buffer_config,
+    )
+
+    # WHEN configuration is copied with include_buffering=True
+    utils.copy_config_to_registered_loggers(
+        source_logger=powertools_logger,
+        include_buffering=True,
+        include={ext_logger.name},
+    )
+
+    # THEN external logger has exactly one handler and it is BufferingHandler (not StreamHandler)
+    assert len(ext_logger.handlers) == 1
+    assert isinstance(ext_logger.handlers[0], BufferingHandler)
+    assert not isinstance(ext_logger.handlers[0], logging.StreamHandler)
+
+    # AND when external logger emits a log with mocked tracer_id, it is buffered
+    with patch("aws_lambda_powertools.logging.logger.get_tracer_id", return_value="test-trace-id"):
+        ext_logger.info("Test message from external logger")
+        output = stdout.getvalue()
+        assert "Test message from external logger" not in output
+
+        # AND when buffer is flushed, the log appears
+        powertools_logger.flush_buffer()
+        output = stdout.getvalue()
+        assert "Test message from external logger" in output
+
+
+def test_copy_config_buffering_without_config_uses_normal_handlers(stdout, logger, log_level):
+    # GIVEN an external logger and Powertools logger WITHOUT buffer_config
+    ext_logger = logger()
+    powertools_logger = Logger(
+        service=service_name(),
+        level=log_level.INFO.value,
+        stream=stdout,
+    )
+
+    # WHEN configuration is copied with include_buffering=True
+    utils.copy_config_to_registered_loggers(
+        source_logger=powertools_logger,
+        include_buffering=True,
+        include={ext_logger.name},
+    )
+
+    # THEN external logger gets normal Powertools handler (no BufferingHandler)
+    assert len(ext_logger.handlers) == 1
+    assert isinstance(ext_logger.handlers[0], logging.StreamHandler)
+    assert isinstance(ext_logger.handlers[0].formatter, formatter.LambdaPowertoolsFormatter)
+    assert not isinstance(ext_logger.handlers[0], BufferingHandler)
+
+
+def test_copy_config_to_ext_loggers_default_include_buffering_false(stdout, logger, log_level):
+    # GIVEN Powertools logger with buffer_config and external logger
+    ext_logger = logger()
+    buffer_config = LoggerBufferConfig(max_bytes=10240)
+    powertools_logger = Logger(
+        service=service_name(),
+        level=log_level.INFO.value,
+        stream=stdout,
+        buffer_config=buffer_config,
+    )
+
+    # WHEN configuration is copied without include_buffering (default False)
+    utils.copy_config_to_registered_loggers(
+        source_logger=powertools_logger,
+        include={ext_logger.name},
+    )
+
+    # THEN external logger gets normal StreamHandler, not BufferingHandler
+    assert len(ext_logger.handlers) == 1
+    assert isinstance(ext_logger.handlers[0], logging.StreamHandler)
+    assert not isinstance(ext_logger.handlers[0], BufferingHandler)

tests/unit/logger/required_dependencies/test_logger_buffer_handler.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import io
+import json
+import logging
+
+from aws_lambda_powertools import Logger
+from aws_lambda_powertools.logging.buffer.config import LoggerBufferConfig
+from aws_lambda_powertools.logging.buffer.handler import BufferingHandler
+from aws_lambda_powertools.shared import constants
+
+
+def test_buffering_handler_init_stores_dependencies():
+    # GIVEN real buffer_config, source_logger (Logger with buffer), and its buffer_cache
+    buffer_config = LoggerBufferConfig(max_bytes=10240)
+    source_logger = Logger(service="test1", buffer_config=buffer_config, stream=io.StringIO())
+    buffer_cache = source_logger._buffer_cache
+
+    # WHEN BufferingHandler is initialized
+    handler = BufferingHandler(
+        buffer_cache=buffer_cache,
+        buffer_config=buffer_config,
+        source_logger=source_logger,
+    )
+
+    # THEN all dependencies are stored on the instance
+    assert handler.buffer_cache is buffer_cache
+    assert handler.buffer_config is buffer_config
+    assert handler.source_logger is source_logger
+    assert handler.level == logging.NOTSET
+
+
+def test_buffering_handler_emit_calls_add_log_record_to_buffer(monkeypatch):
+    # GIVEN a real Logger with buffer and a BufferingHandler (tracer id set so records are buffered)
+    monkeypatch.setenv(constants.XRAY_TRACE_ID_ENV, "1-67c39786-5908a82a246fb67f3089263f")
+    stream = io.StringIO()
+    buffer_config = LoggerBufferConfig(max_bytes=10240)
+    source_logger = Logger(service="test2", buffer_config=buffer_config, stream=stream)
+    handler = BufferingHandler(
+        buffer_cache=source_logger._buffer_cache,
+        buffer_config=source_logger._buffer_config,
+        source_logger=source_logger,
+    )
+    record = logging.LogRecord(
+        name="external",
+        level=logging.INFO,
+        pathname="",
+        lineno=0,
+        msg="test %s",
+        args=("arg",),
+        exc_info=None,
+        func=None,
+    )
+    record.stack_info = None
+
+    # WHEN the handler emits the record and the buffer is flushed
+    handler.emit(record)
+    source_logger.flush_buffer()
+
+    # THEN the buffered message appears in the logger output
+    output = stream.getvalue()
+    log_line = json.loads(output.strip())
+    assert log_line["message"] == "test arg"