initial commit

2026-09-30 13:41:19 +02:00
commit 36973de90a
9 changed files with 466 additions and 0 deletions
@@ -0,0 +1,210 @@
+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 uvicorn.logging import DefaultFormatter
+
+from dataflex_backend_common.logging_config import LoggingSettings
+from pydantic import ValidationError
+
+SECONDS_PER_DAY = 24 * 60 * 60
+
+
+@pytest.fixture
+def restore_root_logger() -> Generator[None, None, 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("dataflex_backend_common.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
+        monkeypatch.setattr(time, "time", lambda: rollover_timestamp)
+        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