local-ci-cd-system/tests/python/test_commands_retention.py

"""Tests for ``ci_orchestrator retention run``."""

from __future__ import annotations

import shutil
import time
from pathlib import Path
from typing import Any

import pytest
from click.testing import CliRunner

import ci_orchestrator.commands.retention as ret
from ci_orchestrator.__main__ import cli


# ── helpers ───────────────────────────────────────────────────────────────────


def _fake_usage(free_gb: float, total_gb: float = 500.0) -> shutil._ntuple_diskusage:
    gb = 1024**3
    return shutil._ntuple_diskusage(
        int(total_gb * gb),
        int((total_gb - free_gb) * gb),
        int(free_gb * gb),
    )


def _make_old_dir(base: Path, name: str, age_days: int = 40) -> Path:
    d = base / name
    d.mkdir(parents=True)
    old_mtime = time.time() - age_days * 86400
    import os

    os.utime(d, (old_mtime, old_mtime))
    return d


def _make_recent_dir(base: Path, name: str) -> Path:
    d = base / name
    d.mkdir(parents=True)
    return d


# ── retention run — basic purge ───────────────────────────────────────────────


def test_purge_old_artifact_dirs(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    art = tmp_path / "artifacts"
    art.mkdir()
    old = _make_old_dir(art, "job-001")
    recent = _make_recent_dir(art, "job-002")

    monkeypatch.setattr(ret.shutil, "disk_usage", lambda _p: _fake_usage(free_gb=200))

    result = CliRunner().invoke(
        cli,
        ["retention", "run", "--artifact-dir", str(art), "--log-dir", str(tmp_path / "logs")],
    )
    assert result.exit_code == 0, result.output
    assert not old.exists(), "old dir should have been deleted"
    assert recent.exists(), "recent dir should survive"
    assert "Purged artifact" in result.output


def test_nothing_to_purge_message(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    art = tmp_path / "artifacts"
    art.mkdir()
    _make_recent_dir(art, "job-001")

    monkeypatch.setattr(ret.shutil, "disk_usage", lambda _p: _fake_usage(free_gb=200))

    result = CliRunner().invoke(
        cli,
        ["retention", "run", "--artifact-dir", str(art), "--log-dir", str(tmp_path / "logs")],
    )
    assert result.exit_code == 0, result.output
    assert "nothing to purge" in result.output


def test_missing_artifact_dir_skipped(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(ret.shutil, "disk_usage", lambda _p: _fake_usage(free_gb=200))

    result = CliRunner().invoke(
        cli,
        [
            "retention",
            "run",
            "--artifact-dir",
            str(tmp_path / "nonexistent"),
            "--log-dir",
            str(tmp_path / "logs"),
        ],
    )
    assert result.exit_code == 0, result.output
    assert "not found" in result.output


# ── aggressive mode ───────────────────────────────────────────────────────────


def test_aggressive_mode_triggered(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    art = tmp_path / "artifacts"
    art.mkdir()
    _make_old_dir(art, "job-001", age_days=10)  # 10 days old, > aggressive threshold (7d)

    monkeypatch.setattr(ret.shutil, "disk_usage", lambda _p: _fake_usage(free_gb=5))

    result = CliRunner().invoke(
        cli,
        [
            "retention",
            "run",
            "--artifact-dir",
            str(art),
            "--log-dir",
            str(tmp_path / "logs"),
            "--min-free-gb",
            "50",
            "--aggressive-retention-days",
            "7",
        ],
    )
    assert result.exit_code == 0, result.output
    assert "aggressive retention" in result.output
    assert "WARNING" in result.output


def test_aggressive_mode_not_triggered_when_space_ok(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    art = tmp_path / "artifacts"
    art.mkdir()
    monkeypatch.setattr(ret.shutil, "disk_usage", lambda _p: _fake_usage(free_gb=200))

    result = CliRunner().invoke(
        cli,
        ["retention", "run", "--artifact-dir", str(art), "--log-dir", str(tmp_path / "logs")],
    )
    assert result.exit_code == 0, result.output
    assert "Normal retention" in result.output
    assert "WARNING" not in result.output


# ── --what-if ────────────────────────────────────────────────────────────────


def test_what_if_does_not_delete(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    art = tmp_path / "artifacts"
    art.mkdir()
    old = _make_old_dir(art, "job-001")

    monkeypatch.setattr(ret.shutil, "disk_usage", lambda _p: _fake_usage(free_gb=200))

    result = CliRunner().invoke(
        cli,
        [
            "retention",
            "run",
            "--artifact-dir",
            str(art),
            "--log-dir",
            str(tmp_path / "logs"),
            "--what-if",
        ],
    )
    assert result.exit_code == 0, result.output
    assert old.exists(), "--what-if must not delete anything"
    assert "WhatIf" in result.output


# ── log dir purge ─────────────────────────────────────────────────────────────


def test_purge_old_log_dirs(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    log = tmp_path / "logs"
    log.mkdir()
    old = _make_old_dir(log, "run-001")

    monkeypatch.setattr(ret.shutil, "disk_usage", lambda _p: _fake_usage(free_gb=200))

    result = CliRunner().invoke(
        cli,
        [
            "retention",
            "run",
            "--artifact-dir",
            str(tmp_path / "artifacts"),
            "--log-dir",
            str(log),
        ],
    )
    assert result.exit_code == 0, result.output
    assert not old.exists()
    assert "Purged log" in result.output