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