Octop/tests/unit/test_logging.py

"""tests/unit/test_logging.py

Covers the daily-rotation + retention logging strategy in ``octop.infra.server``:
the log path lives under ``~/.octop/logs`` and stale rotated files are purged.
"""

from __future__ import annotations

import gzip
import logging
import os
import re
import time
from logging.handlers import TimedRotatingFileHandler
from pathlib import Path

import pytest

from octop.infra.server import (
    DEFAULT_LOG_MAX_BYTES,
    OctopServer,
    SizeTimedRotatingFileHandler,
    _attach_log_handler,
    _build_log_handler,
    _dated_log_namer,
    _migrate_legacy_rotated_logs,
    _parse_log_compress,
    _parse_log_max_bytes,
    _purge_stale_logs,
    delaycompress_rotated_logs,
    gzip_rotated_log,
)


@pytest.fixture(autouse=True)
def _isolate_root_logger() -> None:
    """Restore the global root/uvicorn loggers so tests never leak handlers/levels."""
    root = logging.getLogger()
    saved_root_handlers = list(root.handlers)
    saved_root_level = root.level
    saved_child = {
        name: list(logging.getLogger(name).handlers)
        for name in ("uvicorn", "uvicorn.access", "uvicorn.error")
    }
    yield
    root.handlers = saved_root_handlers
    root.setLevel(saved_root_level)
    for name, handlers in saved_child.items():
        logging.getLogger(name).handlers = handlers


def _stale_path(log_dir: Path, name: str, age_days: int) -> Path:
    """Create a rotated log file with an mtime shifted ``age_days`` into the past."""
    path = log_dir / name
    path.write_text("old\n", encoding="utf-8")
    past = time.time() - age_days * 86400
    os.utime(path, (past, past))
    return path


def test_build_log_handler_uses_daily_rotation_and_retention(tmp_path: Path):
    handler = _build_log_handler(tmp_path / "octop.log", retention_days=7, max_bytes=4096)
    try:
        assert isinstance(handler, SizeTimedRotatingFileHandler)
        assert handler.when == "MIDNIGHT"
        assert handler.backupCount == 7
        assert handler.max_bytes == 4096
        # Date-only suffix keeps filenames valid on Windows (no ':' like %H:%M:%S).
        assert handler.suffix == "%Y-%m-%d"
        assert ":" not in handler.suffix
        assert handler.namer is _dated_log_namer
    finally:
        handler.close()


def test_purge_stale_logs_removes_only_old_files(tmp_path: Path):
    log_dir = tmp_path / "logs"
    log_dir.mkdir()
    old = _stale_path(log_dir, "octop.2020-01-01.log", age_days=100)
    new = _stale_path(log_dir, "octop.2026-07-16.log", age_days=1)

    _purge_stale_logs(log_dir, retention_days=14)

    assert not old.exists()
    assert new.exists()


def test_purge_stale_logs_skipped_when_retention_non_positive(tmp_path: Path):
    log_dir = tmp_path / "logs"
    log_dir.mkdir()
    old = _stale_path(log_dir, "octop.2020-01-01.log", age_days=100)

    _purge_stale_logs(log_dir, retention_days=0)

    assert old.exists()


def test_attach_log_handler_is_idempotent(tmp_path: Path):
    root = logging.getLogger()
    handler = _build_log_handler(tmp_path / "octop.log", retention_days=7)
    try:
        _attach_log_handler(root, handler)
        _attach_log_handler(root, handler)
        matching = [
            h
            for h in root.handlers
            if isinstance(h, TimedRotatingFileHandler) and h.baseFilename == handler.baseFilename
        ]
        assert len(matching) == 1
    finally:
        handler.close()


def test_setup_logging_creates_logs_dir_and_file(tmp_path: Path, monkeypatch):
    monkeypatch.delenv("OCTOP_LOG_LEVEL", raising=False)
    monkeypatch.delenv("OCTOP_LOG_RETENTION_DAYS", raising=False)
    server = OctopServer(home=tmp_path / ".octop")

    server._setup_logging()

    assert server.paths.logs_dir.is_dir()
    root = logging.getLogger()
    handler = next(
        h
        for h in root.handlers
        if isinstance(h, TimedRotatingFileHandler) and h.baseFilename == str(server.paths.log)
    )
    # Emit a record and confirm it lands in the logs directory file.
    root.info("logging smoke test")
    handler.flush()
    assert server.paths.log.exists()
    assert "logging smoke test" in server.paths.log.read_text(encoding="utf-8")
    assert root.level == logging.INFO


def test_setup_logging_migrates_legacy_log(tmp_path: Path):
    home = tmp_path / ".octop"
    legacy = home / "octop.log"
    legacy.parent.mkdir(parents=True, exist_ok=True)
    legacy.write_text("legacy content", encoding="utf-8")
    server = OctopServer(home=home)

    server._setup_logging()

    assert not legacy.exists()
    assert server.paths.log.exists()
    assert server.paths.log.read_text(encoding="utf-8") == "legacy content"


def test_setup_logging_respects_retention_env(tmp_path: Path, monkeypatch):
    monkeypatch.setenv("OCTOP_LOG_RETENTION_DAYS", "3")
    server = OctopServer(home=tmp_path / ".octop")

    server._setup_logging()

    root = logging.getLogger()
    handler = next(
        h
        for h in root.handlers
        if isinstance(h, TimedRotatingFileHandler) and h.baseFilename == str(server.paths.log)
    )
    assert handler.backupCount == 3


def test_setup_logging_respects_level_env(tmp_path: Path, monkeypatch):
    monkeypatch.setenv("OCTOP_LOG_LEVEL", "debug")
    server = OctopServer(home=tmp_path / ".octop")

    server._setup_logging()

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


def test_setup_logging_falls_back_on_invalid_level_env(tmp_path: Path, monkeypatch):
    monkeypatch.setenv("OCTOP_LOG_LEVEL", "not-a-level")
    server = OctopServer(home=tmp_path / ".octop")

    server._setup_logging()

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


def test_setup_logging_falls_back_on_invalid_retention_env(tmp_path: Path, monkeypatch):
    monkeypatch.setenv("OCTOP_LOG_RETENTION_DAYS", "oops")
    server = OctopServer(home=tmp_path / ".octop")

    server._setup_logging()

    root = logging.getLogger()
    handler = next(
        h
        for h in root.handlers
        if isinstance(h, TimedRotatingFileHandler) and h.baseFilename == str(server.paths.log)
    )
    # Default retention is 14 days.
    assert handler.backupCount == 14


def test_parse_log_max_bytes_defaults_and_env(monkeypatch):
    monkeypatch.delenv("OCTOP_LOG_MAX_BYTES", raising=False)
    assert _parse_log_max_bytes() == DEFAULT_LOG_MAX_BYTES
    monkeypatch.setenv("OCTOP_LOG_MAX_BYTES", "2048")
    assert _parse_log_max_bytes() == 2048
    monkeypatch.setenv("OCTOP_LOG_MAX_BYTES", "bad")
    assert _parse_log_max_bytes() == DEFAULT_LOG_MAX_BYTES


def test_log_handler_rotates_when_max_bytes_exceeded(tmp_path: Path):
    log_path = tmp_path / "octop.log"
    handler = _build_log_handler(log_path, retention_days=3, max_bytes=256)
    test_logger = logging.getLogger("test.octop.log.size_roll")
    test_logger.handlers = []
    test_logger.propagate = False
    test_logger.setLevel(logging.INFO)
    test_logger.addHandler(handler)
    try:
        for _ in range(40):
            test_logger.info("x" * 40)
        handler.flush()
        backups = [p for p in tmp_path.glob("octop.*") if p != log_path]
        assert all(
            re.fullmatch(r"octop\.\d{4}-\d{2}-\d{2}\.log(?:\.\d{3})?(?:\.gz)?", p.name)
            for p in backups
        )
        assert len(backups) >= 1
        # delaycompress: newest rotated file stays plain; older ones may be .gz
        plains = [p for p in backups if not p.name.endswith(".gz")]
        gzips = [p for p in backups if p.name.endswith(".gz")]
        assert len(plains) == 1
        if len(backups) > 1:
            assert len(gzips) >= 1
        assert log_path.exists()
        assert log_path.stat().st_size < 512
    finally:
        test_logger.handlers = []
        handler.close()


def test_gzip_rotated_log_replaces_plain_file(tmp_path: Path):
    plain = tmp_path / "octop.2026-09-02.log"
    # Binary write keeps LF on Windows (text mode would expand to CRLF).
    plain.write_bytes(b"hello\nworld\n")
    gzip_rotated_log(plain)
    gz = tmp_path / "octop.2026-09-02.log.gz"
    assert not plain.exists()
    assert gz.exists()
    assert gzip.decompress(gz.read_bytes()) == b"hello\nworld\n"


def test_delaycompress_keeps_newest_plain_on_startup(tmp_path: Path):
    log_dir = tmp_path / "logs"
    log_dir.mkdir()
    older = log_dir / "octop.2026-08-28.log"
    newer = log_dir / "octop.2026-08-29.log"
    older.write_text("older\n", encoding="utf-8")
    newer.write_text("newer\n", encoding="utf-8")
    past = time.time() - 86400
    os.utime(older, (past, past))
    delaycompress_rotated_logs(log_dir, enabled=True)
    assert not older.exists()
    assert (log_dir / "octop.2026-08-28.log.gz").exists()
    assert newer.exists()
    assert not (log_dir / "octop.2026-08-29.log.gz").exists()


def test_delaycompress_single_plain_stays_uncompressed(tmp_path: Path):
    log_dir = tmp_path / "logs"
    log_dir.mkdir()
    plain = log_dir / "octop.2026-08-29.log"
    plain.write_text("only\n", encoding="utf-8")
    delaycompress_rotated_logs(log_dir, enabled=True)
    assert plain.exists()
    assert not (log_dir / "octop.2026-08-29.log.gz").exists()


def test_parse_log_compress_env(monkeypatch):
    monkeypatch.delenv("OCTOP_LOG_COMPRESS", raising=False)
    assert _parse_log_compress() is True
    monkeypatch.setenv("OCTOP_LOG_COMPRESS", "0")
    assert _parse_log_compress() is False


def test_purge_stale_logs_removes_old_gz_files(tmp_path: Path):
    log_dir = tmp_path / "logs"
    log_dir.mkdir()
    old = _stale_path(log_dir, "octop.2020-01-01.log.gz", age_days=100)
    new = _stale_path(log_dir, "octop.2026-07-16.log.gz", age_days=1)
    _purge_stale_logs(log_dir, retention_days=14)
    assert not old.exists()
    assert new.exists()


def test_setup_logging_respects_max_bytes_env(tmp_path: Path, monkeypatch):
    monkeypatch.setenv("OCTOP_LOG_MAX_BYTES", "8192")
    server = OctopServer(home=tmp_path / ".octop")
    server._setup_logging()
    handler = next(
        h
        for h in logging.getLogger().handlers
        if isinstance(h, SizeTimedRotatingFileHandler) and h.baseFilename == str(server.paths.log)
    )
    assert handler.max_bytes == 8192


def test_dated_log_namer_puts_date_before_extension():
    default = os.path.join("logs", "octop.log.2026-10-16")
    assert _dated_log_namer(default) == os.path.join("logs", "octop.2026-10-16.log")
    assert _dated_log_namer("octop.log.2026-10-16") == "octop.2026-10-16.log"


def test_rotation_filename_avoids_same_day_collision(tmp_path: Path):
    log_path = tmp_path / "octop.log"
    handler = _build_log_handler(log_path, retention_days=3, max_bytes=0, compress_rotated=False)
    try:
        first = handler.rotation_filename(str(log_path) + ".2026-10-16")
        assert Path(first).name == "octop.2026-10-16.log"
        Path(first).write_text("kept\n", encoding="utf-8")
        second = handler.rotation_filename(str(log_path) + ".2026-10-16")
        assert Path(second).name == "octop.2026-10-16.log.001"
        Path(first).with_name(Path(first).name + ".gz").write_bytes(b"gz")
        Path(first).unlink()
        third = handler.rotation_filename(str(log_path) + ".2026-10-16")
        assert Path(third).name == "octop.2026-10-16.log.001"
    finally:
        handler.close()


def test_rollover_writes_date_before_extension(tmp_path: Path):
    log_path = tmp_path / "octop.log"
    handler = _build_log_handler(log_path, retention_days=3, max_bytes=0, compress_rotated=False)
    test_logger = logging.getLogger("test.octop.log.date_name")
    test_logger.handlers = []
    test_logger.propagate = False
    test_logger.setLevel(logging.INFO)
    test_logger.addHandler(handler)
    try:
        test_logger.info("before roll")
        handler.flush()
        handler.rolloverAt = int(time.time()) - 1
        test_logger.info("after roll")
        handler.flush()
        rotated = [p for p in tmp_path.iterdir() if p != log_path]
        assert len(rotated) == 1
        assert re.fullmatch(r"octop\.\d{4}-\d{2}-\d{2}\.log", rotated[0].name)
        assert "before roll" in rotated[0].read_text(encoding="utf-8")
        assert "after roll" in log_path.read_text(encoding="utf-8")
    finally:
        test_logger.handlers = []
        handler.close()


def test_migrate_legacy_rotated_logs(tmp_path: Path):
    log_dir = tmp_path / "logs"
    log_dir.mkdir()
    (log_dir / "octop.log").write_text("active\n", encoding="utf-8")
    (log_dir / "octop.log.2026-10-15").write_text("day\n", encoding="utf-8")
    (log_dir / "octop.log.2026-10-15-001").write_text("extra\n", encoding="utf-8")
    (log_dir / "octop.log.2026-10-14.gz").write_bytes(b"gz")
    occupied = log_dir / "octop.2026-10-13.log"
    occupied.write_text("already\n", encoding="utf-8")
    (log_dir / "octop.log.2026-10-13").write_text("legacy\n", encoding="utf-8")

    _migrate_legacy_rotated_logs(log_dir)

    assert (log_dir / "octop.log").read_text(encoding="utf-8") == "active\n"
    assert (log_dir / "octop.2026-10-15.log").read_text(encoding="utf-8") == "day\n"
    assert (log_dir / "octop.2026-10-15.log.001").read_text(encoding="utf-8") == "extra\n"
    assert (log_dir / "octop.2026-10-14.log.gz").read_bytes() == b"gz"
    assert not (log_dir / "octop.log.2026-10-15").exists()
    assert occupied.read_text(encoding="utf-8") == "already\n"
    assert (log_dir / "octop.log.2026-10-13").read_text(encoding="utf-8") == "legacy\n"


def test_setup_logging_renames_legacy_rotated_files(tmp_path: Path):
    home = tmp_path / ".octop"
    log_dir = home / "logs"
    log_dir.mkdir(parents=True)
    legacy = log_dir / "octop.log.2026-10-01"
    legacy.write_text("old rotation\n", encoding="utf-8")
    server = OctopServer(home=home)

    server._setup_logging()

    assert not legacy.exists()
    assert (log_dir / "octop.2026-10-01.log").read_text(encoding="utf-8") == "old rotation\n"


def test_purge_stale_logs_still_removes_legacy_names(tmp_path: Path):
    log_dir = tmp_path / "logs"
    log_dir.mkdir()
    old = _stale_path(log_dir, "octop.log.2020-01-01", age_days=100)
    old_gz = _stale_path(log_dir, "octop.log.2020-01-02.gz", age_days=100)
    new = _stale_path(log_dir, "octop.log.2026-07-16", age_days=1)

    _purge_stale_logs(log_dir, retention_days=14)

    assert not old.exists()
    assert not old_gz.exists()
    assert new.exists()