Merge branch 'main' into pre-commit-ci-update-config

anta/logger.py

@@ -9,15 +9,13 @@
 import traceback
 from datetime import timedelta
 from enum import Enum
-from typing import TYPE_CHECKING, Literal
+from pathlib import Path
+from typing import Literal
 
 from rich.logging import RichHandler
 
 from anta import __DEBUG__
 
-if TYPE_CHECKING:
-    from pathlib import Path
-
 logger = logging.getLogger(__name__)
 
 
@@ -69,27 +67,59 @@ def setup_logging(level: LogLevel = Log.INFO, file: Path | None = None) -> None:
         # httpx as well
         logging.getLogger("httpx").setLevel(logging.WARNING)
 
-    # Add RichHandler for stdout
-    rich_handler = RichHandler(markup=True, rich_tracebacks=True, tracebacks_show_locals=False)
-    # Show Python module in stdout at DEBUG level
-    fmt_string = "[grey58]\\[%(name)s][/grey58] %(message)s" if loglevel == logging.DEBUG else "%(message)s"
-    formatter = logging.Formatter(fmt=fmt_string, datefmt="[%X]")
-    rich_handler.setFormatter(formatter)
-    root.addHandler(rich_handler)
-    # Add FileHandler if file is provided
-    if file:
+    # Add RichHandler for stdout if not already present
+    _maybe_add_rich_handler(loglevel, root)
+
+    # Add FileHandler if file is provided and same File Handler is not already present
+    if file and not _get_file_handler(root, file):
         file_handler = logging.FileHandler(file)
         formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
         file_handler.setFormatter(formatter)
         root.addHandler(file_handler)
         # If level is DEBUG and file is provided, do not send DEBUG level to stdout
-        if loglevel == logging.DEBUG:
+        if loglevel == logging.DEBUG and (rich_handler := _get_rich_handler(root)) is not None:
             rich_handler.setLevel(logging.INFO)
 
     if __DEBUG__:
         logger.debug("ANTA Debug Mode enabled")
 
 
+def _get_file_handler(logger_instance: logging.Logger, file: Path) -> logging.FileHandler | None:
+    """Return the FileHandler if present."""
+    return (
+        next(
+            (
+                handler
+                for handler in logger_instance.handlers
+                if isinstance(handler, logging.FileHandler) and str(Path(handler.baseFilename).resolve()) == str(file.resolve())
+            ),
+            None,
+        )
+        if logger_instance.hasHandlers()
+        else None
+    )
+
+
+def _get_rich_handler(logger_instance: logging.Logger) -> logging.Handler | None:
+    """Return the ANTA Rich Handler."""
+    return next((handler for handler in logger_instance.handlers if handler.get_name() == "ANTA_RICH_HANDLER"), None) if logger_instance.hasHandlers() else None
+
+
+def _maybe_add_rich_handler(loglevel: int, logger_instance: logging.Logger) -> None:
+    """Add RichHandler for stdout if not already present."""
+    if _get_rich_handler(logger_instance) is not None:
+        # Nothing to do.
+        return
+
+    anta_rich_handler = RichHandler(markup=True, rich_tracebacks=True, tracebacks_show_locals=False)
+    anta_rich_handler.set_name("ANTA_RICH_HANDLER")
+    # Show Python module in stdout at DEBUG level
+    fmt_string = "[grey58]\\[%(name)s][/grey58] %(message)s" if loglevel == logging.DEBUG else "%(message)s"
+    formatter = logging.Formatter(fmt=fmt_string, datefmt="[%X]")
+    anta_rich_handler.setFormatter(formatter)
+    logger_instance.addHandler(anta_rich_handler)
+
+
 def format_td(seconds: float, digits: int = 3) -> str:
     """Return a formatted string from a float number representing seconds and a number of digits."""
     isec, fsec = divmod(round(seconds * 10**digits), 10**digits)

tests/units/test_logger.py

@@ -6,11 +6,54 @@
 from __future__ import annotations
 
 import logging
+from pathlib import Path
 from unittest.mock import patch
 
 import pytest
 
-from anta.logger import anta_log_exception, exc_to_str, tb_to_str
+from anta.logger import Log, LogLevel, _get_file_handler, _get_rich_handler, anta_log_exception, exc_to_str, setup_logging, tb_to_str
+
+
+@pytest.mark.parametrize(
+    ("level", "path", "debug_value"),
+    [
+        pytest.param(Log.INFO, None, False, id="INFO no file"),
+        pytest.param(Log.DEBUG, None, False, id="DEBUG no file"),
+        pytest.param(Log.INFO, Path("/tmp/file.log"), False, id="INFO file"),
+        pytest.param(Log.DEBUG, Path("/tmp/file.log"), False, id="DEBUG file"),
+        pytest.param(Log.INFO, None, True, id="INFO no file __DEBUG__ set"),
+        pytest.param(Log.DEBUG, None, True, id="INFO no file __DEBUG__ set"),
+    ],
+)
+def test_setup_logging(level: LogLevel, path: Path | None, debug_value: bool) -> None:
+    """Test setup_logging."""
+    # Clean up any logger on root
+    root = logging.getLogger()
+    if root.hasHandlers():
+        root.handlers = []
+
+    with patch("anta.logger.__DEBUG__", new=debug_value):
+        setup_logging(level, path)
+
+    rich_handler = _get_rich_handler(root)
+    assert rich_handler is not None
+
+    # When __DEBUG__ is True, the log level is overwritten to DEBUG
+    if debug_value:
+        assert root.level == logging.DEBUG
+        if path is not None:
+            assert rich_handler.level == logging.INFO
+
+    if path is not None:
+        assert _get_file_handler(root, path) is not None
+        expected_handlers = 2
+    else:
+        expected_handlers = 1
+    assert len(root.handlers) == expected_handlers
+
+    # Check idempotency
+    setup_logging(level, path)
+    assert len(root.handlers) == expected_handlers
 
 
 @pytest.mark.parametrize(