test_reflog_compact.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """muse#54 item 3 -- `muse reflog compact [--branch <name>] [--dry-run] [--json]`. |
| 2 | |
| 3 | Merges consecutive same-category reflog entries into a single summary |
| 4 | entry, similar in spirit to `muse rebase --squash` at the commit level. |
| 5 | """ |
| 6 | from __future__ import annotations |
| 7 | |
| 8 | import json |
| 9 | import pathlib |
| 10 | |
| 11 | from tests.cli_test_helper import CliRunner |
| 12 | from muse.core.types import fake_id, Manifest, NULL_COMMIT_ID |
| 13 | from muse.core.paths import muse_dir, logs_dir |
| 14 | from muse.core.reflog import read_reflog, compact_reflog, categorize_operation |
| 15 | |
| 16 | cli = None |
| 17 | runner = CliRunner() |
| 18 | |
| 19 | |
| 20 | def _env(root: pathlib.Path) -> Manifest: |
| 21 | return {"MUSE_REPO_ROOT": str(root)} |
| 22 | |
| 23 | |
| 24 | def _run(root: pathlib.Path, *args: str): |
| 25 | return runner.invoke(cli, list(args), env=_env(root)) |
| 26 | |
| 27 | |
| 28 | def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]: |
| 29 | dot_muse = muse_dir(tmp_path) |
| 30 | dot_muse.mkdir() |
| 31 | repo_id = fake_id("repo") |
| 32 | (dot_muse / "repo.json").write_text(json.dumps({ |
| 33 | "repo_id": repo_id, "domain": "code", "default_branch": "main", |
| 34 | "created_at": "2025-01-01T00:00:00+00:00", |
| 35 | }), encoding="utf-8") |
| 36 | (dot_muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8") |
| 37 | (dot_muse / "refs" / "heads").mkdir(parents=True) |
| 38 | (dot_muse / "snapshots").mkdir() |
| 39 | (dot_muse / "commits").mkdir() |
| 40 | (dot_muse / "objects").mkdir() |
| 41 | return tmp_path, repo_id |
| 42 | |
| 43 | |
| 44 | def _commit_file(root: pathlib.Path, content: str, message: str) -> str: |
| 45 | (root / "f.txt").write_text(content) |
| 46 | add = _run(root, "code", "add", "f.txt") |
| 47 | assert add.exit_code == 0, add.output |
| 48 | commit = _run(root, "commit", "-m", message, "--json") |
| 49 | assert commit.exit_code == 0, commit.output |
| 50 | return json.loads(commit.output)["commit_id"] |
| 51 | |
| 52 | |
| 53 | def _write_raw_entry(path: pathlib.Path, old_id: str, new_id: str, ts: int, author: str, operation: str) -> None: |
| 54 | path.parent.mkdir(parents=True, exist_ok=True) |
| 55 | with path.open("a", encoding="utf-8") as f: |
| 56 | f.write(f"{old_id} {new_id} {author} {ts} +0000\t{operation}\n") |
| 57 | |
| 58 | |
| 59 | def _fake_commit_id(n: int) -> str: |
| 60 | return f"{n:064x}" |
| 61 | |
| 62 | |
| 63 | class TestCategorizeOperation: |
| 64 | def test_simple_prefix(self) -> None: |
| 65 | assert categorize_operation("commit: add verse") == "commit" |
| 66 | |
| 67 | def test_prefix_with_parenthetical(self) -> None: |
| 68 | assert categorize_operation("reset (hard): moving to sha256:abc") == "reset" |
| 69 | |
| 70 | |
| 71 | class TestCompactReflogCore: |
| 72 | def test_merges_consecutive_same_category_run(self, tmp_path: pathlib.Path) -> None: |
| 73 | root, repo_id = _init_repo(tmp_path) |
| 74 | head_log = logs_dir(root) / "HEAD" |
| 75 | c0, c1, c2, c3 = (_fake_commit_id(n) for n in range(4)) |
| 76 | _write_raw_entry(head_log, NULL_COMMIT_ID, c1, 1000, "gabriel", "commit: c1") |
| 77 | _write_raw_entry(head_log, c1, c2, 1001, "gabriel", "commit: c2") |
| 78 | _write_raw_entry(head_log, c2, c3, 1002, "gabriel", "commit: c3") |
| 79 | |
| 80 | removed, remaining = compact_reflog(root, branch=None) |
| 81 | assert removed == 2 |
| 82 | assert remaining == 1 |
| 83 | |
| 84 | entries = read_reflog(root, branch=None, limit=100) |
| 85 | assert len(entries) == 1 |
| 86 | assert entries[0].old_id == NULL_COMMIT_ID |
| 87 | assert entries[0].new_id == c3 |
| 88 | assert "commit (compacted 3 entries)" in entries[0].operation |
| 89 | |
| 90 | def test_different_categories_not_merged(self, tmp_path: pathlib.Path) -> None: |
| 91 | root, repo_id = _init_repo(tmp_path) |
| 92 | head_log = logs_dir(root) / "HEAD" |
| 93 | c1, c2, c3 = (_fake_commit_id(n) for n in range(1, 4)) |
| 94 | _write_raw_entry(head_log, NULL_COMMIT_ID, c1, 1000, "gabriel", "commit: c1") |
| 95 | _write_raw_entry(head_log, c1, c2, 1001, "gabriel", "reset (hard): moving to sha256:" + c2) |
| 96 | _write_raw_entry(head_log, c2, c3, 1002, "gabriel", "commit: c3") |
| 97 | |
| 98 | removed, remaining = compact_reflog(root, branch=None) |
| 99 | assert removed == 0 |
| 100 | assert remaining == 3 |
| 101 | |
| 102 | def test_lone_entry_never_compacted(self, tmp_path: pathlib.Path) -> None: |
| 103 | root, repo_id = _init_repo(tmp_path) |
| 104 | head_log = logs_dir(root) / "HEAD" |
| 105 | c1 = _fake_commit_id(1) |
| 106 | _write_raw_entry(head_log, NULL_COMMIT_ID, c1, 1000, "gabriel", "commit: c1") |
| 107 | |
| 108 | removed, remaining = compact_reflog(root, branch=None) |
| 109 | assert removed == 0 |
| 110 | assert remaining == 1 |
| 111 | |
| 112 | def test_different_authors_become_various(self, tmp_path: pathlib.Path) -> None: |
| 113 | root, repo_id = _init_repo(tmp_path) |
| 114 | head_log = logs_dir(root) / "HEAD" |
| 115 | c1, c2 = (_fake_commit_id(n) for n in range(1, 3)) |
| 116 | _write_raw_entry(head_log, NULL_COMMIT_ID, c1, 1000, "alice", "commit: c1") |
| 117 | _write_raw_entry(head_log, c1, c2, 1001, "bob", "commit: c2") |
| 118 | |
| 119 | compact_reflog(root, branch=None) |
| 120 | entries = read_reflog(root, branch=None, limit=100) |
| 121 | assert entries[0].author == "various" |
| 122 | |
| 123 | def test_dry_run_does_not_write(self, tmp_path: pathlib.Path) -> None: |
| 124 | root, repo_id = _init_repo(tmp_path) |
| 125 | head_log = logs_dir(root) / "HEAD" |
| 126 | c1, c2 = (_fake_commit_id(n) for n in range(1, 3)) |
| 127 | _write_raw_entry(head_log, NULL_COMMIT_ID, c1, 1000, "gabriel", "commit: c1") |
| 128 | _write_raw_entry(head_log, c1, c2, 1001, "gabriel", "commit: c2") |
| 129 | |
| 130 | removed, remaining = compact_reflog(root, branch=None, dry_run=True) |
| 131 | assert removed == 1 |
| 132 | assert remaining == 1 |
| 133 | |
| 134 | entries = read_reflog(root, branch=None, limit=100) |
| 135 | assert len(entries) == 2 # unchanged on disk |
| 136 | |
| 137 | def test_second_run_also_merged(self, tmp_path: pathlib.Path) -> None: |
| 138 | """Two separate qualifying runs, split by a differing category, both compact.""" |
| 139 | root, repo_id = _init_repo(tmp_path) |
| 140 | head_log = logs_dir(root) / "HEAD" |
| 141 | c1, c2, c3, c4, c5 = (_fake_commit_id(n) for n in range(1, 6)) |
| 142 | _write_raw_entry(head_log, NULL_COMMIT_ID, c1, 1000, "gabriel", "commit: c1") |
| 143 | _write_raw_entry(head_log, c1, c2, 1001, "gabriel", "commit: c2") |
| 144 | _write_raw_entry(head_log, c2, c3, 1002, "gabriel", "reset (hard): to c3") |
| 145 | _write_raw_entry(head_log, c3, c4, 1003, "gabriel", "commit: c4") |
| 146 | _write_raw_entry(head_log, c4, c5, 1004, "gabriel", "commit: c5") |
| 147 | |
| 148 | removed, remaining = compact_reflog(root, branch=None) |
| 149 | assert removed == 2 |
| 150 | assert remaining == 3 |
| 151 | |
| 152 | entries = read_reflog(root, branch=None, limit=100) |
| 153 | # newest-first: [commit(c4->c5 compacted), reset, commit(c1->c2 compacted)] |
| 154 | assert entries[0].new_id == c5 |
| 155 | assert entries[1].new_id == c3 |
| 156 | assert entries[2].new_id == c2 |
| 157 | |
| 158 | |
| 159 | class TestReflogCompactCli: |
| 160 | def test_cli_compact_json(self, tmp_path: pathlib.Path) -> None: |
| 161 | root, repo_id = _init_repo(tmp_path) |
| 162 | _commit_file(root, "v1\n", "c1") |
| 163 | _commit_file(root, "v2\n", "c2") |
| 164 | _commit_file(root, "v3\n", "c3") |
| 165 | |
| 166 | result = _run(root, "reflog", "compact", "--json") |
| 167 | assert result.exit_code == 0, result.output |
| 168 | data = json.loads(result.output) |
| 169 | assert data["compacted"] == 2 |
| 170 | assert data["remaining"] == 1 |
| 171 | assert data["dry_run"] is False |
| 172 | assert data["ref"] == "HEAD" |
| 173 | |
| 174 | def test_cli_compact_dry_run_json(self, tmp_path: pathlib.Path) -> None: |
| 175 | root, repo_id = _init_repo(tmp_path) |
| 176 | _commit_file(root, "v1\n", "c1") |
| 177 | _commit_file(root, "v2\n", "c2") |
| 178 | |
| 179 | result = _run(root, "reflog", "compact", "--dry-run", "--json") |
| 180 | assert result.exit_code == 0, result.output |
| 181 | data = json.loads(result.output) |
| 182 | assert data["compacted"] == 1 |
| 183 | assert data["dry_run"] is True |
| 184 | |
| 185 | # Unchanged after dry-run. |
| 186 | follow_up = _run(root, "reflog", "--json") |
| 187 | assert json.loads(follow_up.output)["total"] == 2 |
| 188 | |
| 189 | def test_cli_invalid_branch_errors(self, tmp_path: pathlib.Path) -> None: |
| 190 | root, repo_id = _init_repo(tmp_path) |
| 191 | result = _run(root, "reflog", "compact", "--branch", "../escape", "--json") |
| 192 | assert result.exit_code == 1 |