"""tmp_pack_sweep: fail-closed removal of orphaned .git/objects/pack/tmp_pack_* debris. Covers every HARD SAFETY item from the A4 slice: age gate, never touch a real pack, never follow symlinks, never escape the repo's own .git, skip missing paths cleanly, dry-run default, statvfs-based freed accounting, and the never-invoke-git-gc-or-prune constraint. """ from __future__ import annotations import json import os import subprocess import sys import time from pathlib import Path LIB = Path(__file__).resolve().parents[1] / "lib" sys.path.insert(0, str(LIB)) import tmp_pack_sweep # noqa: E402 from tmp_pack_sweep import ( # noqa: E402 find_repo_tmp_packs, pack_dir_for_repo, sweep_tmp_packs, ) SWEEP_PY = LIB / "tmp_pack_sweep.py" DISK_MAINTAIN = Path(__file__).resolve().parents[1] / "slices" / "bin" / "disk-maintain" def git(cwd: Path, args: list[str]) -> str: result = subprocess.run( ["git", "-C", str(cwd), *args], capture_output=True, text=True, check=False, ) if result.returncode != 0: raise RuntimeError(f"git {' '.join(args)} failed: {result.stderr}") return result.stdout def make_repo(projects_root: Path, name: str = "repo") -> Path: repo_root = projects_root / name repo_root.mkdir(parents=True) git(repo_root, ["init", "-q", "-b", "main"]) git(repo_root, ["config", "user.email", "test@test"]) git(repo_root, ["config", "user.name", "test"]) (repo_root / "README").write_text("x", encoding="utf-8") git(repo_root, ["add", "-A"]) git(repo_root, ["commit", "-qm", "init"]) return repo_root def add_worktree(repo_root: Path, slug: str) -> Path: wt_path = repo_root / ".worktrees" / slug wt_path.parent.mkdir(parents=True, exist_ok=True) git(repo_root, ["worktree", "add", "-q", "-b", f"task/{slug}", str(wt_path), "main"]) return wt_path def touch_tmp_pack(pack_dir: Path, name: str, age_s: float, size_bytes: int = 4096) -> Path: pack_dir.mkdir(parents=True, exist_ok=True) target = pack_dir / name target.write_bytes(b"\0" * size_bytes) old = time.time() - age_s os.utime(target, (old, old)) return target # --------------------------------------------------------------------------- # pack dir resolution # --------------------------------------------------------------------------- def test_pack_dir_resolves_for_plain_repo(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) pack_dir = pack_dir_for_repo(str(repo)) assert pack_dir == str(repo / ".git" / "objects" / "pack") def test_pack_dir_resolves_through_linked_worktree_commondir(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) wt = add_worktree(repo, "feature") # Linked worktrees never get their own objects/pack dir. assert not (wt / ".git" / "objects").exists() pack_dir = pack_dir_for_repo(str(wt)) assert pack_dir == str((repo / ".git" / "objects" / "pack").resolve()) def test_pack_dir_none_for_missing_repo(tmp_path: Path): assert pack_dir_for_repo(str(tmp_path / "does-not-exist")) is None def test_pack_dir_none_when_git_is_symlink(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) real_git = repo / ".git" fake_repo = projects / "fake" fake_repo.mkdir() (fake_repo / ".git").symlink_to(real_git, target_is_directory=True) assert pack_dir_for_repo(str(fake_repo)) is None # --------------------------------------------------------------------------- # HARD SAFETY: eligibility # --------------------------------------------------------------------------- def test_removes_only_old_tmp_pack_files(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) pack_dir = repo / ".git" / "objects" / "pack" old = touch_tmp_pack(pack_dir, "tmp_pack_abc123", age_s=200 * 60) found = find_repo_tmp_packs(str(repo), now_s=time.time(), min_age_s=90 * 60) assert found == [str(old)] def test_never_touches_file_younger_than_threshold(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) pack_dir = repo / ".git" / "objects" / "pack" touch_tmp_pack(pack_dir, "tmp_pack_recent", age_s=5 * 60) found = find_repo_tmp_packs(str(repo), now_s=time.time(), min_age_s=90 * 60) assert found == [] summary = sweep_tmp_packs({"apply": True, "project_roots": [str(projects)], "min_age_s": 90 * 60}) assert summary["removed"] == [] assert (pack_dir / "tmp_pack_recent").exists() def test_never_touches_real_pack_files_even_if_old(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) pack_dir = repo / ".git" / "objects" / "pack" pack_dir.mkdir(parents=True, exist_ok=True) real_pack = pack_dir / "pack-deadbeef.pack" real_idx = pack_dir / "pack-deadbeef.idx" real_rev = pack_dir / "pack-deadbeef.rev" for f in (real_pack, real_idx, real_rev): f.write_bytes(b"\0" * 128) old = time.time() - 999 * 60 os.utime(f, (old, old)) found = find_repo_tmp_packs(str(repo), now_s=time.time(), min_age_s=90 * 60) assert found == [] summary = sweep_tmp_packs({"apply": True, "project_roots": [str(projects)], "min_age_s": 90 * 60}) assert summary["removed"] == [] for f in (real_pack, real_idx, real_rev): assert f.exists() def test_never_follows_symlinked_tmp_pack_name(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) pack_dir = repo / ".git" / "objects" / "pack" pack_dir.mkdir(parents=True, exist_ok=True) outside_target = tmp_path / "outside-secret" outside_target.write_bytes(b"do-not-delete") link = pack_dir / "tmp_pack_evil" link.symlink_to(outside_target) old = time.time() - 999 * 60 os.utime(link, (old, old), follow_symlinks=False) found = find_repo_tmp_packs(str(repo), now_s=time.time(), min_age_s=90 * 60) assert found == [] summary = sweep_tmp_packs({"apply": True, "project_roots": [str(projects)], "min_age_s": 90 * 60}) assert summary["removed"] == [] assert outside_target.exists() assert outside_target.read_bytes() == b"do-not-delete" assert link.is_symlink() def test_skips_missing_or_unreadable_path_without_erroring(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() # A directory entry that discover_repos would find but that has no .git at all. (projects / "not-a-repo").mkdir() summary = sweep_tmp_packs({"apply": True, "project_roots": [str(projects)]}) assert summary["errors"] == [] assert summary["removed"] == [] def test_skips_unreadable_pack_dir_permission_denied(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) pack_dir = repo / ".git" / "objects" / "pack" touch_tmp_pack(pack_dir, "tmp_pack_locked", age_s=200 * 60) os.chmod(pack_dir, 0o000) try: summary = sweep_tmp_packs({"apply": True, "project_roots": [str(projects)]}) finally: os.chmod(pack_dir, 0o755) assert summary["errors"] == [] assert summary["removed"] == [] # --------------------------------------------------------------------------- # freed-space accounting: real statvfs delta, not summed du # --------------------------------------------------------------------------- def test_freed_bytes_is_statvfs_delta_not_zero_on_real_removal(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) pack_dir = repo / ".git" / "objects" / "pack" touch_tmp_pack(pack_dir, "tmp_pack_big", age_s=200 * 60, size_bytes=4 * 1024 * 1024) summary = sweep_tmp_packs({"apply": True, "project_roots": [str(projects)], "min_age_s": 90 * 60}) assert len(summary["removed"]) == 1 assert isinstance(summary["freedBytes"], int) assert summary["freedBytes"] >= 0 def test_dry_run_reports_zero_freed_and_removes_nothing(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) pack_dir = repo / ".git" / "objects" / "pack" target = touch_tmp_pack(pack_dir, "tmp_pack_dry", age_s=200 * 60) summary = sweep_tmp_packs({"apply": False, "project_roots": [str(projects)], "min_age_s": 90 * 60}) assert summary["removed"] == [] assert summary["freedBytes"] == 0 assert summary["dryRun"] is True assert target.exists() # --------------------------------------------------------------------------- # CLI: dry-run is the default # --------------------------------------------------------------------------- def test_cli_defaults_to_dry_run(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) pack_dir = repo / ".git" / "objects" / "pack" target = touch_tmp_pack(pack_dir, "tmp_pack_cli", age_s=200 * 60) result = subprocess.run( [sys.executable, str(SWEEP_PY), "--min-age-s", str(90 * 60)], env={**os.environ, "SM_TMP_PACK_PROJECT_ROOTS": str(projects)}, capture_output=True, text=True, check=False, ) assert result.returncode == 0, result.stderr payload = json.loads(result.stdout) assert payload["dryRun"] is True assert payload["removed"] == 0 assert target.exists() def test_cli_apply_actually_removes(tmp_path: Path): projects = tmp_path / "projects" projects.mkdir() repo = make_repo(projects) pack_dir = repo / ".git" / "objects" / "pack" target = touch_tmp_pack(pack_dir, "tmp_pack_cli_apply", age_s=200 * 60) result = subprocess.run( [sys.executable, str(SWEEP_PY), "--apply", "--min-age-s", str(90 * 60)], env={**os.environ, "SM_TMP_PACK_PROJECT_ROOTS": str(projects)}, capture_output=True, text=True, check=False, ) assert result.returncode == 0, result.stderr payload = json.loads(result.stdout) assert payload["removed"] == 1 assert not target.exists() # --------------------------------------------------------------------------- # never invoke git gc / git prune — this sweep is pure filesystem hygiene # --------------------------------------------------------------------------- def test_module_never_shells_out_to_git(tmp_path: Path): # Pure filesystem hygiene: no subprocess anywhere, so it is structurally # impossible for this module to invoke `git gc`/`git prune`/anything else. text = SWEEP_PY.read_text(encoding="utf-8") assert "subprocess" not in text assert "os.system" not in text def test_disk_maintain_wires_tmp_pack_sweep_without_gc_or_prune(): text = DISK_MAINTAIN.read_text(encoding="utf-8") assert "tmp_pack_sweep.py" in text assert "run_tmp_pack_sweep" in text assert "git gc" not in text assert "git prune" not in text # --------------------------------------------------------------------------- # disk-maintain integration: summary line carries tmp-pack:N(MB) # --------------------------------------------------------------------------- def test_disk_maintain_surfaces_tmp_pack_in_summary(tmp_path: Path): xdg = tmp_path / "xdg" home = tmp_path / "home" projects = tmp_path / "projects" home.mkdir() projects.mkdir() proc_root = tmp_path / "proc" proc_root.mkdir() repo = make_repo(projects) pack_dir = repo / ".git" / "objects" / "pack" touch_tmp_pack(pack_dir, "tmp_pack_wired", age_s=200 * 60) env = { "HOME": str(home), "XDG_STATE_HOME": str(xdg), "SM_WORKTREE_GC_PROJECT_ROOTS": str(projects), "SM_WORKTREE_GC_PROC_ROOT": str(proc_root), "SM_TMP_PACK_PROJECT_ROOTS": str(projects), "SM_TMP_PACK_MIN_AGE_S": str(90 * 60), "SM_MAINTAIN_LOCK_WAIT_S": "0", "SM_BUDGET_UV_GIB": "9999", "SM_BUDGET_PNPM_GIB": "9999", "SM_BUDGET_PODMAN_GIB": "9999", } result = subprocess.run( ["bash", str(DISK_MAINTAIN)], env={**os.environ, **env}, capture_output=True, text=True, check=False, ) assert result.returncode == 0, result.stdout + result.stderr assert "tmp-pack:1(" in result.stdout assert not (pack_dir / "tmp_pack_wired").exists()