"""Tests for the pure ``operations.cache`` layer (W2 of the web-UI plan). Covers ``cargo_cache_status`` and ``purge_cargo_cache`` — the seam both the CLI (``retention run``) and the future web UI import for cargo-cache state/maintenance. Zero ``click``/``print`` in the module under test. """ from __future__ import annotations import time from pathlib import Path from ci_orchestrator.cargo_cache import CargoCacheStore, cache_key from ci_orchestrator.config import BackendConfig, CargoCacheConfig, Config, Paths from ci_orchestrator.operations.cache import ( CacheStatus, PurgeResult, cargo_cache_status, purge_cargo_cache, ) def _make_config(root: Path, *, enabled: bool = True, retention_days: int = 30) -> Config: return Config( paths=Paths( root=root.parent, templates=root.parent / "templates", build_vms=root.parent / "build-vms", artifacts=root.parent / "artifacts", keys=root.parent / "keys", ), backend=BackendConfig(), cargo_cache=CargoCacheConfig(enabled=enabled, root=root, retention_days=retention_days), ) def _seed_entry(root: Path, digest: str, *, age_days: int, size: int = 1024) -> Path: store = CargoCacheStore(root) src = root.parent / f"seed-{digest}.tar.zst" src.parent.mkdir(parents=True, exist_ok=True) src.write_bytes(b"x" * size) entry = store.publish(cache_key(lock_digest=digest, rustc_version="r", host_triple="t"), src) stamp = time.time() - age_days * 86400 import os os.utime(entry, (stamp, stamp)) return entry # ── cargo_cache_status ────────────────────────────────────────────────────── def test_cargo_cache_status_missing_root(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" config = _make_config(root) status = cargo_cache_status(config) assert status == CacheStatus( root=root, total_bytes=0, entry_count=0, oldest_age_days=None, retention_days=30, ) def test_cargo_cache_status_empty_root(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" root.mkdir(parents=True) config = _make_config(root) status = cargo_cache_status(config) assert status.entry_count == 0 assert status.total_bytes == 0 assert status.oldest_age_days is None def test_cargo_cache_status_populated_with_aged_entries(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" _seed_entry(root, "old", age_days=90, size=2048) _seed_entry(root, "new", age_days=1, size=1024) config = _make_config(root, retention_days=30) status = cargo_cache_status(config) assert status.entry_count == 2 assert status.total_bytes == 2048 + 1024 assert status.oldest_age_days is not None assert status.oldest_age_days >= 89 assert status.retention_days == 30 assert status.root == root def test_cargo_cache_status_ignores_incoming_staging_dir(tmp_path: Path) -> None: """The ``.incoming`` staging directory used by ``publish()`` is not a real entry.""" root = tmp_path / "cache" / "cargo" _seed_entry(root, "one", age_days=1) (root / ".incoming").mkdir(parents=True, exist_ok=True) config = _make_config(root) status = cargo_cache_status(config) assert status.entry_count == 1 def test_cargo_cache_status_reports_even_when_disabled(tmp_path: Path) -> None: """Status is descriptive of disk state; ``enabled`` only gates purging.""" root = tmp_path / "cache" / "cargo" _seed_entry(root, "old", age_days=90) config = _make_config(root, enabled=False) status = cargo_cache_status(config) assert status.entry_count == 1 # ── purge_cargo_cache ──────────────────────────────────────────────────────── def test_purge_cargo_cache_dry_run_counts_only_no_delete(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" old = _seed_entry(root, "old", age_days=90, size=3 * 1024 * 1024) config = _make_config(root, retention_days=30) result = purge_cargo_cache(config, dry_run=True) assert result == PurgeResult(removed_entries=1, freed_bytes=3 * 1024 * 1024, dry_run=True) assert old.exists(), "dry_run must not delete anything" def test_purge_cargo_cache_real_purge_removes_only_old(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" old = _seed_entry(root, "old", age_days=90) fresh = _seed_entry(root, "new", age_days=1) config = _make_config(root, retention_days=30) result = purge_cargo_cache(config) assert result.removed_entries == 1 assert result.dry_run is False assert not old.exists() assert fresh.exists() def test_purge_cargo_cache_retention_override_wins_over_config(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" entry = _seed_entry(root, "mid", age_days=10) config = _make_config(root, retention_days=30) # config alone would spare this entry result = purge_cargo_cache(config, retention_days=7) assert result.removed_entries == 1 assert not entry.exists() def test_purge_cargo_cache_disabled_returns_zero_and_does_not_touch_disk(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" old = _seed_entry(root, "old", age_days=90) config = _make_config(root, enabled=False, retention_days=30) result = purge_cargo_cache(config) assert result == PurgeResult(removed_entries=0, freed_bytes=0, dry_run=False) assert old.exists() def test_purge_cargo_cache_nothing_to_purge(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" fresh = _seed_entry(root, "new", age_days=1) config = _make_config(root, retention_days=30) result = purge_cargo_cache(config) assert result.removed_entries == 0 assert result.freed_bytes == 0 assert fresh.exists() def test_purge_cargo_cache_force_removes_everything_regardless_of_age(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" fresh = _seed_entry(root, "new", age_days=0) # brand-new, well within retention mid = _seed_entry(root, "mid", age_days=1) config = _make_config(root, retention_days=30) # retention alone would spare both result = purge_cargo_cache(config, force=True) assert result.removed_entries == 2 assert not fresh.exists() assert not mid.exists() def test_purge_cargo_cache_force_dry_run_counts_all_but_keeps_files(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" fresh = _seed_entry(root, "new", age_days=0, size=2 * 1024 * 1024) config = _make_config(root, retention_days=30) result = purge_cargo_cache(config, force=True, dry_run=True) assert result == PurgeResult(removed_entries=1, freed_bytes=2 * 1024 * 1024, dry_run=True) assert fresh.exists(), "force dry_run must not delete anything" def test_purge_cargo_cache_force_disabled_still_zero(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" old = _seed_entry(root, "old", age_days=1) config = _make_config(root, enabled=False, retention_days=30) result = purge_cargo_cache(config, force=True) assert result == PurgeResult(removed_entries=0, freed_bytes=0, dry_run=False) assert old.exists() def test_purge_cargo_cache_missing_root_is_not_an_error(tmp_path: Path) -> None: root = tmp_path / "cache" / "cargo" config = _make_config(root, retention_days=30) result = purge_cargo_cache(config) assert result == PurgeResult(removed_entries=0, freed_bytes=0, dry_run=False)