gabriel / muse public

test_cmd_stash.py file-level

at sha256:6 · View file ↗ · Intel ↗

History
1 files
1 commits
0 hotspots
0 🧊 dead
0 💥 blast risk
sha256:0 test: validate staging push and MP workflow for aaronrene · · Jul 7, 2026
1 """Comprehensive tests for ``muse stash``.
2
3 Covers:
4 - Unit: _load_stash / _save_stash atomic write, size guard
5 - Integration: stash → pop, list, drop
6 - E2E: full CLI via CliRunner
7 - Security: stash.json size limit, atomic writes, sanitized output
8 - Stress: many stash entries, repeated save/load
9 """
10
11 from __future__ import annotations
12
13 import datetime
14 import json
15 import pathlib
16 import uuid
17
18 import pytest
19 from tests.cli_test_helper import CliRunner
20
21 cli = None # argparse migration — CliRunner ignores this arg
22
23 runner = CliRunner()
24
25
26 # ---------------------------------------------------------------------------
27 # Helpers
28 # ---------------------------------------------------------------------------
29
30 def _env(root: pathlib.Path) -> Manifest:
31 return {"MUSE_REPO_ROOT": str(root)}
32
33
34 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
35 muse_dir = tmp_path / ".muse"
36 muse_dir.mkdir()
37 repo_id = str(uuid.uuid4())
38 (muse_dir / "repo.json").write_text(json.dumps({
39 "repo_id": repo_id,
40 "domain": "code",
41 "default_branch": "main",
42 "created_at": "2025-01-01T00:00:00+00:00",
43 }), encoding="utf-8")
44 (muse_dir / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
45 (muse_dir / "refs" / "heads").mkdir(parents=True)
46 (muse_dir / "snapshots").mkdir()
47 (muse_dir / "commits").mkdir()
48 (muse_dir / "objects").mkdir()
49 return tmp_path, repo_id
50
51
52 def _make_commit(root: pathlib.Path, repo_id: str, message: str = "init") -> str:
53 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
54 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
55
56 ref_file = root / ".muse" / "refs" / "heads" / "main"
57 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
58 manifest: Manifest = {}
59 snap_id = compute_snapshot_id(manifest)
60 committed_at = datetime.datetime.now(datetime.timezone.utc)
61 commit_id = compute_commit_id(
62 parent_ids=[parent_id] if parent_id else [],
63 snapshot_id=snap_id, message=message,
64 committed_at_iso=committed_at.isoformat(),
65 )
66 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
67 write_commit(root, CommitRecord(
68 commit_id=commit_id, repo_id=repo_id, branch="main",
69 snapshot_id=snap_id, message=message, committed_at=committed_at,
70 parent_commit_id=parent_id,
71 ))
72 ref_file.parent.mkdir(parents=True, exist_ok=True)
73 ref_file.write_text(commit_id, encoding="utf-8")
74 return commit_id
75
76
77 # ---------------------------------------------------------------------------
78 # Unit tests
79 # ---------------------------------------------------------------------------
80
81 class TestStashUnit:
82 def test_load_stash_empty(self, tmp_path: pathlib.Path) -> None:
83 root, _ = _init_repo(tmp_path)
84 from muse.cli.commands.stash import _load_stash
85 assert _load_stash(root) == []
86
87 def test_save_and_load_stash_roundtrip(self, tmp_path: pathlib.Path) -> None:
88 root, _ = _init_repo(tmp_path)
89 from muse.cli.commands.stash import _load_stash, _save_stash, StashEntry
90 entry = StashEntry(
91 snapshot_id="a" * 64, delta={"file.mid": "b" * 64}, deleted=[],
92 branch="main", stashed_at="2025-01-01T00:00:00+00:00", message=None,
93 )
94 _save_stash(root, [entry])
95 loaded = _load_stash(root)
96 assert len(loaded) == 1
97 assert loaded[0]["snapshot_id"] == "a" * 64
98 assert loaded[0]["branch"] == "main"
99 assert loaded[0]["delta"] == {"file.mid": "b" * 64}
100 assert loaded[0]["deleted"] == []
101
102 def test_save_stash_is_atomic(self, tmp_path: pathlib.Path) -> None:
103 """After _save_stash, no temp files should remain in .muse/."""
104 root, _ = _init_repo(tmp_path)
105 from muse.cli.commands.stash import _save_stash, StashEntry
106 entry = StashEntry(
107 snapshot_id="c" * 64, delta={}, deleted=[],
108 branch="dev", stashed_at="2025-01-01T00:00:00+00:00", message=None,
109 )
110 _save_stash(root, [entry])
111 tmp_files = list((root / ".muse").glob(".stash_tmp_*"))
112 assert tmp_files == []
113 assert (root / ".muse" / "stash.json").exists()
114
115 def test_load_stash_ignores_oversized_file(self, tmp_path: pathlib.Path) -> None:
116 root, _ = _init_repo(tmp_path)
117 stash_path = root / ".muse" / "stash.json"
118 stash_path.write_bytes(b"x" * (65 * 1024 * 1024)) # 65 MiB > 64 MiB limit
119 from muse.cli.commands.stash import _load_stash
120 result = _load_stash(root)
121 assert result == []
122
123
124 # ---------------------------------------------------------------------------
125 # Integration tests
126 # ---------------------------------------------------------------------------
127
128 class TestStashIntegration:
129 def test_stash_with_no_changes_reports_nothing(self, tmp_path: pathlib.Path) -> None:
130 root, repo_id = _init_repo(tmp_path)
131 _make_commit(root, repo_id)
132 result = runner.invoke(cli, ["stash"], env=_env(root), catch_exceptions=False)
133 assert "Nothing to stash" in result.output
134
135 def test_stash_list_empty(self, tmp_path: pathlib.Path) -> None:
136 root, repo_id = _init_repo(tmp_path)
137 _make_commit(root, repo_id)
138 result = runner.invoke(cli, ["stash", "list"], env=_env(root), catch_exceptions=False)
139 assert "No stash entries" in result.output
140
141 def test_stash_pop_empty_fails(self, tmp_path: pathlib.Path) -> None:
142 root, repo_id = _init_repo(tmp_path)
143 _make_commit(root, repo_id)
144 result = runner.invoke(cli, ["stash", "pop"], env=_env(root))
145 assert result.exit_code != 0
146
147 def test_stash_drop_empty_fails(self, tmp_path: pathlib.Path) -> None:
148 root, repo_id = _init_repo(tmp_path)
149 _make_commit(root, repo_id)
150 result = runner.invoke(cli, ["stash", "drop"], env=_env(root))
151 assert result.exit_code != 0
152
153 def test_stash_list_shows_entries(self, tmp_path: pathlib.Path) -> None:
154 root, _ = _init_repo(tmp_path)
155 from muse.cli.commands.stash import _save_stash, StashEntry
156 _save_stash(root, [
157 StashEntry(snapshot_id="a" * 64, delta={}, deleted=[],
158 branch="main", stashed_at="2025-01-01T00:00:00+00:00", message=None),
159 ])
160 result = runner.invoke(cli, ["stash", "list"], env=_env(root), catch_exceptions=False)
161 assert "stash@{0}" in result.output
162 assert "main" in result.output
163
164
165 # ---------------------------------------------------------------------------
166 # Security tests
167 # ---------------------------------------------------------------------------
168
169 class TestStashSecurity:
170 def test_stash_list_sanitizes_branch_name_with_control_chars(
171 self, tmp_path: pathlib.Path
172 ) -> None:
173 root, _ = _init_repo(tmp_path)
174 from muse.cli.commands.stash import _save_stash, StashEntry
175 malicious_branch = "feat/\x1b[31mred\x1b[0m"
176 _save_stash(root, [
177 StashEntry(snapshot_id="a" * 64, delta={}, deleted=[],
178 branch=malicious_branch, stashed_at="2025-01-01T00:00:00+00:00",
179 message=None),
180 ])
181 result = runner.invoke(cli, ["stash", "list"], env=_env(root), catch_exceptions=False)
182 assert result.exit_code == 0
183 assert "\x1b" not in result.output
184
185 def test_stash_pop_sanitizes_branch_name(self, tmp_path: pathlib.Path) -> None:
186 root, repo_id = _init_repo(tmp_path)
187 _make_commit(root, repo_id)
188 from muse.cli.commands.stash import _save_stash, StashEntry
189 _save_stash(root, [
190 StashEntry(snapshot_id="a" * 64, delta={}, deleted=[],
191 branch="feat/\x1b[31mred\x1b[0m",
192 stashed_at="2025-01-01T00:00:00+00:00", message=None),
193 ])
194 result = runner.invoke(cli, ["stash", "pop"], env=_env(root), catch_exceptions=False)
195 assert "\x1b" not in result.output
196
197
198 # ---------------------------------------------------------------------------
199 # Stress tests
200 # ---------------------------------------------------------------------------
201
202 class TestStashStress:
203 def test_many_stash_entries_save_load(self, tmp_path: pathlib.Path) -> None:
204 root, _ = _init_repo(tmp_path)
205 from muse.cli.commands.stash import _save_stash, _load_stash, StashEntry
206 entries = [
207 StashEntry(snapshot_id=f"{'a' * 63}{i % 10}",
208 delta={"file.mid": "b" * 64}, deleted=[], branch="main",
209 stashed_at=f"2025-01-{i % 28 + 1:02d}T00:00:00+00:00", message=None)
210 for i in range(50)
211 ]
212 _save_stash(root, entries)
213 loaded = _load_stash(root)
214 assert len(loaded) == 50
215
216 def test_repeated_save_load_no_corruption(self, tmp_path: pathlib.Path) -> None:
217 root, _ = _init_repo(tmp_path)
218 from muse.cli.commands.stash import _save_stash, _load_stash, StashEntry
219 for i in range(20):
220 entry = StashEntry(snapshot_id=f"{'b' * 63}{i % 10}",
221 delta={}, deleted=[], branch="main",
222 stashed_at="2025-01-01T00:00:00+00:00", message=None)
223 loaded = _load_stash(root)
224 loaded.insert(0, entry)
225 _save_stash(root, loaded)
226
227 final = _load_stash(root)
228 assert len(final) == 20
229
230
231 # ---------------------------------------------------------------------------
232 # Regression tests — delta-based stash correctness
233 # ---------------------------------------------------------------------------
234
235 class TestStashDeltaRegression:
236 """Regression tests for the stash full-snapshot bug.
237
238 Before the fix, ``muse stash`` saved the entire working tree (all tracked
239 files) and ``stash pop`` wrote ALL of them back — overwriting any files
240 that had been committed since the stash was created with their older
241 stashed versions.
242
243 After the fix, only the delta (files that differ from HEAD) is saved and
244 restored, so intervening commits are never clobbered.
245 """
246
247 def _make_commit_with_files(
248 self,
249 root: pathlib.Path,
250 repo_id: str,
251 files: Manifest,
252 message: str = "commit",
253 ) -> str:
254 """Write *files* to disk and record a commit whose snapshot includes them."""
255 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
256 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
257 from muse.core.object_store import write_object_from_path
258 import datetime
259
260 for rel, content in files.items():
261 p = root / rel
262 p.parent.mkdir(parents=True, exist_ok=True)
263 p.write_text(content, encoding="utf-8")
264
265 manifest: Manifest = {}
266 for rel in files:
267 p = root / rel
268 import hashlib
269 data = p.read_bytes()
270 oid = hashlib.sha256(data).hexdigest()
271 manifest[rel] = oid
272 write_object_from_path(root, oid, p)
273
274 ref_file = root / ".muse" / "refs" / "heads" / "main"
275 parent_id = ref_file.read_text().strip() if ref_file.exists() and ref_file.stat().st_size else None
276 snap_id = compute_snapshot_id(manifest)
277 committed_at = datetime.datetime.now(datetime.timezone.utc)
278 commit_id = compute_commit_id(
279 parent_ids=[parent_id] if parent_id else [],
280 snapshot_id=snap_id,
281 message=message,
282 committed_at_iso=committed_at.isoformat(),
283 )
284 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
285 write_commit(root, CommitRecord(
286 commit_id=commit_id, repo_id=repo_id, branch="main",
287 snapshot_id=snap_id, message=message, committed_at=committed_at,
288 parent_commit_id=parent_id,
289 ))
290 ref_file.parent.mkdir(parents=True, exist_ok=True)
291 ref_file.write_text(commit_id, encoding="utf-8")
292 return commit_id
293
294 def test_stash_saves_only_delta_not_full_snapshot(
295 self, tmp_path: pathlib.Path
296 ) -> None:
297 """Stash entry delta must contain only changed files, not all tracked files."""
298 root, repo_id = _init_repo(tmp_path)
299 # Commit two files.
300 self._make_commit_with_files(root, repo_id, {
301 "stable.py": "unchanged content\n",
302 "changing.py": "original content\n",
303 })
304 # Modify only one.
305 (root / "changing.py").write_text("modified content\n", encoding="utf-8")
306
307 result = runner.invoke(cli, ["stash"], env=_env(root), catch_exceptions=False)
308 assert result.exit_code == 0
309
310 from muse.cli.commands.stash import _load_stash
311 entries = _load_stash(root)
312 assert len(entries) == 1
313 # Delta must contain only the changed file.
314 assert "changing.py" in entries[0]["delta"]
315 assert "stable.py" not in entries[0]["delta"]
316 assert entries[0]["deleted"] == []
317
318 def test_stash_pop_does_not_overwrite_newer_commits(
319 self, tmp_path: pathlib.Path
320 ) -> None:
321 """Pop must not clobber files that were committed after the stash was created.
322
323 Sequence:
324 1. Commit file_a="v1" and file_b="original"
325 2. Modify file_b only → stash
326 3. Commit file_a="v2" (new committed version)
327 4. Pop stash
328 5. Assert file_a == "v2" (committed version preserved)
329 6. Assert file_b == "modified" (stashed change restored)
330 """
331 root, repo_id = _init_repo(tmp_path)
332
333 # Step 1: initial commit.
334 self._make_commit_with_files(root, repo_id, {
335 "file_a.py": "v1\n",
336 "file_b.py": "original\n",
337 }, message="initial")
338
339 # Step 2: modify file_b and stash.
340 (root / "file_b.py").write_text("modified\n", encoding="utf-8")
341 stash_result = runner.invoke(cli, ["stash"], env=_env(root), catch_exceptions=False)
342 assert stash_result.exit_code == 0
343 # After stash, working tree is at HEAD — file_b back to "original".
344 assert (root / "file_b.py").read_text() == "original\n"
345
346 # Step 3: commit a new version of file_a (simulates work done after stash).
347 self._make_commit_with_files(root, repo_id, {
348 "file_a.py": "v2\n",
349 "file_b.py": "original\n",
350 }, message="post-stash commit")
351 assert (root / "file_a.py").read_text() == "v2\n"
352
353 # Step 4: pop the stash.
354 pop_result = runner.invoke(cli, ["stash", "pop"], env=_env(root), catch_exceptions=False)
355 assert pop_result.exit_code == 0
356
357 # Step 5 & 6: verify.
358 assert (root / "file_a.py").read_text() == "v2\n", (
359 "stash pop overwrote file_a with its stashed version — regression!"
360 )
361 assert (root / "file_b.py").read_text() == "modified\n", (
362 "stash pop did not restore file_b"
363 )
364
365 def test_stash_pop_restores_deleted_files(self, tmp_path: pathlib.Path) -> None:
366 """A file deleted before stashing must be re-deleted on pop."""
367 root, repo_id = _init_repo(tmp_path)
368 self._make_commit_with_files(root, repo_id, {
369 "keep.py": "keep\n",
370 "to_delete.py": "gone\n",
371 })
372 # Delete one file before stashing.
373 (root / "to_delete.py").unlink()
374 runner.invoke(cli, ["stash"], env=_env(root), catch_exceptions=False)
375 # After stash, HEAD is restored — to_delete.py is back.
376 assert (root / "to_delete.py").exists()
377
378 runner.invoke(cli, ["stash", "pop"], env=_env(root), catch_exceptions=False)
379 assert not (root / "to_delete.py").exists(), (
380 "stash pop did not re-delete the file that was deleted before stashing"
381 )
382 assert (root / "keep.py").exists()
383
384 def test_stash_does_not_record_ignored_extant_as_deleted(
385 self, tmp_path: pathlib.Path
386 ) -> None:
387 """A file that was tracked in HEAD, added to .museignore, and still
388 present on disk must NOT appear in the stash entry's deleted list.
389
390 Recording it as deleted would cause stash pop to unlink it — silently
391 destroying a build artifact that is still needed on disk.
392 """
393 root, repo_id = _init_repo(tmp_path)
394 self._make_commit_with_files(root, repo_id, {
395 "src.py": "source\n",
396 "app.js": "// build output\n",
397 })
398 # Add app.js to .museignore (simulates adding a build artifact to ignore).
399 (root / ".museignore").write_text(
400 '[global]\npatterns = ["app.js"]\n', encoding="utf-8"
401 )
402 # Modify src.py so there's actually something to stash.
403 (root / "src.py").write_text("modified source\n", encoding="utf-8")
404
405 result = runner.invoke(cli, ["stash"], env=_env(root), catch_exceptions=False)
406 assert result.exit_code == 0
407
408 from muse.cli.commands.stash import _load_stash
409 entries = _load_stash(root)
410 assert len(entries) == 1
411 assert "app.js" not in entries[0]["deleted"], (
412 "ignored-and-extant file must not be recorded as deleted in stash"
413 )
414
415 def test_stash_pop_does_not_unlink_ignored_extant_file(
416 self, tmp_path: pathlib.Path
417 ) -> None:
418 """Stash pop must not delete a file that is in .museignore and still on disk.
419
420 Regression guard for the bug where stash recorded ignored-and-extant
421 files as deleted, then pop called unlink() on them.
422 """
423 root, repo_id = _init_repo(tmp_path)
424 self._make_commit_with_files(root, repo_id, {
425 "src.py": "source\n",
426 "app.js": "// build output\n",
427 })
428 (root / ".museignore").write_text(
429 '[global]\npatterns = ["app.js"]\n', encoding="utf-8"
430 )
431 (root / "src.py").write_text("modified\n", encoding="utf-8")
432
433 runner.invoke(cli, ["stash"], env=_env(root), catch_exceptions=False)
434 assert (root / "app.js").exists(), "stash must not delete an ignored-and-extant file"
435
436 runner.invoke(cli, ["stash", "pop"], env=_env(root), catch_exceptions=False)
437 assert (root / "app.js").exists(), (
438 "stash pop must not unlink a file that was merely ignored, not deleted"
439 )
440 assert (root / "src.py").read_text() == "modified\n"
441
442 def test_stash_show_lists_only_changed_files(self, tmp_path: pathlib.Path) -> None:
443 """stash show must list only changed files, not the entire working tree."""
444 root, repo_id = _init_repo(tmp_path)
445 self._make_commit_with_files(root, repo_id, {
446 "a.py": "a\n",
447 "b.py": "b\n",
448 "c.py": "c\n",
449 })
450 # Modify only one file.
451 (root / "b.py").write_text("b modified\n", encoding="utf-8")
452 runner.invoke(cli, ["stash"], env=_env(root), catch_exceptions=False)
453
454 result = runner.invoke(
455 cli, ["stash", "show", "--json"], env=_env(root), catch_exceptions=False
456 )
457 assert result.exit_code == 0
458 data = json.loads(result.output)
459 assert data["files_count"] == 1
460 assert data["files"] == ["b.py"]
461 # Unchanged files must not appear.
462 assert "a.py" not in data["files"]
463 assert "c.py" not in data["files"]
464
465 def test_stash_files_count_is_delta_size(self, tmp_path: pathlib.Path) -> None:
466 """files_count in stash output reflects the delta, not the full repo."""
467 root, repo_id = _init_repo(tmp_path)
468 self._make_commit_with_files(root, repo_id, {
469 "x.py": "x\n", "y.py": "y\n", "z.py": "z\n",
470 })
471 (root / "x.py").write_text("x changed\n", encoding="utf-8")
472 result = runner.invoke(
473 cli, ["stash", "--json"], env=_env(root), catch_exceptions=False
474 )
475 assert result.exit_code == 0
476 data = json.loads(result.output)
477 assert data["files_count"] == 1 # only x.py changed