utils/test/test_logging_config.py

import logging
import re
import time
from collections.abc import Generator
from logging.handlers import TimedRotatingFileHandler
from pathlib import Path
from unittest.mock import patch

import pytest
from pydantic import ValidationError
from uvicorn.logging import DefaultFormatter

from utils.logging_config import LoggingSettings

SECONDS_PER_DAY = 24 * 60 * 60


@pytest.fixture
def restore_root_logger() -> Generator[None]:
    root = logging.getLogger()
    old_handlers = root.handlers[:]
    old_level = root.level
    for handler in old_handlers:
        root.removeHandler(handler)

    try:
        yield
    finally:
        for handler in root.handlers:
            root.removeHandler(handler)
            handler.close()
        for handler in old_handlers:
            root.addHandler(handler)
        root.setLevel(old_level)


def test_sets_valid_log_level(restore_root_logger: None) -> None:
    LoggingSettings(LOG_LEVEL="DEBUG", LOG_TO_FILE=False)

    assert logging.getLogger().level == logging.DEBUG


def test_invalid_log_level_falls_back_to_info(restore_root_logger: None) -> None:
    with patch("utils.logging_config.logger.error") as log_error:
        LoggingSettings(LOG_LEVEL="NOT_A_LEVEL", LOG_TO_FILE=False)

    assert logging.getLogger().level == logging.INFO
    log_error.assert_called_once_with(
        "Invalid log level %r; falling back to INFO.", "NOT_A_LEVEL"
    )


def test_configures_console_handler_only(restore_root_logger: None) -> None:
    LoggingSettings(LOG_TO_FILE=False)

    handlers = logging.getLogger().handlers
    assert len(handlers) == 1
    assert type(handlers[0]) is logging.StreamHandler
    assert isinstance(handlers[0].formatter, DefaultFormatter)


def test_configures_file_handler_and_creates_parent_dir(
    tmp_path: Path,
    restore_root_logger: None,
) -> None:
    log_file = tmp_path / "nested" / "forecast.log"
    LoggingSettings(LOG_TO_FILE=True, LOG_FILE_PATH=str(log_file))

    handlers = logging.getLogger().handlers
    assert any(type(handler) is logging.StreamHandler for handler in handlers)
    file_handlers = [
        handler for handler in handlers if isinstance(handler, logging.FileHandler)
    ]
    assert len(file_handlers) == 1
    assert log_file.parent.exists()
    assert isinstance(file_handlers[0], TimedRotatingFileHandler)
    assert file_handlers[0].when == "MIDNIGHT"
    assert file_handlers[0].interval == SECONDS_PER_DAY
    assert file_handlers[0].backupCount == 30
    assert file_handlers[0].encoding is not None
    assert file_handlers[0].encoding.lower() == "utf-8"


def test_configures_custom_log_backup_count(
    tmp_path: Path, restore_root_logger: None
) -> None:
    log_file = tmp_path / "app.log"
    LoggingSettings(LOG_TO_FILE=True, LOG_FILE_PATH=str(log_file), LOG_BACKUP_COUNT=7)

    file_handler = next(
        handler
        for handler in logging.getLogger().handlers
        if isinstance(handler, TimedRotatingFileHandler)
    )

    assert file_handler.backupCount == 7


def test_writes_records_to_console_and_file(
    tmp_path: Path,
    capsys: pytest.CaptureFixture[str],
    restore_root_logger: None,
) -> None:
    log_file = tmp_path / "app.log"
    LoggingSettings(LOG_FILE_PATH=str(log_file))

    logging.getLogger("test.logging_output").info("record is written")

    file_handler = next(
        handler
        for handler in logging.getLogger().handlers
        if isinstance(handler, TimedRotatingFileHandler)
    )
    file_handler.flush()

    assert "record is written" in capsys.readouterr().err
    assert "record is written" in log_file.read_text(encoding="utf-8")


def test_rotates_active_log_file(tmp_path: Path, restore_root_logger: None) -> None:
    log_file = tmp_path / "app.log"
    LoggingSettings(LOG_FILE_PATH=str(log_file))
    logger = logging.getLogger("test.logging_rotation")
    file_handler = next(
        handler
        for handler in logging.getLogger().handlers
        if isinstance(handler, TimedRotatingFileHandler)
    )

    logger.info("before rollover")
    file_handler.flush()
    file_handler.doRollover()
    logger.info("after rollover")
    file_handler.flush()

    rotated_files = list(tmp_path.glob("app.*.log"))
    assert len(rotated_files) == 1
    assert re.fullmatch(r"app\.\d{4}-\d{2}-\d{2}\.log", rotated_files[0].name)
    assert "before rollover" not in log_file.read_text(encoding="utf-8")
    assert "after rollover" in log_file.read_text(encoding="utf-8")


def test_removes_old_rotated_files_after_backup_limit(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    restore_root_logger: None,
) -> None:
    log_file = tmp_path / "app.log"
    LoggingSettings(LOG_FILE_PATH=str(log_file), LOG_BACKUP_COUNT=2)
    logger = logging.getLogger("test.logging_retention")
    file_handler = next(
        handler
        for handler in logging.getLogger().handlers
        if isinstance(handler, TimedRotatingFileHandler)
    )
    base_timestamp = int(time.time())

    for day in range(4):
        logger.info("record from day %s", day)
        file_handler.flush()
        rollover_timestamp = base_timestamp + day * SECONDS_PER_DAY
        file_handler.rolloverAt = rollover_timestamp + SECONDS_PER_DAY
        # Bind the loop variable via a default argument to avoid late binding (B023).
        monkeypatch.setattr(time, "time", lambda now=rollover_timestamp: now)
        file_handler.doRollover()

    assert len(list(tmp_path.glob("app.*.log"))) == 2


def test_rejects_negative_log_backup_count(restore_root_logger: None) -> None:
    with pytest.raises(ValidationError):
        LoggingSettings(LOG_TO_FILE=False, LOG_BACKUP_COUNT=-1)


def test_setup_logging_is_idempotent(tmp_path: Path, restore_root_logger: None) -> None:
    log_file = tmp_path / "app.log"
    settings = LoggingSettings(LOG_TO_FILE=True, LOG_FILE_PATH=str(log_file))

    first_file_handler = next(
        handler
        for handler in logging.getLogger().handlers
        if isinstance(handler, logging.FileHandler)
    )
    first_handler_count = len(logging.getLogger().handlers)

    settings.setup_logging()
    second_handler_count = len(logging.getLogger().handlers)

    assert first_handler_count == 2
    assert second_handler_count == first_handler_count
    assert first_file_handler.stream is None


def test_applies_formatter_from_settings(restore_root_logger: None) -> None:
    log_format = "%(levelname)s|%(message)s"
    date_format = "%Y"
    LoggingSettings(
        LOG_FORMAT=log_format,
        LOG_DATE_FORMAT=date_format,
        LOG_TO_FILE=False,
    )

    handlers = logging.getLogger().handlers
    assert len(handlers) == 1
    formatter = handlers[0].formatter
    assert formatter is not None
    assert formatter._fmt == log_format
    assert formatter.datefmt == date_format


def test_logging_settings_can_be_overridden_by_environment(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    restore_root_logger: None,
) -> None:
    log_file = tmp_path / "custom" / "application.log"
    monkeypatch.setenv("LOG_LEVEL", "DEBUG")
    monkeypatch.setenv("LOG_TO_FILE", "true")
    monkeypatch.setenv("LOG_FILE_PATH", str(log_file))
    monkeypatch.setenv("LOG_BACKUP_COUNT", "90")

    settings = LoggingSettings()

    assert settings.LOG_LEVEL == "DEBUG"
    assert settings.LOG_TO_FILE is True
    assert settings.LOG_FILE_PATH == str(log_file)
    assert settings.LOG_BACKUP_COUNT == 90