gabriel / muse public
test_cmd_mv.py python
510 lines 17.3 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
1 """Tests for ``muse mv`` — tracked-file move with staging.
2
3 Coverage tiers:
4 - Unit: _resolve_source, _resolve_dest, _get_source_object_id helpers
5 - Integration: basic move (disk + stage), --dry-run, --force, move-into-dir,
6 directory move, staged-only-file move, --json output
7 - End-to-end: full CLI via CliRunner
8 - Security: path traversal in source/dest rejected, outside-repo paths
9 - Edge cases: source not tracked, dest already tracked, dest exists on disk
10 - Stress: 200-file repo, move half
11 """
12
13 from __future__ import annotations
14 from collections.abc import Mapping
15
16 import datetime
17 import json
18 import pathlib
19
20 import pytest
21 from tests.cli_test_helper import CliRunner
22
23 from muse.core.object_store import write_object
24 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
25 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
26 from muse.core._types import Manifest, blob_id
27 from muse.plugins.code.stage import StagedFileMap, make_entry, read_stage, write_stage
28
29 runner = CliRunner()
30
31 _REPO_ID = "mv-test"
32
33
34 # ---------------------------------------------------------------------------
35 # Bootstrap helpers (same pattern as test_cmd_rm)
36 # ---------------------------------------------------------------------------
37
38
39 def _sha(data: bytes) -> str:
40 return blob_id(data)
41
42
43 _counter = 0
44
45
46 def _init_repo(path: pathlib.Path) -> pathlib.Path:
47 muse = path / ".muse"
48 for d in ("commits", "snapshots", "objects", "refs/heads", "code"):
49 (muse / d).mkdir(parents=True, exist_ok=True)
50 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
51 (muse / "repo.json").write_text(
52 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
53 )
54 return path
55
56
57 def _env(repo: pathlib.Path) -> Mapping[str, str]:
58 return {"MUSE_REPO_ROOT": str(repo)}
59
60
61 def _commit_files(root: pathlib.Path, files: Mapping[str, bytes]) -> str:
62 """Write *files* to disk + object store; create and record a commit."""
63 global _counter
64 _counter += 1
65 manifest: Manifest = {}
66 for rel_path, content in files.items():
67 obj_id = _sha(content)
68 write_object(root, obj_id, content)
69 manifest[rel_path] = obj_id
70 abs_path = root / rel_path
71 abs_path.parent.mkdir(parents=True, exist_ok=True)
72 abs_path.write_bytes(content)
73 snap_id = compute_snapshot_id(manifest)
74 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
75 committed_at = datetime.datetime.now(datetime.timezone.utc)
76 commit_id = compute_commit_id(
77 repo_id=_REPO_ID,
78 parent_ids=[],
79 snapshot_id=snap_id,
80 message=f"commit {_counter}",
81 committed_at_iso=committed_at.isoformat(),
82 )
83 write_commit(
84 root,
85 CommitRecord(
86 commit_id=commit_id,
87 repo_id=_REPO_ID,
88 created_on_branch="main",
89 snapshot_id=snap_id,
90 message=f"commit {_counter}",
91 committed_at=committed_at,
92 ),
93 )
94 (root / ".muse" / "refs" / "heads" / "main").write_text(commit_id, encoding="utf-8")
95 return commit_id
96
97
98 def _invoke(repo: pathlib.Path, *args: str):
99 from muse.cli.app import main as cli
100 return runner.invoke(cli, ["mv", *args], env=_env(repo))
101
102
103 # ---------------------------------------------------------------------------
104 # help
105 # ---------------------------------------------------------------------------
106
107
108 def test_mv_help() -> None:
109 from muse.cli.app import main as cli
110 result = runner.invoke(cli, ["mv", "--help"])
111 assert result.exit_code == 0
112 assert "mv" in result.output
113
114
115 # ---------------------------------------------------------------------------
116 # Unit — internal helpers
117 # ---------------------------------------------------------------------------
118
119
120 def test_resolve_source_returns_relative_posix(tmp_path: pathlib.Path) -> None:
121 from muse.cli.commands.mv import _resolve_path
122 root = _init_repo(tmp_path)
123 _commit_files(root, {"alpha.py": b"# a\n"})
124 rel = _resolve_path(root, "alpha.py")
125 assert rel == "alpha.py"
126
127
128 def test_resolve_path_rejects_traversal(tmp_path: pathlib.Path) -> None:
129 from muse.cli.commands.mv import _resolve_path
130 import sys
131 root = _init_repo(tmp_path)
132 with pytest.raises(SystemExit) as exc_info:
133 _resolve_path(root, "../../../etc/passwd")
134 assert exc_info.value.code != 0
135
136
137 def test_get_source_object_id_from_head(tmp_path: pathlib.Path) -> None:
138 from muse.cli.commands.mv import _get_source_object_id
139 root = _init_repo(tmp_path)
140 content = b"hello world\n"
141 _commit_files(root, {"src.py": content})
142 stage: StagedFileMap = {}
143 head_manifest: Manifest = {"src.py": _sha(content)}
144 obj_id = _get_source_object_id("src.py", head_manifest, stage)
145 assert obj_id == _sha(content)
146
147
148 def test_get_source_object_id_prefers_stage(tmp_path: pathlib.Path) -> None:
149 """Staged object_id takes precedence over HEAD manifest."""
150 from muse.cli.commands.mv import _get_source_object_id
151 root = _init_repo(tmp_path)
152 old_content = b"old\n"
153 new_content = b"new\n"
154 head_manifest: Manifest = {"src.py": _sha(old_content)}
155 stage: StagedFileMap = {"src.py": make_entry(_sha(new_content), "M")}
156 obj_id = _get_source_object_id("src.py", head_manifest, stage)
157 assert obj_id == _sha(new_content)
158
159
160 # ---------------------------------------------------------------------------
161 # Integration — basic move
162 # ---------------------------------------------------------------------------
163
164
165 def test_mv_renames_file_on_disk(tmp_path: pathlib.Path) -> None:
166 root = _init_repo(tmp_path)
167 _commit_files(root, {"alpha.py": b"# alpha\n"})
168
169 result = _invoke(root, "alpha.py", "beta.py")
170 assert result.exit_code == 0
171 assert not (root / "alpha.py").exists()
172 assert (root / "beta.py").exists()
173 assert (root / "beta.py").read_bytes() == b"# alpha\n"
174
175
176 def test_mv_stages_source_as_deleted(tmp_path: pathlib.Path) -> None:
177 root = _init_repo(tmp_path)
178 _commit_files(root, {"alpha.py": b"# a\n"})
179
180 _invoke(root, "alpha.py", "beta.py")
181 stage = read_stage(root)
182 assert "alpha.py" in stage
183 assert stage["alpha.py"]["mode"] == "D"
184
185
186 def test_mv_stages_dest_as_added(tmp_path: pathlib.Path) -> None:
187 root = _init_repo(tmp_path)
188 content = b"# a\n"
189 _commit_files(root, {"alpha.py": content})
190
191 _invoke(root, "alpha.py", "beta.py")
192 stage = read_stage(root)
193 assert "beta.py" in stage
194 assert stage["beta.py"]["mode"] == "A"
195
196
197 def test_mv_object_id_preserved(tmp_path: pathlib.Path) -> None:
198 """The dest entry must share the source's object_id — content unchanged."""
199 root = _init_repo(tmp_path)
200 content = b"# source content\n"
201 obj_id = _sha(content)
202 _commit_files(root, {"alpha.py": content})
203
204 _invoke(root, "alpha.py", "beta.py")
205 stage = read_stage(root)
206 assert stage["beta.py"]["object_id"] == obj_id
207
208
209 def test_mv_exit_code_zero_on_success(tmp_path: pathlib.Path) -> None:
210 root = _init_repo(tmp_path)
211 _commit_files(root, {"a.py": b"# a\n"})
212 result = _invoke(root, "a.py", "b.py")
213 assert result.exit_code == 0
214
215
216 def test_mv_prints_rename_line(tmp_path: pathlib.Path) -> None:
217 root = _init_repo(tmp_path)
218 _commit_files(root, {"a.py": b"# a\n"})
219 result = _invoke(root, "a.py", "b.py")
220 assert "a.py" in result.output
221 assert "b.py" in result.output
222
223
224 # ---------------------------------------------------------------------------
225 # Integration — --json
226 # ---------------------------------------------------------------------------
227
228
229 def test_mv_json_output_structure(tmp_path: pathlib.Path) -> None:
230 root = _init_repo(tmp_path)
231 content = b"# j\n"
232 _commit_files(root, {"j.py": content})
233
234 result = _invoke(root, "--json", "j.py", "k.py")
235 assert result.exit_code == 0
236 data = json.loads(result.stdout)
237 assert data["source"] == "j.py"
238 assert data["dest"] == "k.py"
239 assert data["status"] in ("moved", "dry_run")
240 assert data["dry_run"] is False
241 assert data["object_id"] == _sha(content)
242
243
244 # ---------------------------------------------------------------------------
245 # Integration — --dry-run
246 # ---------------------------------------------------------------------------
247
248
249 def test_mv_dry_run_no_disk_change(tmp_path: pathlib.Path) -> None:
250 root = _init_repo(tmp_path)
251 _commit_files(root, {"alpha.py": b"# a\n"})
252
253 result = _invoke(root, "--dry-run", "alpha.py", "beta.py")
254 assert result.exit_code == 0
255 assert (root / "alpha.py").exists()
256 assert not (root / "beta.py").exists()
257
258
259 def test_mv_dry_run_no_stage_change(tmp_path: pathlib.Path) -> None:
260 root = _init_repo(tmp_path)
261 _commit_files(root, {"alpha.py": b"# a\n"})
262
263 _invoke(root, "--dry-run", "alpha.py", "beta.py")
264 stage = read_stage(root)
265 assert "alpha.py" not in stage
266 assert "beta.py" not in stage
267
268
269 def test_mv_dry_run_json(tmp_path: pathlib.Path) -> None:
270 root = _init_repo(tmp_path)
271 _commit_files(root, {"alpha.py": b"# a\n"})
272
273 result = _invoke(root, "--dry-run", "--json", "alpha.py", "beta.py")
274 assert result.exit_code == 0
275 data = json.loads(result.stdout)
276 assert data["dry_run"] is True
277 assert data["status"] == "dry_run"
278
279
280 # ---------------------------------------------------------------------------
281 # Integration — error conditions
282 # ---------------------------------------------------------------------------
283
284
285 def test_mv_source_not_tracked_exits_nonzero(tmp_path: pathlib.Path) -> None:
286 root = _init_repo(tmp_path)
287 _commit_files(root, {"other.py": b"# o\n"})
288 (root / "ghost.py").write_text("# untracked\n")
289
290 result = _invoke(root, "ghost.py", "dest.py")
291 assert result.exit_code != 0
292
293
294 def test_mv_source_not_on_disk_exits_nonzero(tmp_path: pathlib.Path) -> None:
295 """Source tracked in HEAD but deleted from disk → error unless --force."""
296 root = _init_repo(tmp_path)
297 _commit_files(root, {"missing.py": b"# m\n"})
298 (root / "missing.py").unlink()
299
300 result = _invoke(root, "missing.py", "dest.py")
301 assert result.exit_code != 0
302
303
304 def test_mv_dest_already_tracked_exits_nonzero(tmp_path: pathlib.Path) -> None:
305 root = _init_repo(tmp_path)
306 _commit_files(root, {"src.py": b"# src\n", "dst.py": b"# dst\n"})
307
308 result = _invoke(root, "src.py", "dst.py")
309 assert result.exit_code != 0
310
311
312 def test_mv_dest_exists_on_disk_exits_nonzero(tmp_path: pathlib.Path) -> None:
313 root = _init_repo(tmp_path)
314 _commit_files(root, {"src.py": b"# src\n"})
315 (root / "dst.py").write_text("# untracked but on disk\n")
316
317 result = _invoke(root, "src.py", "dst.py")
318 assert result.exit_code != 0
319
320
321 # ---------------------------------------------------------------------------
322 # Integration — --force
323 # ---------------------------------------------------------------------------
324
325
326 def test_mv_force_allows_tracked_dest(tmp_path: pathlib.Path) -> None:
327 root = _init_repo(tmp_path)
328 _commit_files(root, {"src.py": b"# src\n", "dst.py": b"# dst\n"})
329
330 result = _invoke(root, "--force", "src.py", "dst.py")
331 assert result.exit_code == 0
332 stage = read_stage(root)
333 assert "src.py" in stage and stage["src.py"]["mode"] == "D"
334 assert "dst.py" in stage and stage["dst.py"]["mode"] in ("A", "M")
335
336
337 def test_mv_force_allows_untracked_dest_on_disk(tmp_path: pathlib.Path) -> None:
338 root = _init_repo(tmp_path)
339 _commit_files(root, {"src.py": b"# src\n"})
340 (root / "dst.py").write_text("# untracked\n")
341
342 result = _invoke(root, "--force", "src.py", "dst.py")
343 assert result.exit_code == 0
344 assert not (root / "src.py").exists()
345 assert (root / "dst.py").read_bytes() == b"# src\n"
346
347
348 # ---------------------------------------------------------------------------
349 # Integration — move into directory
350 # ---------------------------------------------------------------------------
351
352
353 def test_mv_into_existing_directory(tmp_path: pathlib.Path) -> None:
354 """mv file.py dir/ moves to dir/file.py when dir/ exists."""
355 root = _init_repo(tmp_path)
356 _commit_files(root, {"alpha.py": b"# a\n"})
357 (root / "subdir").mkdir()
358
359 result = _invoke(root, "alpha.py", "subdir/")
360 assert result.exit_code == 0
361 assert (root / "subdir" / "alpha.py").exists()
362 stage = read_stage(root)
363 assert "subdir/alpha.py" in stage
364 assert stage["subdir/alpha.py"]["mode"] == "A"
365 assert stage["alpha.py"]["mode"] == "D"
366
367
368 def test_mv_into_directory_preserves_object_id(tmp_path: pathlib.Path) -> None:
369 root = _init_repo(tmp_path)
370 content = b"# content\n"
371 _commit_files(root, {"f.py": content})
372 (root / "pkg").mkdir()
373
374 _invoke(root, "f.py", "pkg/")
375 stage = read_stage(root)
376 assert stage["pkg/f.py"]["object_id"] == _sha(content)
377
378
379 # ---------------------------------------------------------------------------
380 # Integration — staged-only file (never committed)
381 # ---------------------------------------------------------------------------
382
383
384 def test_mv_staged_only_source_updates_stage(tmp_path: pathlib.Path) -> None:
385 """A file staged as 'A' (never committed) is moved: stage entry replaced."""
386 root = _init_repo(tmp_path)
387 # Repo has at least one commit so HEAD exists
388 _commit_files(root, {"anchor.py": b"# anchor\n"})
389 # Stage a new file that has never been committed
390 content = b"# new file\n"
391 obj_id = _sha(content)
392 write_object(root, obj_id, content)
393 (root / "new.py").write_bytes(content)
394 stage = read_stage(root)
395 stage["new.py"] = make_entry(obj_id, "A")
396 write_stage(root, stage)
397
398 result = _invoke(root, "new.py", "renamed.py")
399 assert result.exit_code == 0
400 stage_after = read_stage(root)
401 # "new.py" must be gone from stage (was never committed → no "D" entry)
402 assert "new.py" not in stage_after
403 assert "renamed.py" in stage_after
404 assert stage_after["renamed.py"]["mode"] == "A"
405 assert stage_after["renamed.py"]["object_id"] == obj_id
406
407
408 # ---------------------------------------------------------------------------
409 # Integration — subdirectory paths
410 # ---------------------------------------------------------------------------
411
412
413 def test_mv_across_subdirectories(tmp_path: pathlib.Path) -> None:
414 root = _init_repo(tmp_path)
415 (root / "src").mkdir()
416 (root / "lib").mkdir()
417 _commit_files(root, {"src/module.py": b"# m\n"})
418
419 result = _invoke(root, "src/module.py", "lib/module.py")
420 assert result.exit_code == 0
421 assert not (root / "src" / "module.py").exists()
422 assert (root / "lib" / "module.py").exists()
423 stage = read_stage(root)
424 assert stage["src/module.py"]["mode"] == "D"
425 assert stage["lib/module.py"]["mode"] == "A"
426
427
428 # ---------------------------------------------------------------------------
429 # Security
430 # ---------------------------------------------------------------------------
431
432
433 def test_mv_source_path_traversal_rejected(tmp_path: pathlib.Path) -> None:
434 root = _init_repo(tmp_path)
435 _commit_files(root, {"anchor.py": b"# a\n"})
436
437 result = _invoke(root, "../../../etc/passwd", "dest.py")
438 assert result.exit_code != 0
439
440
441 def test_mv_dest_path_traversal_rejected(tmp_path: pathlib.Path) -> None:
442 root = _init_repo(tmp_path)
443 _commit_files(root, {"src.py": b"# s\n"})
444
445 result = _invoke(root, "src.py", "../../../tmp/evil.py")
446 assert result.exit_code != 0
447
448
449 def test_mv_outside_repo_source_rejected(tmp_path: pathlib.Path) -> None:
450 root = _init_repo(tmp_path / "repo")
451 _commit_files(root, {"anchor.py": b"# a\n"})
452 outside = tmp_path / "outside.py"
453 outside.write_text("# outside\n")
454
455 result = _invoke(root, str(outside), "dest.py")
456 assert result.exit_code != 0
457
458
459 # ---------------------------------------------------------------------------
460 # Stress
461 # ---------------------------------------------------------------------------
462
463
464 def test_mv_stress_200_files_move_half(tmp_path: pathlib.Path) -> None:
465 """Move 100 of 200 tracked files; all stage entries must be consistent."""
466 root = _init_repo(tmp_path)
467 files = {f"file_{i}.py": f"# {i}\n".encode() for i in range(200)}
468 _commit_files(root, files)
469
470 (root / "dest").mkdir()
471 for i in range(0, 200, 2): # move even-numbered files
472 result = _invoke(root, f"file_{i}.py", f"dest/file_{i}.py")
473 assert result.exit_code == 0, f"Move failed for file_{i}.py: {result.output}"
474
475 stage = read_stage(root)
476 for i in range(0, 200, 2):
477 assert stage[f"file_{i}.py"]["mode"] == "D"
478 assert stage[f"dest/file_{i}.py"]["mode"] == "A"
479 # Odd-numbered files untouched
480 for i in range(1, 200, 2):
481 assert f"file_{i}.py" not in stage
482
483
484 class TestRegisterFlags:
485 def test_default_json_out_is_false(self):
486 import argparse
487 from muse.cli.commands.mv import register
488 p = argparse.ArgumentParser()
489 subs = p.add_subparsers()
490 register(subs)
491 args = p.parse_args(["mv", "a.py", "b.py"])
492 assert args.json_out is False
493
494 def test_json_flag_sets_json_out(self):
495 import argparse
496 from muse.cli.commands.mv import register
497 p = argparse.ArgumentParser()
498 subs = p.add_subparsers()
499 register(subs)
500 args = p.parse_args(["mv", "a.py", "b.py", "--json"])
501 assert args.json_out is True
502
503 def test_j_shorthand_sets_json_out(self):
504 import argparse
505 from muse.cli.commands.mv import register
506 p = argparse.ArgumentParser()
507 subs = p.add_subparsers()
508 register(subs)
509 args = p.parse_args(["mv", "a.py", "b.py", "-j"])
510 assert args.json_out is True
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 137 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor 140 days ago