"""Phase 4 — RL_18–RL_20: GC integration with reflog expiry. TDD: tests written before implementation. Coverage tiers -------------- RL_18 muse gc --json prunes old reflog entries; output has reflog_expired field RL_19 muse gc --dry-run --json reports reflog_expired count; writes nothing RL_20 Integration: entries older than 90 days gone after muse gc; verified on disk """ from __future__ import annotations import json import pathlib import time import pytest from muse.core.paths import muse_dir, logs_dir from muse.core.types import fake_id from tests.cli_test_helper import CliRunner runner = CliRunner() cli = None _DAY = 86_400 # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _env(root: pathlib.Path) -> dict[str, str]: return {"MUSE_REPO_ROOT": str(root)} def _make_repo(tmp_path: pathlib.Path, branch: str = "main") -> pathlib.Path: """Minimal .muse/ repo with no commits and an empty config.""" from muse._version import __version__ dot = muse_dir(tmp_path) for sub in ( "refs/heads", "objects/sha256", "commits", "snapshots", "logs/refs/heads", "logs", ): (dot / sub).mkdir(parents=True, exist_ok=True) (dot / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"}) ) (dot / "HEAD").write_text(f"ref: refs/heads/{branch}\n") (dot / "config.toml").write_text("") return tmp_path def _head_log(root: pathlib.Path) -> pathlib.Path: return logs_dir(root) / "HEAD" def _branch_log(root: pathlib.Path, branch: str) -> pathlib.Path: return logs_dir(root) / "refs" / "heads" / branch def _write_entries(log_path: pathlib.Path, entries: list[tuple[int, str]]) -> None: """Write raw reflog entries at specified unix timestamps.""" log_path.parent.mkdir(parents=True, exist_ok=True) old = fake_id("old") new = fake_id("new") lines = [f"{old} {new} user {ts} +0000\t{op}\n" for ts, op in entries] log_path.write_text("".join(lines), encoding="utf-8") def _now() -> int: return int(time.time()) def _old_ts(days: int) -> int: return _now() - days * _DAY def _recent_ts(days: int = 1) -> int: return _now() - days * _DAY # --------------------------------------------------------------------------- # RL_18 — muse gc --json has reflog_expired field and prunes old entries # --------------------------------------------------------------------------- class TestGcReflogExpired: """RL_18: gc --json gains reflog_expired and actually prunes old entries.""" def test_json_output_has_reflog_expired_field(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) # Write one recent entry so the log exists but nothing expires. _write_entries(_head_log(root), [(_recent_ts(5), "recent")]) r = runner.invoke(cli, ["gc", "--json", "--grace-period", "0"], env=_env(root)) assert r.exit_code == 0, r.output data = json.loads(r.output) assert "reflog_expired" in data, ( f"'reflog_expired' missing from gc JSON output; keys: {list(data)}" ) def test_old_entries_counted_in_reflog_expired(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_entries(_head_log(root), [ (_old_ts(100), "old-a"), (_old_ts(95), "old-b"), (_recent_ts(5), "recent"), ]) r = runner.invoke( cli, ["gc", "--json", "--grace-period", "0"], env=_env(root), ) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["reflog_expired"] == 2 def test_recent_entries_not_expired(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_entries(_head_log(root), [ (_recent_ts(5), "a"), (_recent_ts(10), "b"), ]) r = runner.invoke(cli, ["gc", "--json", "--grace-period", "0"], env=_env(root)) data = json.loads(r.output) assert data["reflog_expired"] == 0 def test_gc_expires_branch_logs_too(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) # HEAD log: 1 old _write_entries(_head_log(root), [(_old_ts(100), "head-old")]) # branch log: 1 old _write_entries(_branch_log(root, "main"), [(_old_ts(100), "main-old")]) r = runner.invoke(cli, ["gc", "--json", "--grace-period", "0"], env=_env(root)) data = json.loads(r.output) # 2 total expired (1 from HEAD + 1 from main) assert data["reflog_expired"] == 2 def test_reflog_expired_zero_when_no_log_files(self, tmp_path: pathlib.Path) -> None: """GC on a repo with no reflog files must not crash and reports 0.""" root = _make_repo(tmp_path) r = runner.invoke(cli, ["gc", "--json", "--grace-period", "0"], env=_env(root)) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["reflog_expired"] == 0 def test_reflog_expire_days_config_respected(self, tmp_path: pathlib.Path) -> None: """When reflog.expire-days is 15, entries older than 15 days expire.""" root = _make_repo(tmp_path) _write_entries(_head_log(root), [ (_old_ts(20), "twenty-days-old"), # expires at TTL=15 (_recent_ts(5), "five-days-old"), # kept at TTL=15 ]) # Set TTL to 15 days. runner.invoke(cli, ["config", "set", "reflog.expire-days", "15"], env=_env(root)) r = runner.invoke(cli, ["gc", "--json", "--grace-period", "0"], env=_env(root)) data = json.loads(r.output) assert data["reflog_expired"] == 1 # --------------------------------------------------------------------------- # RL_19 — muse gc --dry-run --json reports count but writes nothing # --------------------------------------------------------------------------- class TestGcReflogDryRun: """RL_19: dry-run reports reflog_expired count but leaves log files untouched.""" def test_dry_run_reports_reflog_expired(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_entries(_head_log(root), [ (_old_ts(100), "old"), (_recent_ts(5), "recent"), ]) r = runner.invoke( cli, ["gc", "--dry-run", "--json", "--grace-period", "0"], env=_env(root), ) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["reflog_expired"] == 1 assert data["dry_run"] is True def test_dry_run_does_not_delete_log_entries(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) log = _head_log(root) _write_entries(log, [(_old_ts(100), "old")]) original_content = log.read_text() runner.invoke( cli, ["gc", "--dry-run", "--json", "--grace-period", "0"], env=_env(root), ) assert log.exists(), "dry-run must not delete the log file" assert log.read_text() == original_content, "dry-run must not modify log content" def test_dry_run_does_not_remove_log_file(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) log = _head_log(root) _write_entries(log, [(_old_ts(200), "very-old")]) runner.invoke( cli, ["gc", "--dry-run", "--json", "--grace-period", "0"], env=_env(root), ) assert log.exists(), "dry-run must not remove the log file" def test_dry_run_branch_log_untouched(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) blog = _branch_log(root, "dev") _write_entries(blog, [(_old_ts(100), "dev-old")]) original = blog.read_text() runner.invoke( cli, ["gc", "--dry-run", "--grace-period", "0"], env=_env(root), ) assert blog.read_text() == original # --------------------------------------------------------------------------- # RL_20 — Integration: entries older than 90 days gone after muse gc # --------------------------------------------------------------------------- class TestGcReflogIntegration: """RL_20: old entries are gone from disk after gc; recent ones survive.""" def test_old_head_entries_removed_after_gc(self, tmp_path: pathlib.Path) -> None: from muse.core.reflog import read_reflog root = _make_repo(tmp_path) _write_entries(_head_log(root), [ (_old_ts(100), "old-entry"), (_recent_ts(10), "recent-entry"), ]) r = runner.invoke(cli, ["gc", "--grace-period", "0"], env=_env(root)) assert r.exit_code == 0, r.output remaining = read_reflog(root, branch=None, limit=100) ops = [e.operation for e in remaining] assert "old-entry" not in ops assert "recent-entry" in ops def test_all_old_entries_removed_file_deleted(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) log = _head_log(root) _write_entries(log, [(_old_ts(200), "very-old")]) runner.invoke(cli, ["gc", "--grace-period", "0"], env=_env(root)) assert not log.exists(), "log file must be deleted when all entries expire" def test_branch_old_entries_removed_after_gc(self, tmp_path: pathlib.Path) -> None: from muse.core.reflog import read_reflog root = _make_repo(tmp_path) _write_entries(_branch_log(root, "main"), [ (_old_ts(120), "main-old"), (_recent_ts(5), "main-recent"), ]) runner.invoke(cli, ["gc", "--grace-period", "0"], env=_env(root)) remaining = read_reflog(root, branch="main", limit=100) ops = [e.operation for e in remaining] assert "main-old" not in ops assert "main-recent" in ops def test_exact_counts_from_100_entry_log(self, tmp_path: pathlib.Path) -> None: from muse.core.reflog import read_reflog root = _make_repo(tmp_path) old_entries = [(_old_ts(100), f"old-{i}") for i in range(60)] new_entries = [(_recent_ts(10), f"recent-{i}") for i in range(40)] _write_entries(_head_log(root), old_entries + new_entries) r = runner.invoke(cli, ["gc", "--json", "--grace-period", "0"], env=_env(root)) data = json.loads(r.output) assert data["reflog_expired"] == 60 remaining = read_reflog(root, branch=None, limit=200) assert len(remaining) == 40 def test_gc_output_without_json_flag_mentions_reflog(self, tmp_path: pathlib.Path) -> None: """Human-readable gc output should mention reflog entries when any expired.""" root = _make_repo(tmp_path) _write_entries(_head_log(root), [(_old_ts(100), "old")]) r = runner.invoke(cli, ["gc", "--grace-period", "0"], env=_env(root)) assert r.exit_code == 0, r.output assert "reflog" in r.output.lower(), ( f"Expected 'reflog' in human output when entries expired; got: {r.output!r}" )