Add new tests. Fix cyclomatic complexity check error

Author: Michal Ploski

aws_lambda_powertools/logging/utils.py

@@ -8,23 +8,58 @@
 
 def copy_config_to_registered_loggers(
     source_logger: PowertoolsLogger, exclude: Optional[List[str]] = None, include: Optional[List[str]] = None
-):
-    root_loggers = [
+) -> None:
+    """Enable powertools logging for imported libraries.
+
+    Attach source logger handlers to external loggers.
+    Modify logger level based on source logger attribute.
+    """
+    root_loggers = _find_root_loggers(source_logger, exclude, include)
+    for logger in root_loggers:
+        _configure_logger(source_logger, logger)
+
+
+def _include_root_loggers_filter(logger_list: List[str]):
+    return [
+        logging.getLogger(name) for name in logging.root.manager.loggerDict if "." not in name and name in logger_list
+    ]
+
+
+def _exclude_root_loggers_filter(logger_list: List[str]) -> List[logging.Logger]:
+    return [
         logging.getLogger(name)
         for name in logging.root.manager.loggerDict
-        if "." not in name and name != source_logger.name
+        if "." not in name and name not in logger_list
     ]
-    source_logger.debug(f"Found registered root loggers: {root_loggers}")
+
+
+def _find_root_loggers(
+    source_logger: PowertoolsLogger, exclude: Optional[List[str]] = None, include: Optional[List[str]] = None
+) -> list[logging.Logger]:
+    """Filter root loggers based on provided parameters.
+
+    Ensure powertools logger itself is excluded from final list.
+    """
+    root_loggers = []
     if include and not exclude:
-        root_loggers = [logger for logger in root_loggers if logger.name in include]
-    elif not include and exclude:
-        root_loggers = [logger for logger in root_loggers if logger.name not in exclude]
+        root_loggers = _include_root_loggers_filter(logger_list=include)
     elif include and exclude:
-        root_loggers = [logger for logger in root_loggers if logger.name in include and logger.name not in exclude]
+        exclude = [source_logger.name, *exclude]
+        root_loggers = _include_root_loggers_filter(logger_list=list(set(include) - set(exclude)))
+    elif not include and exclude:
+        exclude = [source_logger.name, *exclude]
+        root_loggers = _exclude_root_loggers_filter(logger_list=exclude)
+    else:
+        root_loggers = _exclude_root_loggers_filter(logger_list=[source_logger.name])
 
     source_logger.debug(f"Filtered root loggers: {root_loggers}")
-    for logger in root_loggers:
-        logger.handlers = []
-        logger.setLevel(source_logger.level)
-        for source_handler in source_logger.handlers:
-            logger.addHandler(source_handler)
+    return root_loggers
+
+
+def _configure_logger(source_logger: PowertoolsLogger, logger: logging.Logger) -> None:
+    logger.handlers = []
+    logger.setLevel(source_logger.level)
+    source_logger.debug(f"Logger {logger} reconfigured to use logging level {source_logger.level}")
+    for source_handler in source_logger.handlers:
+        logger.addHandler(source_handler)
+        source_logger.debug(f"Logger {logger} reconfigured to use {source_handler}")

tests/functional/test_logger_utils.py

@@ -25,15 +25,6 @@ class LogLevel(Enum):
     return LogLevel
 
 
-def capture_logging_output(stdout):
-    return json.loads(stdout.getvalue().strip())
-
-
-def service_name():
-    chars = string.ascii_letters + string.digits
-    return "".join(random.SystemRandom().choice(chars) for _ in range(15))
-
-
 @pytest.fixture
 def logger(stdout, log_level):
     def _logger():
@@ -44,30 +35,43 @@ def _logger():
     return _logger
 
 
+def capture_logging_output(stdout):
+    return json.loads(stdout.getvalue().strip())
+
+
+def capture_multiple_logging_statements_output(stdout):
+    return [json.loads(line.strip()) for line in stdout.getvalue().split("\n") if line]
+
+
+def service_name():
+    chars = string.ascii_letters + string.digits
+    return "".join(random.SystemRandom().choice(chars) for _ in range(15))
+
+
 def test_copy_config_to_ext_loggers(stdout, logger, log_level):
 
     msg = "test message"
 
     # GIVEN a external logger and powertools logger initialized
-    logger = logger()
-    logger_initial_handlers = logger.handlers.copy()
-    logger_initial_level = logger.level
+    logger_1 = logger()
+    logger_2 = logger()
+
     powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout)
 
     # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used
     utils.copy_config_to_registered_loggers(source_logger=powertools_logger)
-    logger.info(msg)
-    log = capture_logging_output(stdout)
+    logger_1.info(msg)
+    logger_2.info(msg)
+    logs = capture_multiple_logging_statements_output(stdout)
 
     # THEN
-    assert not logger_initial_handlers
-    assert logger_initial_level == log_level.NOTSET.value
-    assert len(logger.handlers) == 1
-    assert type(logger.handlers[0]) is logging.StreamHandler
-    assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter
-    assert logger.level == log_level.INFO.value
-    assert log["message"] == msg
-    assert log["level"] == log_level.INFO.name
+    for index, logger in enumerate([logger_1, logger_2]):
+        assert len(logger.handlers) == 1
+        assert type(logger.handlers[0]) is logging.StreamHandler
+        assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter
+        assert logger.level == log_level.INFO.value
+        assert logs[index]["message"] == msg
+        assert logs[index]["level"] == log_level.INFO.name
 
 
 def test_copy_config_to_ext_loggers_include(stdout, logger, log_level):
@@ -134,6 +138,7 @@ def test_copy_config_to_ext_loggers_include_exclude(stdout, logger, log_level):
     )
     logger_2.info(msg)
     log = capture_logging_output(stdout)
+
     # THEN
     assert not logger_1.handlers
     assert len(logger_2.handlers) == 1
@@ -142,3 +147,20 @@ def test_copy_config_to_ext_loggers_include_exclude(stdout, logger, log_level):
     assert logger_2.level == log_level.INFO.value
     assert log["message"] == msg
     assert log["level"] == log_level.INFO.name
+
+
+def test_copy_config_to_ext_loggers_clean_old_handlers(stdout, logger, log_level):
+
+    # GIVEN a external logger with handler and powertools logger initialized
+    logger = logger()
+    handler = logging.FileHandler("logfile")
+    logger.addHandler(handler)
+    powertools_logger = Logger(service=service_name(), level=log_level.INFO.value, stream=stdout)
+
+    # WHEN configuration copied from powertools logger to ALL external loggers AND our external logger used
+    utils.copy_config_to_registered_loggers(source_logger=powertools_logger)
+
+    # THEN
+    assert len(logger.handlers) == 1
+    assert type(logger.handlers[0]) is logging.StreamHandler
+    assert type(logger.handlers[0].formatter) is formatter.LambdaPowertoolsFormatter