test_archive_rotation.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_formatter/test_archive_rotation.py
 
"""
Purpose
-------
Validate the OI-424 formatter backup archive rotation in
``oct.formatter.oct_format``.
 
Responsibilities
----------------
- Confirm ``_archive_file`` trims old backups to the configured cap.
- Confirm pruning is scoped to the source file (other files untouched).
- Confirm ``max_backups == 0`` disables pruning.
- Confirm a per-file unlink failure is swallowed.
- Confirm ``FormatterContext.max_backups_per_file`` is threaded through
  to the archiver in the write path.
 
Diagnostics
-----------
Domain: TESTS.FORMATTER
L2: test lifecycle
L3: fixture creation, assertion details
L4: deep tracing of archive operations
 
Contracts
---------
Inputs:  tmp_path pytest fixture
Outputs: pass/fail assertions
"""
 
import os
import time
from pathlib import Path
 
import pytest
 
from oct.formatter.oct_format import (
    _DEFAULT_MAX_ARCHIVE_BYTES,
    _DEFAULT_MAX_BACKUPS_PER_FILE,
    _archive_file,
    _prune_archive,
    _prune_archive_by_size,
)
 
 
def _seed_backups(
    archive_dir: Path, source_name: str, count: int, start_age: float = 100.0,
) -> list[Path]:
    """Create ``count`` fake backups with staggered mtimes (oldest first).
 
    Returns the list of created paths in the order they were created.
    """
    archive_dir.mkdir(parents=True, exist_ok=True)
    created: list[Path] = []
    now = time.time()
    for i in range(count):
        stamp = 20260101_000000 + i
        path = archive_dir / f"{source_name}.{stamp}.bak"
        path.write_text(f"backup {i}\n", encoding="utf-8")
        # Oldest backup has the oldest mtime.
        mtime = now - (start_age - i)
        os.utime(path, (mtime, mtime))
        created.append(path)
    return created
 
 
def test_archive_keeps_most_recent_n(tmp_path: Path) -> None:
    """Archiving into a dir with 15 backups prunes to ``max_backups``."""
    source = tmp_path / "foo.py"
    source.write_text("print('x')\n", encoding="utf-8")
    archive_dir = tmp_path / ".formatter_archive"
    seeded = _seed_backups(archive_dir, "foo.py", 15)
 
    _archive_file(source, max_backups=10)
 
    remaining = sorted(archive_dir.glob("foo.py.*.bak"))
    # 10 retained total: the newly-written one plus the 9 newest seeded.
    assert len(remaining) == 10
    # The oldest 6 seeded files must be gone.
    for stale in seeded[:6]:
        assert not stale.exists()
    # The 9 most recent seeded files survive.
    for fresh in seeded[6:]:
        assert fresh.exists()
 
 
def test_archive_prune_handles_empty_dir(tmp_path: Path) -> None:
    """A fresh archive dir retains exactly one backup after archival."""
    source = tmp_path / "bar.py"
    source.write_text("print('y')\n", encoding="utf-8")
 
    _archive_file(source, max_backups=10)
 
    archive_dir = tmp_path / ".formatter_archive"
    assert len(list(archive_dir.glob("bar.py.*.bak"))) == 1
 
 
def test_archive_prune_respects_per_file_scoping(tmp_path: Path) -> None:
    """Pruning ``foo.py`` must not delete ``bar.py`` backups."""
    archive_dir = tmp_path / ".formatter_archive"
    _seed_backups(archive_dir, "foo.py", 5)
    bar_seeds = _seed_backups(archive_dir, "bar.py", 5)
 
    removed = _prune_archive(archive_dir, "foo.py", max_backups=2)
 
    assert removed == 3
    # All bar.py backups must survive.
    for bar in bar_seeds:
        assert bar.exists()
    # foo.py should have exactly 2 left.
    assert len(list(archive_dir.glob("foo.py.*.bak"))) == 2
 
 
def test_archive_prune_max_backups_zero_disables(tmp_path: Path) -> None:
    """``max_backups=0`` preserves every existing backup."""
    archive_dir = tmp_path / ".formatter_archive"
    _seed_backups(archive_dir, "foo.py", 20)
 
    removed = _prune_archive(archive_dir, "foo.py", max_backups=0)
 
    assert removed == 0
    assert len(list(archive_dir.glob("foo.py.*.bak"))) == 20
 
 
def test_archive_prune_handles_unlink_failure(
    tmp_path: Path, monkeypatch,
) -> None:
    """A single ``OSError`` during unlink must not stop pruning the rest."""
    archive_dir = tmp_path / ".formatter_archive"
    _seed_backups(archive_dir, "foo.py", 5)
 
    # Make one specific unlink raise; all others succeed.
    real_unlink = Path.unlink
    fail_name = sorted(archive_dir.glob("foo.py.*.bak"))[0].name
 
    def flaky_unlink(self, *args, **kwargs):
        if self.name == fail_name:
            raise OSError("simulated")
        return real_unlink(self, *args, **kwargs)
 
    monkeypatch.setattr(Path, "unlink", flaky_unlink)
 
    removed = _prune_archive(archive_dir, "foo.py", max_backups=2)
 
    # 3 candidates beyond the cap; 1 fails to unlink; 2 are removed.
    assert removed == 2
 
 
def test_default_max_backups_constant() -> None:
    """Sanity check on the documented default."""
    assert _DEFAULT_MAX_BACKUPS_PER_FILE == 10
 
 
# ---------------------------------------------------------------------
# OI-516 — per-directory aggregate size cap.
# ---------------------------------------------------------------------
 
 
def _mk_sized_backup(
    archive_dir: Path, name: str, size: int, mtime_offset: float,
) -> Path:
    """Create a ``.bak`` of given size with an offset mtime (seconds)."""
    archive_dir.mkdir(parents=True, exist_ok=True)
    path = archive_dir / name
    path.write_bytes(b"x" * size)
    now = time.time()
    os.utime(path, (now + mtime_offset, now + mtime_offset))
    return path
 
 
def test_size_prune_evicts_oldest_until_under_cap(tmp_path: Path) -> None:
    """OI-516: aggregate size cap evicts oldest-first."""
    archive = tmp_path / ".formatter_archive"
    _mk_sized_backup(archive, "a.py.001.bak", 200, -30)
    _mk_sized_backup(archive, "a.py.002.bak", 200, -20)
    _mk_sized_backup(archive, "a.py.003.bak", 200, -10)
 
    removed = _prune_archive_by_size(archive, max_total_bytes=500)
 
    assert removed == 1
    remaining = sorted(p.name for p in archive.glob("*.bak"))
    assert remaining == ["a.py.002.bak", "a.py.003.bak"]
 
 
def test_size_prune_noop_when_under_cap(tmp_path: Path) -> None:
    """OI-516: already under cap → nothing removed."""
    archive = tmp_path / ".formatter_archive"
    _mk_sized_backup(archive, "a.py.001.bak", 100, -10)
    assert _prune_archive_by_size(archive, max_total_bytes=10_000) == 0
    assert len(list(archive.glob("*.bak"))) == 1
 
 
def test_size_prune_disabled_with_nonpositive_cap(tmp_path: Path) -> None:
    """OI-516: ``max_total_bytes <= 0`` disables size pruning entirely."""
    archive = tmp_path / ".formatter_archive"
    _mk_sized_backup(archive, "a.py.001.bak", 500, -10)
    assert _prune_archive_by_size(archive, 0) == 0
    assert _prune_archive_by_size(archive, -1) == 0
    assert len(list(archive.glob("*.bak"))) == 1
 
 
def test_archive_file_enforces_size_cap_end_to_end(tmp_path: Path) -> None:
    """OI-516: ``_archive_file`` triggers size pruning after writing the backup."""
    source = tmp_path / "big.py"
    source.write_bytes(b"y" * 400)
 
    archive = tmp_path / ".formatter_archive"
    _mk_sized_backup(archive, "big.py.001.bak", 400, -30)
    _mk_sized_backup(archive, "big.py.002.bak", 400, -20)
    _mk_sized_backup(archive, "big.py.003.bak", 400, -10)
 
    # Cap well below the 4 × 400 = 1600-byte aggregate the write would
    # create. Count cap stays at 10 so the size cap is the active pass.
    _archive_file(source, max_backups=10, max_archive_bytes=800)
 
    total = sum(p.stat().st_size for p in archive.glob("*.bak"))
    assert total <= 800
 
 
def test_default_max_archive_bytes_constant() -> None:
    """OI-516: default aggregate cap is 5 MiB."""
    assert _DEFAULT_MAX_ARCHIVE_BYTES == 5 * 1024 * 1024