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