test_cmd_reflog_delete.py
python
sha256:24d9cb81ac1dd071e5d98862c740ad101180588361b4f96a790af79a99aed901
docs: architectural plan for unified-store snapshot-content…
Sonnet 5
9 days ago
| 1 | """Phase 3 — RL_13–RL_17: ``muse reflog delete`` subcommand. |
| 2 | |
| 3 | TDD: tests are written before the implementation. Each class maps to one or |
| 4 | more deliverables from issue #47, Phase 3. |
| 5 | |
| 6 | Coverage tiers |
| 7 | -------------- |
| 8 | RL_13 delete @{N} removes the entry at index N; deleted:1, remaining:N-1 |
| 9 | RL_14 delete --all removes all entries; log file deleted; deleted == prior count |
| 10 | RL_15 out-of-bounds index exits USER_ERROR with a message naming valid range |
| 11 | RL_16 delete is atomic (write-tmp + os.replace); original survives mock-kill |
| 12 | RL_17 integration: 3-entry log → delete @{1} → 2 entries remain (newest+oldest) |
| 13 | """ |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | import json |
| 17 | import pathlib |
| 18 | from unittest.mock import patch |
| 19 | |
| 20 | import pytest |
| 21 | |
| 22 | from muse.core.paths import muse_dir, logs_dir |
| 23 | from muse.core.types import fake_id |
| 24 | from tests.cli_test_helper import CliRunner |
| 25 | |
| 26 | runner = CliRunner() |
| 27 | cli = None # argparse migration stub |
| 28 | |
| 29 | _DAY = 86400 |
| 30 | |
| 31 | # --------------------------------------------------------------------------- |
| 32 | # Helpers |
| 33 | # --------------------------------------------------------------------------- |
| 34 | |
| 35 | def _env(root: pathlib.Path) -> dict[str, str]: |
| 36 | return {"MUSE_REPO_ROOT": str(root)} |
| 37 | |
| 38 | |
| 39 | def _make_repo(tmp_path: pathlib.Path, branch: str = "main") -> pathlib.Path: |
| 40 | """Create a minimal .muse/ structure with no commits.""" |
| 41 | from muse._version import __version__ |
| 42 | |
| 43 | dot = muse_dir(tmp_path) |
| 44 | for sub in ( |
| 45 | "refs/heads", "objects", "commits", "snapshots", |
| 46 | "logs/refs/heads", "logs", |
| 47 | ): |
| 48 | (dot / sub).mkdir(parents=True, exist_ok=True) |
| 49 | (dot / "repo.json").write_text( |
| 50 | json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"}) |
| 51 | ) |
| 52 | (dot / "HEAD").write_text(f"ref: refs/heads/{branch}\n") |
| 53 | (dot / "config.toml").write_text("") |
| 54 | return tmp_path |
| 55 | |
| 56 | |
| 57 | def _head_log(root: pathlib.Path) -> pathlib.Path: |
| 58 | return logs_dir(root) / "HEAD" |
| 59 | |
| 60 | |
| 61 | def _branch_log(root: pathlib.Path, branch: str) -> pathlib.Path: |
| 62 | return logs_dir(root) / "refs" / "heads" / branch |
| 63 | |
| 64 | |
| 65 | def _write_entries(log_path: pathlib.Path, operations: list[str]) -> None: |
| 66 | """Write one raw reflog line per operation, in chronological order (oldest first). |
| 67 | |
| 68 | Timestamps are spaced 1 second apart starting at unix epoch + 1000 so |
| 69 | ordering is deterministic and all entries are old enough to avoid |
| 70 | accidental expiry in other tests. |
| 71 | """ |
| 72 | import time as _time |
| 73 | log_path.parent.mkdir(parents=True, exist_ok=True) |
| 74 | base_ts = int(_time.time()) - len(operations) * 2 # all in the recent past |
| 75 | lines = [] |
| 76 | for i, op in enumerate(operations): |
| 77 | old = fake_id(f"old{i}") |
| 78 | new = fake_id(f"new{i}") |
| 79 | ts = base_ts + i |
| 80 | lines.append(f"{old} {new} user {ts} +0000\t{op}\n") |
| 81 | log_path.write_text("".join(lines), encoding="utf-8") |
| 82 | |
| 83 | |
| 84 | def _read_operations(root: pathlib.Path, branch: str | None = None) -> list[str]: |
| 85 | """Return operation strings newest-first (mirrors read_reflog order).""" |
| 86 | from muse.core.reflog import read_reflog |
| 87 | return [e.operation for e in read_reflog(root, branch=branch, limit=1000)] |
| 88 | |
| 89 | |
| 90 | # --------------------------------------------------------------------------- |
| 91 | # RL_13 — delete @{N} removes the entry at index N |
| 92 | # --------------------------------------------------------------------------- |
| 93 | |
| 94 | class TestDeleteByIndex: |
| 95 | """RL_13: delete @{0} removes the newest entry.""" |
| 96 | |
| 97 | def test_delete_newest_entry(self, tmp_path: pathlib.Path) -> None: |
| 98 | root = _make_repo(tmp_path) |
| 99 | _write_entries(_head_log(root), ["oldest", "middle", "newest"]) |
| 100 | |
| 101 | r = runner.invoke(cli, ["reflog", "delete", "@{0}", "--json"], env=_env(root)) |
| 102 | |
| 103 | assert r.exit_code == 0, r.output |
| 104 | data = json.loads(r.output) |
| 105 | assert data["deleted"] == 1 |
| 106 | assert data["remaining"] == 2 |
| 107 | |
| 108 | def test_newest_entry_is_gone_after_delete(self, tmp_path: pathlib.Path) -> None: |
| 109 | root = _make_repo(tmp_path) |
| 110 | _write_entries(_head_log(root), ["oldest", "middle", "newest"]) |
| 111 | |
| 112 | runner.invoke(cli, ["reflog", "delete", "@{0}"], env=_env(root)) |
| 113 | |
| 114 | ops = _read_operations(root) |
| 115 | assert "newest" not in ops |
| 116 | assert "middle" in ops |
| 117 | assert "oldest" in ops |
| 118 | |
| 119 | def test_json_schema_has_required_fields(self, tmp_path: pathlib.Path) -> None: |
| 120 | root = _make_repo(tmp_path) |
| 121 | _write_entries(_head_log(root), ["a", "b"]) |
| 122 | |
| 123 | r = runner.invoke(cli, ["reflog", "delete", "@{0}", "--json"], env=_env(root)) |
| 124 | |
| 125 | data = json.loads(r.output) |
| 126 | for key in ("deleted", "remaining", "ref", "exit_code", "duration_ms"): |
| 127 | assert key in data, f"missing key {key!r}" |
| 128 | |
| 129 | def test_ref_field_is_head_by_default(self, tmp_path: pathlib.Path) -> None: |
| 130 | root = _make_repo(tmp_path) |
| 131 | _write_entries(_head_log(root), ["a"]) |
| 132 | |
| 133 | r = runner.invoke(cli, ["reflog", "delete", "@{0}", "--json"], env=_env(root)) |
| 134 | |
| 135 | data = json.loads(r.output) |
| 136 | assert data["ref"] == "HEAD" |
| 137 | |
| 138 | def test_delete_last_entry_removes_log_file(self, tmp_path: pathlib.Path) -> None: |
| 139 | """Deleting the only entry must remove the log file.""" |
| 140 | root = _make_repo(tmp_path) |
| 141 | log = _head_log(root) |
| 142 | _write_entries(log, ["only"]) |
| 143 | |
| 144 | r = runner.invoke(cli, ["reflog", "delete", "@{0}", "--json"], env=_env(root)) |
| 145 | |
| 146 | assert r.exit_code == 0, r.output |
| 147 | assert not log.exists(), "log file must be removed when all entries are deleted" |
| 148 | |
| 149 | def test_delete_by_branch_flag(self, tmp_path: pathlib.Path) -> None: |
| 150 | """--branch targets the branch-specific log, not HEAD.""" |
| 151 | root = _make_repo(tmp_path) |
| 152 | _write_entries(_branch_log(root, "main"), ["branch-a", "branch-b"]) |
| 153 | _write_entries(_head_log(root), ["head-a", "head-b"]) |
| 154 | |
| 155 | r = runner.invoke( |
| 156 | cli, |
| 157 | ["reflog", "delete", "@{0}", "--branch", "main", "--json"], |
| 158 | env=_env(root), |
| 159 | ) |
| 160 | |
| 161 | assert r.exit_code == 0, r.output |
| 162 | data = json.loads(r.output) |
| 163 | assert data["deleted"] == 1 |
| 164 | assert data["ref"] == "refs/heads/main" |
| 165 | # HEAD log must be untouched. |
| 166 | assert len(_read_operations(root)) == 2 |
| 167 | |
| 168 | def test_delete_middle_entry(self, tmp_path: pathlib.Path) -> None: |
| 169 | """@{1} removes the second-newest entry.""" |
| 170 | root = _make_repo(tmp_path) |
| 171 | _write_entries(_head_log(root), ["oldest", "middle", "newest"]) |
| 172 | |
| 173 | runner.invoke(cli, ["reflog", "delete", "@{1}"], env=_env(root)) |
| 174 | |
| 175 | ops = _read_operations(root) |
| 176 | assert "middle" not in ops |
| 177 | assert "newest" in ops |
| 178 | assert "oldest" in ops |
| 179 | |
| 180 | def test_delete_oldest_entry(self, tmp_path: pathlib.Path) -> None: |
| 181 | """@{2} in a 3-entry log removes the oldest entry.""" |
| 182 | root = _make_repo(tmp_path) |
| 183 | _write_entries(_head_log(root), ["oldest", "middle", "newest"]) |
| 184 | |
| 185 | runner.invoke(cli, ["reflog", "delete", "@{2}"], env=_env(root)) |
| 186 | |
| 187 | ops = _read_operations(root) |
| 188 | assert "oldest" not in ops |
| 189 | assert "middle" in ops |
| 190 | assert "newest" in ops |
| 191 | |
| 192 | |
| 193 | # --------------------------------------------------------------------------- |
| 194 | # RL_14 — delete --all removes all entries; log file is deleted |
| 195 | # --------------------------------------------------------------------------- |
| 196 | |
| 197 | class TestDeleteAll: |
| 198 | """RL_14: delete --all wipes the entire log.""" |
| 199 | |
| 200 | def test_all_flag_removes_every_entry(self, tmp_path: pathlib.Path) -> None: |
| 201 | root = _make_repo(tmp_path) |
| 202 | log = _head_log(root) |
| 203 | _write_entries(log, ["a", "b", "c"]) |
| 204 | |
| 205 | r = runner.invoke(cli, ["reflog", "delete", "--all", "--json"], env=_env(root)) |
| 206 | |
| 207 | assert r.exit_code == 0, r.output |
| 208 | data = json.loads(r.output) |
| 209 | assert data["deleted"] == 3 |
| 210 | assert data["remaining"] == 0 |
| 211 | |
| 212 | def test_all_flag_removes_log_file(self, tmp_path: pathlib.Path) -> None: |
| 213 | root = _make_repo(tmp_path) |
| 214 | log = _head_log(root) |
| 215 | _write_entries(log, ["a", "b"]) |
| 216 | |
| 217 | runner.invoke(cli, ["reflog", "delete", "--all"], env=_env(root)) |
| 218 | |
| 219 | assert not log.exists(), "log file must be removed after --all delete" |
| 220 | |
| 221 | def test_all_on_branch_log(self, tmp_path: pathlib.Path) -> None: |
| 222 | root = _make_repo(tmp_path) |
| 223 | blog = _branch_log(root, "dev") |
| 224 | _write_entries(blog, ["x", "y"]) |
| 225 | _write_entries(_head_log(root), ["h"]) |
| 226 | |
| 227 | runner.invoke(cli, ["reflog", "delete", "--all", "--branch", "dev"], env=_env(root)) |
| 228 | |
| 229 | assert not blog.exists() |
| 230 | # HEAD log must be untouched. |
| 231 | assert _head_log(root).exists() |
| 232 | |
| 233 | def test_all_on_empty_log_is_graceful(self, tmp_path: pathlib.Path) -> None: |
| 234 | """--all with no log file must exit 0 and report deleted:0.""" |
| 235 | root = _make_repo(tmp_path) |
| 236 | |
| 237 | r = runner.invoke(cli, ["reflog", "delete", "--all", "--json"], env=_env(root)) |
| 238 | |
| 239 | assert r.exit_code == 0, r.output |
| 240 | data = json.loads(r.output) |
| 241 | assert data["deleted"] == 0 |
| 242 | assert data["remaining"] == 0 |
| 243 | |
| 244 | def test_all_and_index_together_rejected(self, tmp_path: pathlib.Path) -> None: |
| 245 | """--all and a positional @{N} index together must fail.""" |
| 246 | root = _make_repo(tmp_path) |
| 247 | r = runner.invoke(cli, ["reflog", "delete", "--all", "@{0}"], env=_env(root)) |
| 248 | assert r.exit_code != 0 |
| 249 | |
| 250 | def test_muse_reflog_returns_empty_after_all_delete(self, tmp_path: pathlib.Path) -> None: |
| 251 | root = _make_repo(tmp_path) |
| 252 | _write_entries(_head_log(root), ["a", "b"]) |
| 253 | |
| 254 | runner.invoke(cli, ["reflog", "delete", "--all"], env=_env(root)) |
| 255 | |
| 256 | r = runner.invoke(cli, ["reflog", "--json"], env=_env(root)) |
| 257 | data = json.loads(r.output) |
| 258 | assert data["entries"] == [] |
| 259 | |
| 260 | |
| 261 | # --------------------------------------------------------------------------- |
| 262 | # RL_15 — out-of-bounds index exits USER_ERROR with valid-range message |
| 263 | # --------------------------------------------------------------------------- |
| 264 | |
| 265 | class TestDeleteOutOfBounds: |
| 266 | """RL_15: index beyond valid range → USER_ERROR with range in the message.""" |
| 267 | |
| 268 | def test_out_of_bounds_exits_user_error(self, tmp_path: pathlib.Path) -> None: |
| 269 | root = _make_repo(tmp_path) |
| 270 | _write_entries(_head_log(root), ["a", "b", "c"]) # valid: 0–2 |
| 271 | |
| 272 | r = runner.invoke(cli, ["reflog", "delete", "@{5}"], env=_env(root)) |
| 273 | |
| 274 | assert r.exit_code == 1 # USER_ERROR |
| 275 | |
| 276 | def test_out_of_bounds_message_names_valid_range(self, tmp_path: pathlib.Path) -> None: |
| 277 | root = _make_repo(tmp_path) |
| 278 | _write_entries(_head_log(root), ["a", "b", "c"]) # 3 entries → valid: 0–2 |
| 279 | |
| 280 | r = runner.invoke(cli, ["reflog", "delete", "@{5}"], env=_env(root)) |
| 281 | |
| 282 | # Valid range "0" and "2" (or "0–2") must appear somewhere in output |
| 283 | combined = r.output + (r.stderr if hasattr(r, "stderr") else "") |
| 284 | assert "0" in combined |
| 285 | assert "2" in combined |
| 286 | |
| 287 | def test_out_of_bounds_does_not_modify_log(self, tmp_path: pathlib.Path) -> None: |
| 288 | root = _make_repo(tmp_path) |
| 289 | log = _head_log(root) |
| 290 | _write_entries(log, ["a", "b"]) |
| 291 | original = log.read_text() |
| 292 | |
| 293 | runner.invoke(cli, ["reflog", "delete", "@{99}"], env=_env(root)) |
| 294 | |
| 295 | assert log.read_text() == original, "log must not be modified on out-of-bounds" |
| 296 | |
| 297 | def test_no_log_file_gives_user_error(self, tmp_path: pathlib.Path) -> None: |
| 298 | """Deleting from a nonexistent log is an error.""" |
| 299 | root = _make_repo(tmp_path) |
| 300 | |
| 301 | r = runner.invoke(cli, ["reflog", "delete", "@{0}"], env=_env(root)) |
| 302 | |
| 303 | assert r.exit_code == 1 # USER_ERROR — nothing to delete |
| 304 | |
| 305 | def test_malformed_index_is_rejected(self, tmp_path: pathlib.Path) -> None: |
| 306 | """@{abc} is not a valid index — must fail.""" |
| 307 | root = _make_repo(tmp_path) |
| 308 | _write_entries(_head_log(root), ["a"]) |
| 309 | |
| 310 | r = runner.invoke(cli, ["reflog", "delete", "@{abc}"], env=_env(root)) |
| 311 | |
| 312 | assert r.exit_code == 1 |
| 313 | |
| 314 | |
| 315 | # --------------------------------------------------------------------------- |
| 316 | # RL_16 — delete is atomic (write-tmp + os.replace) |
| 317 | # --------------------------------------------------------------------------- |
| 318 | |
| 319 | class TestDeleteAtomic: |
| 320 | """RL_16: single-entry delete uses write-tmp + os.replace; original survives mock-kill.""" |
| 321 | |
| 322 | def test_original_intact_when_replace_fails(self, tmp_path: pathlib.Path) -> None: |
| 323 | """If os.replace raises, the original log is unchanged.""" |
| 324 | root = _make_repo(tmp_path) |
| 325 | log = _head_log(root) |
| 326 | _write_entries(log, ["oldest", "newest"]) # 2 entries → kept_lines not empty |
| 327 | original = log.read_text() |
| 328 | |
| 329 | with patch("muse.core.reflog.os.replace", side_effect=OSError("disk full")): |
| 330 | runner.invoke(cli, ["reflog", "delete", "@{0}"], env=_env(root)) |
| 331 | |
| 332 | assert log.read_text() == original, "original log must be unchanged when replace fails" |
| 333 | |
| 334 | def test_no_tmp_files_left_on_success(self, tmp_path: pathlib.Path) -> None: |
| 335 | """After a successful delete no .lock / .tmp files remain.""" |
| 336 | root = _make_repo(tmp_path) |
| 337 | _write_entries(_head_log(root), ["a", "b", "c"]) |
| 338 | |
| 339 | runner.invoke(cli, ["reflog", "delete", "@{0}"], env=_env(root)) |
| 340 | |
| 341 | log_dir = logs_dir(root) |
| 342 | tmp_files = list(log_dir.rglob("*.lock")) + list(log_dir.rglob("*.tmp")) |
| 343 | assert not tmp_files, f"stale tmp files after delete: {tmp_files}" |
| 344 | |
| 345 | |
| 346 | # --------------------------------------------------------------------------- |
| 347 | # RL_17 — Integration: 3-entry log → delete @{1} → correct 2 entries remain |
| 348 | # --------------------------------------------------------------------------- |
| 349 | |
| 350 | class TestDeleteIntegration: |
| 351 | """RL_17: delete @{1} from a 3-entry log removes the middle entry.""" |
| 352 | |
| 353 | def test_correct_entries_remain_after_middle_delete(self, tmp_path: pathlib.Path) -> None: |
| 354 | from muse.core.reflog import read_reflog |
| 355 | |
| 356 | root = _make_repo(tmp_path) |
| 357 | _write_entries(_head_log(root), ["oldest", "middle", "newest"]) |
| 358 | |
| 359 | r = runner.invoke(cli, ["reflog", "delete", "@{1}", "--json"], env=_env(root)) |
| 360 | |
| 361 | assert r.exit_code == 0, r.output |
| 362 | data = json.loads(r.output) |
| 363 | assert data["deleted"] == 1 |
| 364 | assert data["remaining"] == 2 |
| 365 | |
| 366 | entries = read_reflog(root, branch=None, limit=10) |
| 367 | ops = [e.operation for e in entries] |
| 368 | assert ops == ["newest", "oldest"], ( |
| 369 | f"Expected ['newest', 'oldest'] (newest-first), got: {ops}" |
| 370 | ) |
| 371 | |
| 372 | def test_entry_count_is_exact_after_delete(self, tmp_path: pathlib.Path) -> None: |
| 373 | from muse.core.reflog import read_reflog |
| 374 | |
| 375 | root = _make_repo(tmp_path) |
| 376 | _write_entries(_head_log(root), [f"entry-{i}" for i in range(10)]) |
| 377 | |
| 378 | runner.invoke(cli, ["reflog", "delete", "@{5}"], env=_env(root)) |
| 379 | |
| 380 | entries = read_reflog(root, branch=None, limit=100) |
| 381 | assert len(entries) == 9 |
| 382 | |
| 383 | def test_repeated_deletes_converge_to_empty(self, tmp_path: pathlib.Path) -> None: |
| 384 | """Deleting @{0} repeatedly empties the log without error.""" |
| 385 | root = _make_repo(tmp_path) |
| 386 | _write_entries(_head_log(root), ["a", "b", "c"]) |
| 387 | |
| 388 | for _ in range(3): |
| 389 | r = runner.invoke(cli, ["reflog", "delete", "@{0}"], env=_env(root)) |
| 390 | assert r.exit_code == 0, r.output |
| 391 | |
| 392 | assert not _head_log(root).exists() |
| 393 | |
| 394 | def test_branch_delete_does_not_touch_head_log(self, tmp_path: pathlib.Path) -> None: |
| 395 | """Deleting from a branch log must not affect the HEAD log.""" |
| 396 | root = _make_repo(tmp_path) |
| 397 | _write_entries(_branch_log(root, "main"), ["b0", "b1", "b2"]) |
| 398 | _write_entries(_head_log(root), ["h0", "h1"]) |
| 399 | |
| 400 | runner.invoke(cli, ["reflog", "delete", "@{1}", "--branch", "main"], env=_env(root)) |
| 401 | |
| 402 | head_ops = _read_operations(root) |
| 403 | assert len(head_ops) == 2, "HEAD log must not be modified by branch delete" |
File History
1 commit
sha256:24d9cb81ac1dd071e5d98862c740ad101180588361b4f96a790af79a99aed901
docs: architectural plan for unified-store snapshot-content…
Sonnet 5
9 days ago