"""Phase 2 — RL_06–RL_12: ``muse reflog expire`` subcommand. TDD: tests are written before the implementation. Each class maps to one or more deliverables from issue #47, Phase 2. Coverage tiers -------------- RL_06 expire --expire-days N removes old entries; JSON reports expired count RL_07 expire --all applies to all branch logs; refs_processed listed RL_08 expire --dry-run reports but writes nothing RL_09 muse config reflog.expire-days N sets default TTL RL_10 expire is atomic (write-tmp + os.replace); original survives mock-kill RL_11 100-entry log, 50 old → exactly 50 removed, 50 kept RL_12 all entries expire → log file removed; muse reflog returns entries:[] """ from __future__ import annotations import datetime import json import os import pathlib import time from unittest.mock import patch, call import pytest from muse.core.paths import muse_dir, logs_dir from muse.core.types import fake_id, NULL_COMMIT_ID from tests.cli_test_helper import CliRunner runner = CliRunner() cli = None # argparse migration stub _DAY = 86400 # seconds # --------------------------------------------------------------------------- # 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: """Create a minimal .muse/ structure, no commits.""" from muse._version import __version__ dot = muse_dir(tmp_path) for sub in ( "refs/heads", "objects", "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 _write_log_entries( log_path: pathlib.Path, entries: list[tuple[int, str]], # list of (unix_ts, operation) ) -> None: """Write raw reflog entries at given timestamps (bypasses append_reflog).""" log_path.parent.mkdir(parents=True, exist_ok=True) old = fake_id("old") new = fake_id("new") lines = [] for ts, op in entries: lines.append(f"{old} {new} user {ts} +0000\t{op}\n") log_path.write_text("".join(lines), encoding="utf-8") def _head_log_path(root: pathlib.Path) -> pathlib.Path: return logs_dir(root) / "HEAD" def _branch_log_path(root: pathlib.Path, branch: str) -> pathlib.Path: return logs_dir(root) / "refs" / "heads" / branch def _now_ts() -> int: return int(time.time()) def _old_ts(days: int) -> int: """Unix timestamp *days* days ago.""" return _now_ts() - days * _DAY # --------------------------------------------------------------------------- # RL_06 — expire removes old entries; JSON reports expired/kept # --------------------------------------------------------------------------- class TestExpireBasic: """RL_06: expire removes entries older than the threshold.""" def test_removes_old_entries_from_head_log(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) log = _head_log_path(root) # 3 old (100 days) + 2 recent (10 days) entries = ( [(_old_ts(100), f"old-{i}") for i in range(3)] + [(_old_ts(10), f"recent-{i}") for i in range(2)] ) _write_log_entries(log, entries) r = runner.invoke(cli, ["reflog", "expire", "--expire-days", "90", "--json"], env=_env(root)) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["expired"] == 3 assert data["kept"] == 2 assert data["dry_run"] is False def test_remaining_entries_are_the_recent_ones(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) log = _head_log_path(root) _write_log_entries(log, [ (_old_ts(200), "very-old"), (_old_ts(10), "recent"), ]) runner.invoke(cli, ["reflog", "expire", "--expire-days", "90", "--json"], env=_env(root)) from muse.core.reflog import read_reflog remaining = read_reflog(root, branch=None, limit=100) assert len(remaining) == 1 assert remaining[0].operation == "recent" def test_json_schema_has_all_required_fields(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_log_entries(_head_log_path(root), [(_old_ts(5), "keep")]) r = runner.invoke(cli, ["reflog", "expire", "--expire-days", "90", "--json"], env=_env(root)) data = json.loads(r.output) for key in ("expired", "kept", "dry_run", "refs_processed", "exit_code", "duration_ms"): assert key in data, f"missing key {key!r}" def test_no_entries_expire_when_all_recent(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_log_entries(_head_log_path(root), [ (_old_ts(1), "a"), (_old_ts(2), "b"), ]) r = runner.invoke(cli, ["reflog", "expire", "--expire-days", "90", "--json"], env=_env(root)) data = json.loads(r.output) assert data["expired"] == 0 assert data["kept"] == 2 def test_no_log_file_is_graceful(self, tmp_path: pathlib.Path) -> None: """Repo with no reflog entries must not crash.""" root = _make_repo(tmp_path) r = runner.invoke(cli, ["reflog", "expire", "--expire-days", "90", "--json"], env=_env(root)) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["expired"] == 0 assert data["kept"] == 0 # --------------------------------------------------------------------------- # RL_07 — expire --all covers every branch log # --------------------------------------------------------------------------- class TestExpireAll: """RL_07: --all processes HEAD + all branch logs.""" def test_all_flag_processes_all_branch_logs(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) # Head log — 1 old entry _write_log_entries(_head_log_path(root), [(_old_ts(100), "head-old")]) # Two branch logs — each with 1 old entry for branch in ("main", "dev"): _write_log_entries(_branch_log_path(root, branch), [(_old_ts(100), f"{branch}-old")]) r = runner.invoke(cli, ["reflog", "expire", "--all", "--expire-days", "90", "--json"], env=_env(root)) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["expired"] >= 2 # at least the two branch logs # refs_processed must include the branch refs refs = data["refs_processed"] assert any("main" in ref for ref in refs) assert any("dev" in ref for ref in refs) def test_refs_processed_lists_every_touched_ref(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) for branch in ("main", "feature"): _write_log_entries(_branch_log_path(root, branch), [(_old_ts(5), "keep")]) r = runner.invoke(cli, ["reflog", "expire", "--all", "--expire-days", "90", "--json"], env=_env(root)) data = json.loads(r.output) refs = data["refs_processed"] # Both branches must appear even though nothing expired from them assert any("main" in ref for ref in refs) assert any("feature" in ref for ref in refs) def test_all_with_specific_branch_flag_is_rejected(self, tmp_path: pathlib.Path) -> None: """--all and --branch together is ambiguous; must fail with USER_ERROR.""" root = _make_repo(tmp_path) r = runner.invoke(cli, ["reflog", "expire", "--all", "--branch", "main", "--json"], env=_env(root)) assert r.exit_code != 0 # --------------------------------------------------------------------------- # RL_08 — --dry-run reports without writing # --------------------------------------------------------------------------- class TestExpireDryRun: """RL_08: dry-run reports what would be removed but writes nothing.""" def test_dry_run_reports_expired_count(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_log_entries(_head_log_path(root), [ (_old_ts(100), "old"), (_old_ts(5), "new"), ]) r = runner.invoke(cli, ["reflog", "expire", "--expire-days", "90", "--dry-run", "--json"], env=_env(root)) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["expired"] == 1 assert data["kept"] == 1 assert data["dry_run"] is True def test_dry_run_does_not_modify_log(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) log = _head_log_path(root) _write_log_entries(log, [(_old_ts(100), "old")]) original_content = log.read_text(encoding="utf-8") runner.invoke(cli, ["reflog", "expire", "--expire-days", "90", "--dry-run"], env=_env(root)) assert log.read_text(encoding="utf-8") == original_content, ( "dry-run must not modify the log file" ) def test_dry_run_all_does_not_modify_any_log(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) for branch in ("main", "dev"): _write_log_entries(_branch_log_path(root, branch), [(_old_ts(100), f"{branch}-old")]) originals = { b: _branch_log_path(root, b).read_text() for b in ("main", "dev") } runner.invoke(cli, ["reflog", "expire", "--all", "--expire-days", "90", "--dry-run"], env=_env(root)) for b, orig in originals.items(): assert _branch_log_path(root, b).read_text() == orig, ( f"dry-run modified branch log for {b!r}" ) # --------------------------------------------------------------------------- # RL_09 — muse config reflog.expire-days sets default TTL # --------------------------------------------------------------------------- class TestExpireConfigTTL: """RL_09: reflog.expire-days config key is read when --expire-days is omitted.""" def test_config_expire_days_is_used_as_default(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_log_entries(_head_log_path(root), [ (_old_ts(20), "old"), # 20 days old — expires at 15-day TTL (_old_ts(5), "recent"), # 5 days old — kept at 15-day TTL ]) # Set config to 15 days. set_r = runner.invoke(cli, ["config", "set", "reflog.expire-days", "15"], env=_env(root)) assert set_r.exit_code == 0, set_r.output # Run expire without --expire-days; should use 15 from config. r = runner.invoke(cli, ["reflog", "expire", "--json"], env=_env(root)) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["expired"] == 1 assert data["kept"] == 1 def test_explicit_flag_overrides_config(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_log_entries(_head_log_path(root), [ (_old_ts(50), "fifty-days-old"), ]) # Config says 90 days — 50 days old should be kept. set_r = runner.invoke(cli, ["config", "set", "reflog.expire-days", "90"], env=_env(root)) assert set_r.exit_code == 0, set_r.output # But explicit --expire-days 30 means 50 days is old. r = runner.invoke(cli, ["reflog", "expire", "--expire-days", "30", "--json"], env=_env(root)) data = json.loads(r.output) assert data["expired"] == 1 def test_default_ttl_is_90_when_config_not_set(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) _write_log_entries(_head_log_path(root), [ (_old_ts(100), "very-old"), # 100 > 90 — should expire (_old_ts(80), "recent"), # 80 < 90 — kept ]) r = runner.invoke(cli, ["reflog", "expire", "--json"], env=_env(root)) data = json.loads(r.output) assert data["expired"] == 1 assert data["kept"] == 1 def test_muse_config_get_returns_set_value(self, tmp_path: pathlib.Path) -> None: """muse config get reflog.expire-days returns what was set.""" root = _make_repo(tmp_path) set_r = runner.invoke(cli, ["config", "set", "reflog.expire-days", "45"], env=_env(root)) assert set_r.exit_code == 0, set_r.output get_r = runner.invoke(cli, ["config", "get", "reflog.expire-days"], env=_env(root)) assert get_r.exit_code == 0, get_r.output assert "45" in get_r.output # --------------------------------------------------------------------------- # RL_10 — Atomic write (write-tmp + os.replace) # --------------------------------------------------------------------------- class TestExpireAtomic: """RL_10: expire uses write-to-tmp + os.replace; original survives mock-kill.""" def test_original_log_intact_when_replace_fails(self, tmp_path: pathlib.Path) -> None: """If os.replace raises mid-write, the original log is unchanged. The atomic-write path (write-tmp + os.replace) is only taken when there are kept entries. Mix old and recent so expire filters to kept_lines > 0 and the replace call is actually exercised. """ root = _make_repo(tmp_path) log = _head_log_path(root) # One old entry (will be expired) + one recent (will be kept). # This forces the atomic write path rather than plain unlink. original_content = ( f"{fake_id('old')} {fake_id('new')} user {_old_ts(100)} +0000\told\n" f"{fake_id('old')} {fake_id('new')} user {_old_ts(5)} +0000\trecent\n" ) log.write_text(original_content, encoding="utf-8") with patch("muse.core.reflog.os.replace", side_effect=OSError("disk full")): runner.invoke(cli, ["reflog", "expire", "--expire-days", "90", "--json"], env=_env(root)) # Original must be intact — os.replace failed before the swap happened. assert log.read_text(encoding="utf-8") == original_content, ( "original log must be unchanged when os.replace raises" ) def test_no_partial_tmp_file_survives_on_success(self, tmp_path: pathlib.Path) -> None: """After a successful expire the tmp file must be gone (os.replace cleaned it).""" root = _make_repo(tmp_path) _write_log_entries(_head_log_path(root), [(_old_ts(100), "old")]) runner.invoke(cli, ["reflog", "expire", "--expire-days", "90"], env=_env(root)) # No .lock / .tmp files should remain anywhere in .muse/logs/ log_dir = logs_dir(root) tmp_files = list(log_dir.rglob("*.lock")) + list(log_dir.rglob("*.tmp")) assert not tmp_files, f"Stale tmp files found: {tmp_files}" # --------------------------------------------------------------------------- # RL_11 — Integration: 100-entry log, exactly 50 removed, 50 kept # --------------------------------------------------------------------------- class TestExpireIntegration: """RL_11: large log; expire removes exactly the expected entries.""" def test_exactly_50_removed_from_100_entry_log(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) log = _head_log_path(root) # 50 entries at 100 days old, 50 entries at 10 days old. old_entries = [(_old_ts(100), f"old-{i}") for i in range(50)] new_entries = [(_old_ts(10), f"recent-{i}") for i in range(50)] _write_log_entries(log, old_entries + new_entries) r = runner.invoke(cli, ["reflog", "expire", "--expire-days", "90", "--json"], env=_env(root)) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["expired"] == 50 assert data["kept"] == 50 def test_remaining_entries_are_the_correct_ones(self, tmp_path: pathlib.Path) -> None: """After expire, re-reading the log returns exactly the 50 recent entries.""" from muse.core.reflog import read_reflog root = _make_repo(tmp_path) log = _head_log_path(root) old_entries = [(_old_ts(100), f"old-{i}") for i in range(50)] new_entries = [(_old_ts(10), f"recent-{i}") for i in range(50)] _write_log_entries(log, old_entries + new_entries) runner.invoke(cli, ["reflog", "expire", "--expire-days", "90"], env=_env(root)) remaining = read_reflog(root, branch=None, limit=200) assert len(remaining) == 50 assert all("recent" in e.operation for e in remaining) # --------------------------------------------------------------------------- # RL_12 — All entries expire → log file removed; muse reflog returns [] # --------------------------------------------------------------------------- class TestExpireAllEntriesGone: """RL_12: when all entries expire the log file is deleted (not left as zero-byte).""" def test_log_file_removed_when_all_entries_expire(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) log = _head_log_path(root) _write_log_entries(log, [(_old_ts(200), "very-old")]) runner.invoke(cli, ["reflog", "expire", "--expire-days", "90"], env=_env(root)) assert not log.exists(), "log file must be removed when all entries expire" def test_muse_reflog_returns_empty_entries_after_full_expire( self, tmp_path: pathlib.Path ) -> None: """muse reflog --json must return entries:[] cleanly when log is gone.""" root = _make_repo(tmp_path) _write_log_entries(_head_log_path(root), [(_old_ts(200), "old")]) runner.invoke(cli, ["reflog", "expire", "--expire-days", "90"], env=_env(root)) r = runner.invoke(cli, ["reflog", "--json"], env=_env(root)) assert r.exit_code == 0, r.output data = json.loads(r.output) assert data["entries"] == [] assert data["total"] == 0 def test_branch_log_removed_when_all_entries_expire(self, tmp_path: pathlib.Path) -> None: root = _make_repo(tmp_path) branch_log = _branch_log_path(root, "main") _write_log_entries(branch_log, [(_old_ts(200), "old")]) runner.invoke(cli, ["reflog", "expire", "--branch", "main", "--expire-days", "90"], env=_env(root)) assert not branch_log.exists(), "branch log must be removed when all entries expire" def test_zero_byte_file_is_never_left_behind(self, tmp_path: pathlib.Path) -> None: """Ensure expire doesn't write an empty file instead of deleting it.""" root = _make_repo(tmp_path) log = _head_log_path(root) _write_log_entries(log, [(_old_ts(200), "old")]) runner.invoke(cli, ["reflog", "expire", "--expire-days", "90"], env=_env(root)) # Either the file is gone or it has content — never zero bytes. if log.exists(): assert log.stat().st_size > 0, "log file must not be left as zero bytes"