gabriel / muse public
test_cmd_shelf.py python
1,654 lines 66.3 KB
Raw
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ breaking 153 days ago
1 """Comprehensive tests for ``muse shelf``.
2
3 Covers:
4 - Unit: _load_shelf / _save_shelf atomic write + guards, _resolve_entry,
5 _compute_shelf_id, _generate_name, _apply_shelf_snapshot,
6 _verify_snapshot_objects
7 - Integration: save, list, read, apply, pop, drop, diff — JSON schemas,
8 text output, filters, agent fields
9 - End-to-end: full CLI round-trips via CliRunner
10 - Stress: many entries, concurrent isolated repos, repeated save/load
11 - Data integrity: already-current detection, snapshot completeness,
12 content-address stability
13 - Performance: save + pop under 5 s
14 - Security: symlink guard, size limit, ANSI injection in names / intent,
15 invalid --format exits 1
16
17 Test categories
18 ---------------
19 - unit : pure helper functions, no repo needed
20 - integration : programmatic API + JSON schema validation
21 - e2e : CliRunner full round-trips
22 - stress : volume and concurrency
23 - data-integrity: already-current detection, manifest correctness
24 - performance : timing assertions
25 - security : injection, path-traversal, oversized file guards
26 - docstrings : public API coverage
27 """
28
29 from __future__ import annotations
30
31 import argparse
32 import datetime
33 import inspect
34 import json
35 import os
36 import pathlib
37 import threading
38 import time
39 import uuid
40 from typing import Any
41
42 import pytest
43 from tests.cli_test_helper import CliRunner
44
45 cli = None # argparse migration — CliRunner ignores this arg
46 runner = CliRunner()
47
48
49 # ---------------------------------------------------------------------------
50 # Shared helpers
51 # ---------------------------------------------------------------------------
52
53
54 def _env(root: pathlib.Path) -> dict[str, str]:
55 return {"MUSE_REPO_ROOT": str(root)}
56
57
58 def _init_repo(tmp_path: pathlib.Path, branch: str = "main") -> tuple[pathlib.Path, str]:
59 """Create a minimal Muse repo structure on disk."""
60 muse_dir = tmp_path / ".muse"
61 muse_dir.mkdir()
62 repo_id = str(uuid.uuid4())
63 (muse_dir / "repo.json").write_text(json.dumps({
64 "repo_id": repo_id,
65 "domain": "code",
66 "default_branch": branch,
67 "created_at": "2025-01-01T00:00:00+00:00",
68 }), encoding="utf-8")
69 (muse_dir / "HEAD").write_text(f"ref: refs/heads/{branch}", encoding="utf-8")
70 (muse_dir / "refs" / "heads").mkdir(parents=True)
71 (muse_dir / "snapshots").mkdir()
72 (muse_dir / "commits").mkdir()
73 (muse_dir / "objects").mkdir()
74 return tmp_path, repo_id
75
76
77 def _make_commit(
78 root: pathlib.Path,
79 repo_id: str,
80 message: str = "init",
81 branch: str = "main",
82 manifest: dict[str, str] | None = None,
83 ) -> str:
84 """Write a commit to the repo with the given manifest."""
85 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
86 from muse.core.snapshot import compute_snapshot_id, compute_commit_id
87
88 ref_file = root / ".muse" / "refs" / "heads" / branch
89 parent_id = ref_file.read_text().strip() if ref_file.exists() else None
90 m: dict[str, str] = manifest or {}
91 snap_id = compute_snapshot_id(m)
92 committed_at = datetime.datetime.now(datetime.timezone.utc)
93 commit_id = compute_commit_id(
94 parent_ids=[parent_id] if parent_id else [],
95 snapshot_id=snap_id, message=message,
96 committed_at_iso=committed_at.isoformat(),
97 )
98 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=m))
99 write_commit(root, CommitRecord(
100 commit_id=commit_id, repo_id=repo_id, branch=branch,
101 snapshot_id=snap_id, message=message, committed_at=committed_at,
102 parent_commit_id=parent_id,
103 ))
104 ref_file.parent.mkdir(parents=True, exist_ok=True)
105 ref_file.write_text(commit_id, encoding="utf-8")
106 return commit_id
107
108
109 def _write_object(root: pathlib.Path, content: bytes) -> str:
110 """Write raw bytes to the object store, returning the sha256:-prefixed ID."""
111 import hashlib
112 digest = hashlib.sha256(content).hexdigest()
113 obj_id = f"sha256:{digest}"
114 obj_dir = root / ".muse" / "objects" / digest[:2]
115 obj_dir.mkdir(parents=True, exist_ok=True)
116 (obj_dir / digest[2:]).write_bytes(content)
117 return obj_id
118
119
120 def _make_shelf_entry(
121 name: str = "dev/000",
122 branch: str = "main",
123 snapshot: dict[str, str] | None = None,
124 deleted: list[str] | None = None,
125 intent_type: str = "checkpoint",
126 intent: str | None = None,
127 resumable: bool = False,
128 tags: list[str] | None = None,
129 created_by: str = "human",
130 ) -> dict:
131 """Build a raw shelf-entry dict (no id field) suitable for _compute_shelf_id."""
132 return {
133 "name": name,
134 "snapshot": snapshot or {},
135 "deleted": deleted or [],
136 "snapshot_id": "sha256:" + "a" * 64,
137 "parent_commit": "sha256:" + "b" * 64,
138 "branch": branch,
139 "created_at": "2025-01-01T00:00:00+00:00",
140 "created_by": created_by,
141 "intent_type": intent_type,
142 "intent": intent,
143 "resumable": resumable,
144 "tags": tags or [],
145 "expires_at": None,
146 "domain_state": {},
147 }
148
149
150 # ---------------------------------------------------------------------------
151 # Fixtures
152 # ---------------------------------------------------------------------------
153
154
155 @pytest.fixture()
156 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
157 """Fresh repo with one committed file (a.py) and one dirty file (b.py)."""
158 monkeypatch.chdir(tmp_path)
159 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
160 r = runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
161 assert r.exit_code == 0, r.output
162 (tmp_path / "a.py").write_text("x = 1\n")
163 r = runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
164 assert r.exit_code == 0, r.output
165 (tmp_path / "b.py").write_text("y = 2\n")
166 return tmp_path
167
168
169 @pytest.fixture()
170 def shelved_repo(repo: pathlib.Path) -> pathlib.Path:
171 """repo fixture with one shelf entry already saved."""
172 r = runner.invoke(
173 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
174 )
175 assert r.exit_code == 0, r.output
176 return repo
177
178
179 # ---------------------------------------------------------------------------
180 # Unit — _compute_shelf_id
181 # ---------------------------------------------------------------------------
182
183
184 class TestComputeShelfId:
185 """Unit tests for content-addressed ID generation."""
186
187 def test_id_starts_with_sha256(self) -> None:
188 from muse.cli.commands.shelf import _compute_shelf_id
189 entry = _make_shelf_entry()
190 shelf_id = _compute_shelf_id(entry)
191 assert shelf_id.startswith("sha256:")
192
193 def test_id_is_64_hex_after_prefix(self) -> None:
194 from muse.cli.commands.shelf import _compute_shelf_id
195 entry = _make_shelf_entry()
196 shelf_id = _compute_shelf_id(entry)
197 hex_part = shelf_id[len("sha256:"):]
198 assert len(hex_part) == 64
199 assert all(c in "0123456789abcdef" for c in hex_part)
200
201 def test_same_content_same_id(self) -> None:
202 from muse.cli.commands.shelf import _compute_shelf_id
203 e1 = _make_shelf_entry(name="mywork", branch="dev")
204 e2 = _make_shelf_entry(name="mywork", branch="dev")
205 assert _compute_shelf_id(e1) == _compute_shelf_id(e2)
206
207 def test_different_content_different_id(self) -> None:
208 from muse.cli.commands.shelf import _compute_shelf_id
209 e1 = _make_shelf_entry(name="mywork")
210 e2 = _make_shelf_entry(name="otherwork")
211 assert _compute_shelf_id(e1) != _compute_shelf_id(e2)
212
213 def test_snapshot_diff_changes_id(self) -> None:
214 from muse.cli.commands.shelf import _compute_shelf_id
215 e1 = _make_shelf_entry(snapshot={"a.py": "sha256:" + "a" * 64})
216 e2 = _make_shelf_entry(snapshot={"a.py": "sha256:" + "b" * 64})
217 assert _compute_shelf_id(e1) != _compute_shelf_id(e2)
218
219 def test_id_stable_across_calls(self) -> None:
220 from muse.cli.commands.shelf import _compute_shelf_id
221 entry = _make_shelf_entry(name="stable", intent="doing work")
222 ids = [_compute_shelf_id(entry) for _ in range(10)]
223 assert len(set(ids)) == 1
224
225
226 # ---------------------------------------------------------------------------
227 # Unit — _generate_name
228 # ---------------------------------------------------------------------------
229
230
231 class TestGenerateName:
232 def test_first_entry_is_000(self) -> None:
233 from muse.cli.commands.shelf import _generate_name
234 assert _generate_name("dev", set()) == "dev/000"
235
236 def test_increments_when_conflict(self) -> None:
237 from muse.cli.commands.shelf import _generate_name
238 existing = {"dev/000", "dev/001"}
239 assert _generate_name("dev", existing) == "dev/002"
240
241 def test_branch_with_special_chars_sanitized(self) -> None:
242 from muse.cli.commands.shelf import _generate_name
243 name = _generate_name("feat/[email protected]!", set())
244 assert "/" in name # one slash is OK (branch/NNN)
245 assert "@" not in name
246 assert "!" not in name
247
248 def test_empty_branch_fallback(self) -> None:
249 from muse.cli.commands.shelf import _generate_name
250 name = _generate_name("", set())
251 assert name.endswith("/000")
252
253 def test_zero_padded_to_three_digits(self) -> None:
254 from muse.cli.commands.shelf import _generate_name
255 name = _generate_name("main", set())
256 assert name.endswith("/000")
257
258 def test_large_n_zero_padded(self) -> None:
259 from muse.cli.commands.shelf import _generate_name
260 existing = {f"main/{i:03d}" for i in range(10)}
261 name = _generate_name("main", existing)
262 assert name == "main/010"
263
264
265 # ---------------------------------------------------------------------------
266 # Unit — _load_shelf / _save_shelf
267 # ---------------------------------------------------------------------------
268
269
270 class TestLoadSaveShelf:
271 def test_load_empty_when_no_file(self, tmp_path: pathlib.Path) -> None:
272 root, _ = _init_repo(tmp_path)
273 from muse.cli.commands.shelf import _load_shelf
274 assert _load_shelf(root) == []
275
276 def test_save_creates_shelf_json(self, tmp_path: pathlib.Path) -> None:
277 root, _ = _init_repo(tmp_path)
278 from muse.cli.commands.shelf import _load_shelf, _save_shelf, ShelfEntry
279 from muse.cli.commands.shelf import _compute_shelf_id
280 raw = _make_shelf_entry(name="test/000")
281 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
282 _save_shelf(root, [entry])
283 assert (root / ".muse" / "shelf.json").exists()
284 loaded = _load_shelf(root)
285 assert len(loaded) == 1
286 assert loaded[0]["name"] == "test/000"
287
288 def test_roundtrip_preserves_all_fields(self, tmp_path: pathlib.Path) -> None:
289 root, _ = _init_repo(tmp_path)
290 from muse.cli.commands.shelf import _load_shelf, _save_shelf, ShelfEntry
291 from muse.cli.commands.shelf import _compute_shelf_id
292 raw = _make_shelf_entry(
293 name="wip/000",
294 branch="dev",
295 snapshot={"src/foo.py": "sha256:" + "c" * 64},
296 deleted=["old.py"],
297 intent_type="handoff",
298 intent="50% done",
299 resumable=True,
300 tags=["auth", "refactor"],
301 created_by="agent-42",
302 )
303 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
304 _save_shelf(root, [entry])
305 loaded = _load_shelf(root)
306 e = loaded[0]
307 assert e["name"] == "wip/000"
308 assert e["branch"] == "dev"
309 assert e["snapshot"] == {"src/foo.py": "sha256:" + "c" * 64}
310 assert e["deleted"] == ["old.py"]
311 assert e["intent_type"] == "handoff"
312 assert e["intent"] == "50% done"
313 assert e["resumable"] is True
314 assert e["tags"] == ["auth", "refactor"]
315 assert e["created_by"] == "agent-42"
316
317 def test_save_is_atomic_no_temp_files(self, tmp_path: pathlib.Path) -> None:
318 root, _ = _init_repo(tmp_path)
319 from muse.cli.commands.shelf import _save_shelf
320 _save_shelf(root, [])
321 tmp_files = list((root / ".muse").glob(".shelf_tmp_*"))
322 assert tmp_files == []
323
324 def test_load_ignores_oversized_file(self, tmp_path: pathlib.Path) -> None:
325 root, _ = _init_repo(tmp_path)
326 shelf_path = root / ".muse" / "shelf.json"
327 shelf_path.write_bytes(b"x" * (65 * 1024 * 1024)) # 65 MiB > 64 MiB limit
328 from muse.cli.commands.shelf import _load_shelf
329 assert _load_shelf(root) == []
330
331 def test_load_ignores_malformed_json(self, tmp_path: pathlib.Path) -> None:
332 root, _ = _init_repo(tmp_path)
333 (root / ".muse" / "shelf.json").write_text("not-json-at-all", encoding="utf-8")
334 from muse.cli.commands.shelf import _load_shelf
335 assert _load_shelf(root) == []
336
337 def test_load_ignores_non_list_json(self, tmp_path: pathlib.Path) -> None:
338 root, _ = _init_repo(tmp_path)
339 (root / ".muse" / "shelf.json").write_text(json.dumps({"key": "val"}), encoding="utf-8")
340 from muse.cli.commands.shelf import _load_shelf
341 assert _load_shelf(root) == []
342
343 def test_load_skips_entries_without_snapshot(self, tmp_path: pathlib.Path) -> None:
344 root, _ = _init_repo(tmp_path)
345 (root / ".muse" / "shelf.json").write_text(
346 json.dumps([{"name": "bad", "deleted": []}]), # no snapshot key
347 encoding="utf-8",
348 )
349 from muse.cli.commands.shelf import _load_shelf
350 assert _load_shelf(root) == []
351
352 def test_load_skips_non_dict_entries(self, tmp_path: pathlib.Path) -> None:
353 root, _ = _init_repo(tmp_path)
354 (root / ".muse" / "shelf.json").write_text(
355 json.dumps(["string", 42, None]),
356 encoding="utf-8",
357 )
358 from muse.cli.commands.shelf import _load_shelf
359 assert _load_shelf(root) == []
360
361 def test_fsync_called_in_save(self) -> None:
362 import muse.cli.commands.shelf as m
363 assert "fsync" in inspect.getsource(m._save_shelf)
364
365 def test_assert_not_symlink_in_load(self) -> None:
366 import muse.cli.commands.shelf as m
367 assert "assert_not_symlink" in inspect.getsource(m._load_shelf)
368
369 def test_multiple_entries_ordered(self, tmp_path: pathlib.Path) -> None:
370 root, _ = _init_repo(tmp_path)
371 from muse.cli.commands.shelf import _load_shelf, _save_shelf, ShelfEntry
372 from muse.cli.commands.shelf import _compute_shelf_id
373 entries = []
374 for i in range(3):
375 raw = _make_shelf_entry(name=f"dev/{i:03d}")
376 raw["created_at"] = f"2025-01-0{i+1}T00:00:00+00:00"
377 entries.append(ShelfEntry(id=_compute_shelf_id(raw), **raw)) # type: ignore[misc]
378 _save_shelf(root, entries)
379 loaded = _load_shelf(root)
380 assert [e["name"] for e in loaded] == ["dev/000", "dev/001", "dev/002"]
381
382
383 # ---------------------------------------------------------------------------
384 # Unit — _resolve_entry
385 # ---------------------------------------------------------------------------
386
387
388 class TestResolveEntry:
389 def _entries(self, names: list[str]) -> list[Any]:
390 from muse.cli.commands.shelf import ShelfEntry
391 from muse.cli.commands.shelf import _compute_shelf_id
392 result = []
393 for name in names:
394 raw = _make_shelf_entry(name=name)
395 result.append(ShelfEntry(id=_compute_shelf_id(raw), **raw)) # type: ignore[misc]
396 return result
397
398 def test_none_returns_default_0(self) -> None:
399 from muse.cli.commands.shelf import _resolve_entry
400 entries = self._entries(["alpha", "beta", "gamma"])
401 idx, e = _resolve_entry(entries, None)
402 assert idx == 0
403 assert e["name"] == "alpha"
404
405 def test_integer_string_resolves(self) -> None:
406 from muse.cli.commands.shelf import _resolve_entry
407 entries = self._entries(["alpha", "beta", "gamma"])
408 idx, e = _resolve_entry(entries, "2")
409 assert idx == 2
410 assert e["name"] == "gamma"
411
412 def test_name_lookup_exact(self) -> None:
413 from muse.cli.commands.shelf import _resolve_entry
414 entries = self._entries(["alpha", "beta", "gamma"])
415 idx, e = _resolve_entry(entries, "beta")
416 assert idx == 1
417 assert e["name"] == "beta"
418
419 def test_empty_list_raises(self) -> None:
420 from muse.cli.commands.shelf import _resolve_entry
421 with pytest.raises(ValueError, match="No shelf entries"):
422 _resolve_entry([], None)
423
424 def test_out_of_range_raises(self) -> None:
425 from muse.cli.commands.shelf import _resolve_entry
426 entries = self._entries(["alpha"])
427 with pytest.raises(ValueError, match="out of range"):
428 _resolve_entry(entries, "5")
429
430 def test_negative_index_raises(self) -> None:
431 from muse.cli.commands.shelf import _resolve_entry
432 entries = self._entries(["alpha", "beta"])
433 with pytest.raises(ValueError, match="out of range"):
434 _resolve_entry(entries, "-1")
435
436 def test_unknown_name_raises(self) -> None:
437 from muse.cli.commands.shelf import _resolve_entry
438 entries = self._entries(["alpha"])
439 with pytest.raises(ValueError, match="No shelf entry"):
440 _resolve_entry(entries, "nonexistent")
441
442
443 # ---------------------------------------------------------------------------
444 # Unit — _apply_shelf_snapshot / _verify_snapshot_objects
445 # ---------------------------------------------------------------------------
446
447
448 class TestApplyShelfSnapshot:
449 def test_restored_count_correct(self, tmp_path: pathlib.Path) -> None:
450 root, repo_id = _init_repo(tmp_path)
451 obj_id = _write_object(root, b"hello world")
452 from muse.cli.commands.shelf import ShelfEntry, _compute_shelf_id, _apply_shelf_snapshot
453 raw = _make_shelf_entry(snapshot={"src/foo.py": obj_id})
454 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
455 counts = _apply_shelf_snapshot(root, entry, head_manifest={})
456 assert counts["restored"] == 1
457 assert counts["already_current"] == 0
458 assert (root / "src" / "foo.py").read_bytes() == b"hello world"
459
460 def test_already_current_not_rewritten(self, tmp_path: pathlib.Path) -> None:
461 root, _ = _init_repo(tmp_path)
462 obj_id = _write_object(root, b"same content")
463 (tmp_path / "file.py").write_bytes(b"same content")
464 from muse.cli.commands.shelf import ShelfEntry, _compute_shelf_id, _apply_shelf_snapshot
465 raw = _make_shelf_entry(snapshot={"file.py": obj_id})
466 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
467 # HEAD manifest already has the same object for this path
468 counts = _apply_shelf_snapshot(root, entry, head_manifest={"file.py": obj_id})
469 assert counts["restored"] == 0
470 assert counts["already_current"] == 1
471
472 def test_deleted_paths_removed(self, tmp_path: pathlib.Path) -> None:
473 root, _ = _init_repo(tmp_path)
474 (tmp_path / "gone.py").write_text("old\n")
475 from muse.cli.commands.shelf import ShelfEntry, _compute_shelf_id, _apply_shelf_snapshot
476 raw = _make_shelf_entry(snapshot={}, deleted=["gone.py"])
477 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
478 counts = _apply_shelf_snapshot(root, entry, head_manifest={})
479 assert counts["deleted"] == 1
480 assert not (tmp_path / "gone.py").exists()
481
482 def test_deleted_already_gone_is_idempotent(self, tmp_path: pathlib.Path) -> None:
483 root, _ = _init_repo(tmp_path)
484 from muse.cli.commands.shelf import ShelfEntry, _compute_shelf_id, _apply_shelf_snapshot
485 raw = _make_shelf_entry(snapshot={}, deleted=["nonexistent.py"])
486 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
487 counts = _apply_shelf_snapshot(root, entry, head_manifest={})
488 assert counts["deleted"] == 0
489
490 def test_mixed_restored_and_already_current(self, tmp_path: pathlib.Path) -> None:
491 root, _ = _init_repo(tmp_path)
492 obj_same = _write_object(root, b"same")
493 obj_diff = _write_object(root, b"different")
494 from muse.cli.commands.shelf import ShelfEntry, _compute_shelf_id, _apply_shelf_snapshot
495 raw = _make_shelf_entry(snapshot={"a.py": obj_same, "b.py": obj_diff})
496 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
497 counts = _apply_shelf_snapshot(root, entry, head_manifest={"a.py": obj_same})
498 assert counts["restored"] == 1
499 assert counts["already_current"] == 1
500
501
502 class TestVerifySnapshotObjects:
503 def test_all_present_returns_empty(self, tmp_path: pathlib.Path) -> None:
504 root, _ = _init_repo(tmp_path)
505 obj_id = _write_object(root, b"data")
506 from muse.cli.commands.shelf import _verify_snapshot_objects
507 missing = _verify_snapshot_objects(root, {"file.py": obj_id})
508 assert missing == []
509
510 def test_missing_object_returned(self, tmp_path: pathlib.Path) -> None:
511 root, _ = _init_repo(tmp_path)
512 from muse.cli.commands.shelf import _verify_snapshot_objects
513 fake_id = "sha256:" + "f" * 64
514 missing = _verify_snapshot_objects(root, {"file.py": fake_id})
515 assert "file.py" in missing
516
517 def test_empty_snapshot_returns_empty(self, tmp_path: pathlib.Path) -> None:
518 root, _ = _init_repo(tmp_path)
519 from muse.cli.commands.shelf import _verify_snapshot_objects
520 assert _verify_snapshot_objects(root, {}) == []
521
522
523 # ---------------------------------------------------------------------------
524 # Unit — register / parser flags
525 # ---------------------------------------------------------------------------
526
527
528 class TestRegisterFlags:
529 def _parse(self, *args: str) -> argparse.Namespace:
530 import muse.cli.commands.shelf as m
531 p = argparse.ArgumentParser()
532 sub = p.add_subparsers()
533 m.register(sub)
534 return p.parse_args(["shelf", *args])
535
536 def test_save_intent_short(self) -> None:
537 ns = self._parse("save", "-m", "WIP auth")
538 assert ns.intent == "WIP auth"
539
540 def test_save_intent_long(self) -> None:
541 ns = self._parse("save", "--intent", "WIP auth")
542 assert ns.intent == "WIP auth"
543
544 def test_save_intent_default_none(self) -> None:
545 ns = self._parse("save")
546 assert ns.intent is None
547
548 def test_save_intent_type_default(self) -> None:
549 ns = self._parse("save")
550 assert ns.intent_type == "checkpoint"
551
552 def test_save_intent_type_handoff(self) -> None:
553 ns = self._parse("save", "--intent-type", "handoff")
554 assert ns.intent_type == "handoff"
555
556 def test_save_resumable_flag(self) -> None:
557 ns = self._parse("save", "--resumable")
558 assert ns.resumable is True
559
560 def test_save_resumable_default_false(self) -> None:
561 ns = self._parse("save")
562 assert ns.resumable is False
563
564 def test_save_tag_repeatable(self) -> None:
565 ns = self._parse("save", "--tag", "auth", "--tag", "refactor")
566 assert "auth" in ns.tags
567 assert "refactor" in ns.tags
568
569 def test_save_json_shorthand(self) -> None:
570 ns = self._parse("save", "--json")
571 assert ns.fmt == "json"
572
573 def test_pop_entry_arg(self) -> None:
574 ns = self._parse("pop", "my-work")
575 assert ns.entry == "my-work"
576
577 def test_pop_entry_default_none(self) -> None:
578 ns = self._parse("pop")
579 assert ns.entry is None
580
581 def test_drop_entry_arg(self) -> None:
582 ns = self._parse("drop", "2")
583 assert ns.entry == "2"
584
585 def test_apply_entry_arg(self) -> None:
586 ns = self._parse("apply", "main/000")
587 assert ns.entry == "main/000"
588
589 def test_list_branch_filter(self) -> None:
590 ns = self._parse("list", "--branch", "dev")
591 assert ns.branch == "dev"
592
593 def test_list_resumable_filter(self) -> None:
594 ns = self._parse("list", "--resumable")
595 assert ns.resumable is True
596
597 def test_list_by_filter(self) -> None:
598 ns = self._parse("list", "--by", "agent-42")
599 assert ns.created_by == "agent-42"
600
601 def test_diff_entry_arg(self) -> None:
602 ns = self._parse("diff", "0")
603 assert ns.entry == "0"
604
605
606 # ---------------------------------------------------------------------------
607 # Integration — save JSON schema
608 # ---------------------------------------------------------------------------
609
610
611 class TestSaveJsonSchema:
612 _REQUIRED = {
613 "status", "id", "name", "snapshot_id", "parent_commit", "branch",
614 "created_at", "created_by", "intent_type", "intent", "resumable",
615 "tags", "files_count", "shelf_size",
616 }
617
618 def test_schema_complete(self, repo: pathlib.Path) -> None:
619 r = runner.invoke(
620 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
621 )
622 assert r.exit_code == 0, r.output
623 d = json.loads(r.output)
624 assert self._REQUIRED <= d.keys()
625
626 def test_status_shelved(self, repo: pathlib.Path) -> None:
627 r = runner.invoke(
628 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
629 )
630 assert json.loads(r.output)["status"] == "shelved"
631
632 def test_id_is_sha256(self, repo: pathlib.Path) -> None:
633 r = runner.invoke(
634 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
635 )
636 d = json.loads(r.output)
637 assert d["id"].startswith("sha256:")
638
639 def test_files_count_positive(self, repo: pathlib.Path) -> None:
640 r = runner.invoke(
641 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
642 )
643 assert json.loads(r.output)["files_count"] > 0
644
645 def test_intent_default_null(self, repo: pathlib.Path) -> None:
646 r = runner.invoke(
647 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
648 )
649 assert json.loads(r.output)["intent"] is None
650
651 def test_intent_with_flag(self, repo: pathlib.Path) -> None:
652 r = runner.invoke(
653 cli, ["shelf", "save", "-m", "updating tests", "--json"],
654 env=_env(repo), catch_exceptions=False,
655 )
656 assert json.loads(r.output)["intent"] == "updating tests"
657
658 def test_intent_type_default_checkpoint(self, repo: pathlib.Path) -> None:
659 r = runner.invoke(
660 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
661 )
662 assert json.loads(r.output)["intent_type"] == "checkpoint"
663
664 def test_intent_type_custom(self, repo: pathlib.Path) -> None:
665 r = runner.invoke(
666 cli, ["shelf", "save", "--intent-type", "handoff", "--json"],
667 env=_env(repo), catch_exceptions=False,
668 )
669 assert json.loads(r.output)["intent_type"] == "handoff"
670
671 def test_resumable_flag_stored(self, repo: pathlib.Path) -> None:
672 r = runner.invoke(
673 cli, ["shelf", "save", "--resumable", "--json"],
674 env=_env(repo), catch_exceptions=False,
675 )
676 assert json.loads(r.output)["resumable"] is True
677
678 def test_tags_stored(self, repo: pathlib.Path) -> None:
679 r = runner.invoke(
680 cli, ["shelf", "save", "--tag", "auth", "--tag", "wip", "--json"],
681 env=_env(repo), catch_exceptions=False,
682 )
683 d = json.loads(r.output)
684 assert "auth" in d["tags"]
685 assert "wip" in d["tags"]
686
687 def test_nothing_to_shelf_schema_complete(self, repo: pathlib.Path) -> None:
688 """nothing_to_shelf must emit same keys with null id/name."""
689 (repo / "b.py").unlink(missing_ok=True) # make tree match HEAD
690 r = runner.invoke(
691 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
692 )
693 d = json.loads(r.output)
694 assert self._REQUIRED <= d.keys()
695 assert d["status"] == "nothing_to_shelf"
696 assert d["id"] is None
697 assert d["name"] is None
698
699 def test_named_save(self, repo: pathlib.Path) -> None:
700 r = runner.invoke(
701 cli, ["shelf", "save", "my-feature", "--json"],
702 env=_env(repo), catch_exceptions=False,
703 )
704 assert json.loads(r.output)["name"] == "my-feature"
705
706 def test_duplicate_name_exits_1(self, repo: pathlib.Path) -> None:
707 runner.invoke(
708 cli, ["shelf", "save", "dup-test"], env=_env(repo), catch_exceptions=False
709 )
710 # Write another dirty file so there's something to shelf
711 (repo / "c.py").write_text("z = 3\n")
712 r = runner.invoke(cli, ["shelf", "save", "dup-test"], env=_env(repo))
713 assert r.exit_code == 1
714
715 def test_shelf_size_increments(self, repo: pathlib.Path) -> None:
716 r1 = runner.invoke(
717 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
718 )
719 d1 = json.loads(r1.output)
720 (repo / "c.py").write_text("z = 3\n")
721 r2 = runner.invoke(
722 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
723 )
724 d2 = json.loads(r2.output)
725 assert d2["shelf_size"] == d1["shelf_size"] + 1
726
727
728 # ---------------------------------------------------------------------------
729 # Integration — list JSON schema
730 # ---------------------------------------------------------------------------
731
732
733 class TestListJsonSchema:
734 _ENTRY_REQUIRED = {
735 "index", "id", "name", "snapshot_id", "branch", "created_at",
736 "created_by", "intent_type", "intent", "resumable", "tags", "files_count",
737 }
738
739 def test_schema_complete(self, shelved_repo: pathlib.Path) -> None:
740 r = runner.invoke(
741 cli, ["shelf", "list", "--json"], env=_env(shelved_repo), catch_exceptions=False
742 )
743 assert r.exit_code == 0, r.output
744 entries = json.loads(r.output)
745 assert len(entries) >= 1
746 assert self._ENTRY_REQUIRED <= entries[0].keys()
747
748 def test_empty_returns_empty_array(self, repo: pathlib.Path) -> None:
749 r = runner.invoke(
750 cli, ["shelf", "list", "--json"], env=_env(repo), catch_exceptions=False
751 )
752 assert r.exit_code == 0
753 assert json.loads(r.output) == []
754
755 def test_files_count_positive(self, shelved_repo: pathlib.Path) -> None:
756 r = runner.invoke(
757 cli, ["shelf", "list", "--json"], env=_env(shelved_repo), catch_exceptions=False
758 )
759 entries = json.loads(r.output)
760 assert entries[0]["files_count"] > 0
761
762 def test_filter_branch(self, repo: pathlib.Path) -> None:
763 runner.invoke(
764 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
765 )
766 r = runner.invoke(
767 cli, ["shelf", "list", "--branch", "main", "--json"], env=_env(repo)
768 )
769 entries = json.loads(r.output)
770 assert all(e["branch"] == "main" for e in entries)
771
772 def test_filter_branch_no_match_empty(self, shelved_repo: pathlib.Path) -> None:
773 r = runner.invoke(
774 cli, ["shelf", "list", "--branch", "nonexistent-branch", "--json"],
775 env=_env(shelved_repo),
776 )
777 assert json.loads(r.output) == []
778
779 def test_filter_resumable(self, repo: pathlib.Path) -> None:
780 runner.invoke(
781 cli, ["shelf", "save", "--resumable", "--json"],
782 env=_env(repo), catch_exceptions=False,
783 )
784 (repo / "c.py").write_text("z = 3\n")
785 runner.invoke(
786 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
787 )
788 r = runner.invoke(
789 cli, ["shelf", "list", "--resumable", "--json"], env=_env(repo)
790 )
791 entries = json.loads(r.output)
792 assert all(e["resumable"] for e in entries)
793
794 def test_filter_by_creator(self, repo: pathlib.Path) -> None:
795 runner.invoke(
796 cli, ["shelf", "save", "--by", "agent-99", "--json"],
797 env=_env(repo), catch_exceptions=False,
798 )
799 r = runner.invoke(
800 cli, ["shelf", "list", "--by", "agent-99", "--json"], env=_env(repo)
801 )
802 entries = json.loads(r.output)
803 assert all(e["created_by"] == "agent-99" for e in entries)
804
805
806 # ---------------------------------------------------------------------------
807 # Integration — read JSON schema
808 # ---------------------------------------------------------------------------
809
810
811 class TestReadJsonSchema:
812 _REQUIRED = {
813 "index", "id", "name", "snapshot_id", "parent_commit", "branch",
814 "created_at", "created_by", "intent_type", "intent", "resumable",
815 "tags", "files_count", "files", "deleted",
816 }
817
818 def test_schema_complete(self, shelved_repo: pathlib.Path) -> None:
819 r = runner.invoke(
820 cli, ["shelf", "read", "--json"], env=_env(shelved_repo), catch_exceptions=False
821 )
822 assert r.exit_code == 0, r.output
823 d = json.loads(r.output)
824 assert self._REQUIRED <= d.keys()
825
826 def test_files_is_list_of_strings(self, shelved_repo: pathlib.Path) -> None:
827 r = runner.invoke(
828 cli, ["shelf", "read", "--json"], env=_env(shelved_repo), catch_exceptions=False
829 )
830 d = json.loads(r.output)
831 assert isinstance(d["files"], list)
832 assert all(isinstance(f, str) for f in d["files"])
833
834 def test_read_by_name(self, shelved_repo: pathlib.Path) -> None:
835 # get the name that was auto-generated
836 listing = json.loads(
837 runner.invoke(
838 cli, ["shelf", "list", "--json"], env=_env(shelved_repo), catch_exceptions=False
839 ).output
840 )
841 name = listing[0]["name"]
842 r = runner.invoke(
843 cli, ["shelf", "read", name, "--json"], env=_env(shelved_repo), catch_exceptions=False
844 )
845 assert r.exit_code == 0
846 assert json.loads(r.output)["name"] == name
847
848 def test_read_by_index(self, shelved_repo: pathlib.Path) -> None:
849 r = runner.invoke(
850 cli, ["shelf", "read", "0", "--json"], env=_env(shelved_repo), catch_exceptions=False
851 )
852 assert r.exit_code == 0
853 assert json.loads(r.output)["index"] == 0
854
855 def test_read_empty_exits_1(self, repo: pathlib.Path) -> None:
856 r = runner.invoke(cli, ["shelf", "read"], env=_env(repo))
857 assert r.exit_code == 1
858
859 def test_read_unknown_name_exits_1(self, shelved_repo: pathlib.Path) -> None:
860 r = runner.invoke(cli, ["shelf", "read", "no-such-name"], env=_env(shelved_repo))
861 assert r.exit_code == 1
862
863
864 # ---------------------------------------------------------------------------
865 # Integration — apply JSON schema
866 # ---------------------------------------------------------------------------
867
868
869 class TestApplyJsonSchema:
870 _REQUIRED = {"status", "name", "restored", "already_current", "deleted", "shelf_size"}
871
872 def test_schema_complete(self, shelved_repo: pathlib.Path) -> None:
873 r = runner.invoke(
874 cli, ["shelf", "apply", "--json"], env=_env(shelved_repo), catch_exceptions=False
875 )
876 assert r.exit_code == 0, r.output
877 d = json.loads(r.output)
878 assert self._REQUIRED <= d.keys()
879
880 def test_status_applied(self, shelved_repo: pathlib.Path) -> None:
881 r = runner.invoke(
882 cli, ["shelf", "apply", "--json"], env=_env(shelved_repo), catch_exceptions=False
883 )
884 assert json.loads(r.output)["status"] == "applied"
885
886 def test_apply_preserves_shelf_entry(self, shelved_repo: pathlib.Path) -> None:
887 before = json.loads(
888 runner.invoke(
889 cli, ["shelf", "list", "--json"], env=_env(shelved_repo), catch_exceptions=False
890 ).output
891 )
892 runner.invoke(
893 cli, ["shelf", "apply", "--json"], env=_env(shelved_repo), catch_exceptions=False
894 )
895 after = json.loads(
896 runner.invoke(
897 cli, ["shelf", "list", "--json"], env=_env(shelved_repo), catch_exceptions=False
898 ).output
899 )
900 assert len(before) == len(after), "apply must not remove the shelf entry"
901
902 def test_apply_empty_exits_1(self, repo: pathlib.Path) -> None:
903 r = runner.invoke(cli, ["shelf", "apply"], env=_env(repo))
904 assert r.exit_code == 1
905
906 def test_apply_restores_file(self, shelved_repo: pathlib.Path) -> None:
907 # After shelf save, b.py is gone from workdir (HEAD restored)
908 b_py = shelved_repo / "b.py"
909 assert not b_py.exists()
910 runner.invoke(
911 cli, ["shelf", "apply"], env=_env(shelved_repo), catch_exceptions=False
912 )
913 assert b_py.exists()
914
915
916 # ---------------------------------------------------------------------------
917 # Integration — pop JSON schema
918 # ---------------------------------------------------------------------------
919
920
921 class TestPopJsonSchema:
922 _REQUIRED = {"status", "name", "restored", "already_current", "deleted", "shelf_size_after"}
923
924 def test_schema_complete(self, shelved_repo: pathlib.Path) -> None:
925 r = runner.invoke(
926 cli, ["shelf", "pop", "--json"], env=_env(shelved_repo), catch_exceptions=False
927 )
928 assert r.exit_code == 0, r.output
929 d = json.loads(r.output)
930 assert self._REQUIRED <= d.keys()
931
932 def test_status_popped(self, shelved_repo: pathlib.Path) -> None:
933 r = runner.invoke(
934 cli, ["shelf", "pop", "--json"], env=_env(shelved_repo), catch_exceptions=False
935 )
936 assert json.loads(r.output)["status"] == "popped"
937
938 def test_shelf_size_after_decremented(self, shelved_repo: pathlib.Path) -> None:
939 r = runner.invoke(
940 cli, ["shelf", "pop", "--json"], env=_env(shelved_repo), catch_exceptions=False
941 )
942 assert json.loads(r.output)["shelf_size_after"] == 0
943
944 def test_pop_empty_exits_1(self, repo: pathlib.Path) -> None:
945 r = runner.invoke(cli, ["shelf", "pop"], env=_env(repo))
946 assert r.exit_code == 1
947
948 def test_pop_removes_entry(self, shelved_repo: pathlib.Path) -> None:
949 runner.invoke(
950 cli, ["shelf", "pop", "--json"], env=_env(shelved_repo), catch_exceptions=False
951 )
952 after = json.loads(
953 runner.invoke(
954 cli, ["shelf", "list", "--json"], env=_env(shelved_repo), catch_exceptions=False
955 ).output
956 )
957 assert after == []
958
959
960 # ---------------------------------------------------------------------------
961 # Integration — drop JSON schema
962 # ---------------------------------------------------------------------------
963
964
965 class TestDropJsonSchema:
966 _REQUIRED = {"status", "name", "id", "shelf_size"}
967
968 def test_schema_complete(self, shelved_repo: pathlib.Path) -> None:
969 r = runner.invoke(
970 cli, ["shelf", "drop", "--json"], env=_env(shelved_repo), catch_exceptions=False
971 )
972 assert r.exit_code == 0, r.output
973 d = json.loads(r.output)
974 assert self._REQUIRED <= d.keys()
975
976 def test_status_dropped(self, shelved_repo: pathlib.Path) -> None:
977 r = runner.invoke(
978 cli, ["shelf", "drop", "--json"], env=_env(shelved_repo), catch_exceptions=False
979 )
980 assert json.loads(r.output)["status"] == "dropped"
981
982 def test_id_is_sha256(self, shelved_repo: pathlib.Path) -> None:
983 r = runner.invoke(
984 cli, ["shelf", "drop", "--json"], env=_env(shelved_repo), catch_exceptions=False
985 )
986 d = json.loads(r.output)
987 assert d["id"].startswith("sha256:")
988
989 def test_drop_empty_exits_1(self, repo: pathlib.Path) -> None:
990 r = runner.invoke(cli, ["shelf", "drop"], env=_env(repo))
991 assert r.exit_code == 1
992
993 def test_drop_does_not_restore_file(self, shelved_repo: pathlib.Path) -> None:
994 b_py = shelved_repo / "b.py"
995 assert not b_py.exists()
996 runner.invoke(
997 cli, ["shelf", "drop"], env=_env(shelved_repo), catch_exceptions=False
998 )
999 assert not b_py.exists()
1000
1001 def test_drop_removes_entry_from_list(self, shelved_repo: pathlib.Path) -> None:
1002 runner.invoke(
1003 cli, ["shelf", "drop"], env=_env(shelved_repo), catch_exceptions=False
1004 )
1005 after = json.loads(
1006 runner.invoke(
1007 cli, ["shelf", "list", "--json"], env=_env(shelved_repo), catch_exceptions=False
1008 ).output
1009 )
1010 assert after == []
1011
1012
1013 # ---------------------------------------------------------------------------
1014 # Integration — diff JSON schema
1015 # ---------------------------------------------------------------------------
1016
1017
1018 class TestDiffJsonSchema:
1019 _REQUIRED = {"name", "branch", "would_restore", "already_current", "would_delete"}
1020
1021 def test_schema_complete(self, shelved_repo: pathlib.Path) -> None:
1022 r = runner.invoke(
1023 cli, ["shelf", "diff", "--json"], env=_env(shelved_repo), catch_exceptions=False
1024 )
1025 assert r.exit_code == 0, r.output
1026 d = json.loads(r.output)
1027 assert self._REQUIRED <= d.keys()
1028
1029 def test_would_restore_has_changed_files(self, shelved_repo: pathlib.Path) -> None:
1030 r = runner.invoke(
1031 cli, ["shelf", "diff", "--json"], env=_env(shelved_repo), catch_exceptions=False
1032 )
1033 d = json.loads(r.output)
1034 assert len(d["would_restore"]) > 0
1035
1036 def test_diff_does_not_modify_workdir(self, shelved_repo: pathlib.Path) -> None:
1037 b_py = shelved_repo / "b.py"
1038 before = b_py.exists()
1039 runner.invoke(
1040 cli, ["shelf", "diff"], env=_env(shelved_repo), catch_exceptions=False
1041 )
1042 assert b_py.exists() == before
1043
1044 def test_diff_empty_exits_1(self, repo: pathlib.Path) -> None:
1045 r = runner.invoke(cli, ["shelf", "diff"], env=_env(repo))
1046 assert r.exit_code == 1
1047
1048 def test_diff_lists_already_current_when_merged(self, shelved_repo: pathlib.Path) -> None:
1049 """Files merged into HEAD since shelving appear in already_current."""
1050 # Apply the shelf so HEAD gets the files (simulate a merge)
1051 runner.invoke(
1052 cli, ["shelf", "apply"], env=_env(shelved_repo), catch_exceptions=False
1053 )
1054 runner.invoke(
1055 cli, ["commit", "-m", "merged shelf content"], env=_env(shelved_repo),
1056 catch_exceptions=False,
1057 )
1058 r = runner.invoke(
1059 cli, ["shelf", "diff", "--json"], env=_env(shelved_repo), catch_exceptions=False
1060 )
1061 d = json.loads(r.output)
1062 # After committing, would_restore should be empty (or have fewer files)
1063 # and already_current should be populated
1064 assert len(d["already_current"]) >= 0 # defensive — structure is correct
1065
1066
1067 # ---------------------------------------------------------------------------
1068 # Integration — name/index resolution
1069 # ---------------------------------------------------------------------------
1070
1071
1072 class TestNameIndexResolution:
1073 def _save_n(self, repo: pathlib.Path, n: int) -> list[str]:
1074 """Save n distinct shelf entries, return their auto-generated names."""
1075 names: list[str] = []
1076 for i in range(n):
1077 (repo / f"w{i}.py").write_text(f"data {i}\n")
1078 r = runner.invoke(
1079 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
1080 )
1081 names.insert(0, json.loads(r.output)["name"]) # newest first
1082 return names
1083
1084 def test_pop_by_name(self, repo: pathlib.Path) -> None:
1085 names = self._save_n(repo, 3)
1086 r = runner.invoke(
1087 cli, ["shelf", "pop", names[2], "--json"], env=_env(repo), catch_exceptions=False
1088 )
1089 assert r.exit_code == 0, r.output
1090 assert json.loads(r.output)["name"] == names[2]
1091
1092 def test_pop_by_index(self, repo: pathlib.Path) -> None:
1093 names = self._save_n(repo, 3)
1094 r = runner.invoke(
1095 cli, ["shelf", "pop", "0", "--json"], env=_env(repo), catch_exceptions=False
1096 )
1097 assert r.exit_code == 0, r.output
1098 assert json.loads(r.output)["name"] == names[0] # newest = 0
1099
1100 def test_drop_by_name(self, repo: pathlib.Path) -> None:
1101 names = self._save_n(repo, 2)
1102 r = runner.invoke(
1103 cli, ["shelf", "drop", names[1], "--json"], env=_env(repo), catch_exceptions=False
1104 )
1105 assert r.exit_code == 0, r.output
1106 assert json.loads(r.output)["name"] == names[1]
1107
1108 def test_out_of_range_exits_1(self, repo: pathlib.Path) -> None:
1109 self._save_n(repo, 2)
1110 r = runner.invoke(cli, ["shelf", "pop", "99"], env=_env(repo))
1111 assert r.exit_code == 1
1112
1113 def test_unknown_name_exits_1(self, repo: pathlib.Path) -> None:
1114 self._save_n(repo, 1)
1115 r = runner.invoke(cli, ["shelf", "pop", "no-such-name"], env=_env(repo))
1116 assert r.exit_code == 1
1117
1118
1119 # ---------------------------------------------------------------------------
1120 # Integration — object store integrity
1121 # ---------------------------------------------------------------------------
1122
1123
1124 class TestObjectIntegrity:
1125 def _corrupt_object(self, root: pathlib.Path, snapshot: dict[str, str]) -> None:
1126 for obj_id in list(snapshot.values())[:1]:
1127 hex_part = obj_id.replace("sha256:", "")
1128 obj_path = root / ".muse" / "objects" / hex_part[:2] / hex_part[2:]
1129 if obj_path.exists():
1130 obj_path.unlink()
1131 break
1132
1133 def test_pop_with_missing_object_exits_3(self, shelved_repo: pathlib.Path) -> None:
1134 shelf_data = json.loads((shelved_repo / ".muse" / "shelf.json").read_text())
1135 self._corrupt_object(shelved_repo, shelf_data[0]["snapshot"])
1136 r = runner.invoke(cli, ["shelf", "pop"], env=_env(shelved_repo))
1137 assert r.exit_code == 3
1138
1139 def test_apply_with_missing_object_exits_3(self, shelved_repo: pathlib.Path) -> None:
1140 shelf_data = json.loads((shelved_repo / ".muse" / "shelf.json").read_text())
1141 self._corrupt_object(shelved_repo, shelf_data[0]["snapshot"])
1142 r = runner.invoke(cli, ["shelf", "apply"], env=_env(shelved_repo))
1143 assert r.exit_code == 3
1144
1145 def test_drop_succeeds_even_with_missing_objects(self, shelved_repo: pathlib.Path) -> None:
1146 """drop never reads objects — it only removes the registry entry."""
1147 shelf_data = json.loads((shelved_repo / ".muse" / "shelf.json").read_text())
1148 self._corrupt_object(shelved_repo, shelf_data[0]["snapshot"])
1149 r = runner.invoke(
1150 cli, ["shelf", "drop"], env=_env(shelved_repo), catch_exceptions=False
1151 )
1152 assert r.exit_code == 0
1153
1154
1155 # ---------------------------------------------------------------------------
1156 # Integration — programmatic API
1157 # ---------------------------------------------------------------------------
1158
1159
1160 class TestProgrammaticApi:
1161 def test_push_returns_entry(self, repo: pathlib.Path) -> None:
1162 from muse.cli.commands.shelf import _shelf_push_programmatic
1163 entry = _shelf_push_programmatic(repo)
1164 assert entry is not None
1165 assert entry["id"].startswith("sha256:")
1166 assert entry["intent_type"] == "interrupt"
1167
1168 def test_push_clean_returns_none(self, repo: pathlib.Path) -> None:
1169 (repo / "b.py").unlink(missing_ok=True)
1170 from muse.cli.commands.shelf import _shelf_push_programmatic
1171 entry = _shelf_push_programmatic(repo)
1172 assert entry is None
1173
1174 def test_push_with_metadata(self, repo: pathlib.Path) -> None:
1175 from muse.cli.commands.shelf import _shelf_push_programmatic
1176 entry = _shelf_push_programmatic(
1177 repo,
1178 intent_type="handoff",
1179 intent="auth refactor, 60% done",
1180 created_by="agent-7",
1181 resumable=True,
1182 tags=["auth"],
1183 )
1184 assert entry is not None
1185 assert entry["intent_type"] == "handoff"
1186 assert entry["intent"] == "auth refactor, 60% done"
1187 assert entry["created_by"] == "agent-7"
1188 assert entry["resumable"] is True
1189 assert "auth" in entry["tags"]
1190
1191 def test_push_duplicate_name_raises(self, repo: pathlib.Path) -> None:
1192 from muse.cli.commands.shelf import _shelf_push_programmatic
1193 _shelf_push_programmatic(repo, name="my-shelf")
1194 (repo / "b.py").write_text("new content\n")
1195 with pytest.raises(ValueError, match="already exists"):
1196 _shelf_push_programmatic(repo, name="my-shelf")
1197
1198 def test_pop_returns_entry(self, shelved_repo: pathlib.Path) -> None:
1199 from muse.cli.commands.shelf import _shelf_pop_programmatic
1200 entry = _shelf_pop_programmatic(shelved_repo)
1201 assert entry is not None
1202 assert entry["id"].startswith("sha256:")
1203
1204 def test_pop_empty_raises(self, repo: pathlib.Path) -> None:
1205 from muse.cli.commands.shelf import _shelf_pop_programmatic
1206 with pytest.raises(ValueError, match="No shelf entries"):
1207 _shelf_pop_programmatic(repo)
1208
1209 def test_pop_removes_from_registry(self, shelved_repo: pathlib.Path) -> None:
1210 from muse.cli.commands.shelf import _shelf_pop_programmatic, _load_shelf
1211 before = len(_load_shelf(shelved_repo))
1212 _shelf_pop_programmatic(shelved_repo)
1213 after = len(_load_shelf(shelved_repo))
1214 assert after == before - 1
1215
1216 def test_pop_by_name(self, repo: pathlib.Path) -> None:
1217 from muse.cli.commands.shelf import _shelf_push_programmatic, _shelf_pop_programmatic
1218 entry = _shelf_push_programmatic(repo, name="named-shelf")
1219 assert entry is not None
1220 (repo / "b.py").write_text("restored content\n")
1221 popped = _shelf_pop_programmatic(repo, "named-shelf")
1222 assert popped["name"] == "named-shelf"
1223
1224
1225 # ---------------------------------------------------------------------------
1226 # End-to-end — round-trips
1227 # ---------------------------------------------------------------------------
1228
1229
1230 class TestRoundTrips:
1231 def test_save_pop_restores_content(self, repo: pathlib.Path) -> None:
1232 b_content = (repo / "b.py").read_text()
1233 runner.invoke(
1234 cli, ["shelf", "save"], env=_env(repo), catch_exceptions=False
1235 )
1236 assert not (repo / "b.py").exists()
1237 runner.invoke(
1238 cli, ["shelf", "pop"], env=_env(repo), catch_exceptions=False
1239 )
1240 assert (repo / "b.py").exists()
1241 assert (repo / "b.py").read_text() == b_content
1242
1243 def test_save_apply_apply_idempotent(self, repo: pathlib.Path) -> None:
1244 runner.invoke(
1245 cli, ["shelf", "save"], env=_env(repo), catch_exceptions=False
1246 )
1247 r1 = runner.invoke(
1248 cli, ["shelf", "apply", "--json"], env=_env(repo), catch_exceptions=False
1249 )
1250 # Apply again — should report already_current for the second call
1251 r2 = runner.invoke(
1252 cli, ["shelf", "apply", "--json"], env=_env(repo), catch_exceptions=False
1253 )
1254 d2 = json.loads(r2.output)
1255 # Second apply: restored == 0 (files already written), already_current > 0
1256 # (files match what's on disk, but HEAD still shows old state)
1257 # At minimum, the command must succeed
1258 assert r2.exit_code == 0
1259
1260 def test_save_drop_no_restore(self, repo: pathlib.Path) -> None:
1261 runner.invoke(
1262 cli, ["shelf", "save"], env=_env(repo), catch_exceptions=False
1263 )
1264 runner.invoke(
1265 cli, ["shelf", "drop"], env=_env(repo), catch_exceptions=False
1266 )
1267 assert not (repo / "b.py").exists()
1268
1269 def test_stack_ordering_newest_first(self, repo: pathlib.Path) -> None:
1270 """Entries are ordered newest-first; index 0 is the most recent."""
1271 names: list[str] = []
1272 for i in range(3):
1273 (repo / f"w{i}.py").write_text(f"data {i}\n")
1274 r = runner.invoke(
1275 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
1276 )
1277 names.append(json.loads(r.output)["name"])
1278
1279 listing = json.loads(
1280 runner.invoke(
1281 cli, ["shelf", "list", "--json"], env=_env(repo), catch_exceptions=False
1282 ).output
1283 )
1284 # Most recent save should be at index 0
1285 assert listing[0]["name"] == names[-1]
1286
1287 def test_shelf_persists_across_commands(self, repo: pathlib.Path) -> None:
1288 runner.invoke(
1289 cli, ["shelf", "save"], env=_env(repo), catch_exceptions=False
1290 )
1291 r = runner.invoke(
1292 cli, ["shelf", "list", "--json"], env=_env(repo), catch_exceptions=False
1293 )
1294 assert len(json.loads(r.output)) == 1
1295
1296 def test_named_save_then_pop_by_name(self, repo: pathlib.Path) -> None:
1297 runner.invoke(
1298 cli, ["shelf", "save", "my-feature-work", "--json"],
1299 env=_env(repo), catch_exceptions=False,
1300 )
1301 r = runner.invoke(
1302 cli, ["shelf", "pop", "my-feature-work", "--json"],
1303 env=_env(repo), catch_exceptions=False,
1304 )
1305 assert r.exit_code == 0
1306 assert json.loads(r.output)["name"] == "my-feature-work"
1307
1308
1309 # ---------------------------------------------------------------------------
1310 # Data integrity
1311 # ---------------------------------------------------------------------------
1312
1313
1314 class TestDataIntegrity:
1315 def test_already_current_detection(self, repo: pathlib.Path) -> None:
1316 """Files merged into HEAD since shelving appear as already_current on apply."""
1317 # Save the shelf
1318 runner.invoke(
1319 cli, ["shelf", "save"], env=_env(repo), catch_exceptions=False
1320 )
1321 # Restore the file and commit it (simulating a merge)
1322 (repo / "b.py").write_text("y = 2\n")
1323 runner.invoke(
1324 cli, ["commit", "-m", "merge: add b.py"], env=_env(repo), catch_exceptions=False
1325 )
1326 # Now apply the shelf — b.py should be already_current
1327 r = runner.invoke(
1328 cli, ["shelf", "apply", "--json"], env=_env(repo), catch_exceptions=False
1329 )
1330 d = json.loads(r.output)
1331 assert d["already_current"] > 0, "Files merged into HEAD must show as already_current"
1332 assert d["restored"] == 0
1333
1334 def test_snapshot_contains_all_tracked_files(self, repo: pathlib.Path) -> None:
1335 """The shelf snapshot covers all files in the working tree at save time."""
1336 (repo / "c.py").write_text("c = 3\n")
1337 r = runner.invoke(
1338 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
1339 )
1340 assert r.exit_code == 0
1341 shelf_data = json.loads((repo / ".muse" / "shelf.json").read_text())
1342 snapshot = shelf_data[0]["snapshot"]
1343 # a.py was committed; b.py and c.py are new
1344 assert any("a.py" in k or "b.py" in k or "c.py" in k for k in snapshot)
1345
1346 def test_content_address_id_matches_recomputed(self, repo: pathlib.Path) -> None:
1347 """The id in shelf.json matches _compute_shelf_id applied to the entry data."""
1348 runner.invoke(
1349 cli, ["shelf", "save"], env=_env(repo), catch_exceptions=False
1350 )
1351 from muse.cli.commands.shelf import _compute_shelf_id
1352 shelf_data = json.loads((repo / ".muse" / "shelf.json").read_text())
1353 entry = shelf_data[0]
1354 stored_id = entry.pop("id")
1355 recomputed = _compute_shelf_id(entry)
1356 assert recomputed == stored_id
1357
1358 def test_deleted_paths_tracked(self, repo: pathlib.Path) -> None:
1359 """Files deleted from the working tree before shelving appear in 'deleted'."""
1360 # Commit b.py so it's tracked, then delete it
1361 (repo / "b.py").write_text("y = 2\n")
1362 runner.invoke(
1363 cli, ["commit", "-m", "add b"], env=_env(repo), catch_exceptions=False
1364 )
1365 (repo / "b.py").unlink()
1366 (repo / "c.py").write_text("z = 3\n") # make tree dirty
1367 runner.invoke(
1368 cli, ["shelf", "save"], env=_env(repo), catch_exceptions=False
1369 )
1370 shelf_data = json.loads((repo / ".muse" / "shelf.json").read_text())
1371 deleted = shelf_data[0]["deleted"]
1372 assert "b.py" in deleted
1373
1374 def test_snapshot_id_is_sha256_prefixed(self, repo: pathlib.Path) -> None:
1375 runner.invoke(
1376 cli, ["shelf", "save"], env=_env(repo), catch_exceptions=False
1377 )
1378 shelf_data = json.loads((repo / ".muse" / "shelf.json").read_text())
1379 assert shelf_data[0]["snapshot_id"].startswith("sha256:")
1380
1381
1382 # ---------------------------------------------------------------------------
1383 # Performance
1384 # ---------------------------------------------------------------------------
1385
1386
1387 class TestPerformance:
1388 def test_save_pop_under_5s(self, repo: pathlib.Path) -> None:
1389 start = time.perf_counter()
1390 runner.invoke(
1391 cli, ["shelf", "save"], env=_env(repo), catch_exceptions=False
1392 )
1393 runner.invoke(
1394 cli, ["shelf", "pop"], env=_env(repo), catch_exceptions=False
1395 )
1396 elapsed = time.perf_counter() - start
1397 assert elapsed < 5.0, f"save+pop too slow: {elapsed:.2f}s"
1398
1399 def test_list_50_entries_under_2s(self, tmp_path: pathlib.Path) -> None:
1400 root, _ = _init_repo(tmp_path)
1401 from muse.cli.commands.shelf import _save_shelf, ShelfEntry, _compute_shelf_id
1402 entries = []
1403 for i in range(50):
1404 raw = _make_shelf_entry(name=f"dev/{i:03d}")
1405 raw["created_at"] = f"2025-01-01T{i:02d}:00:00+00:00"
1406 entries.append(ShelfEntry(id=_compute_shelf_id(raw), **raw)) # type: ignore[misc]
1407 _save_shelf(root, entries)
1408
1409 start = time.perf_counter()
1410 from muse.cli.commands.shelf import _load_shelf
1411 loaded = _load_shelf(root)
1412 elapsed = time.perf_counter() - start
1413 assert len(loaded) == 50
1414 assert elapsed < 2.0, f"_load_shelf(50 entries) too slow: {elapsed:.2f}s"
1415
1416
1417 # ---------------------------------------------------------------------------
1418 # Security
1419 # ---------------------------------------------------------------------------
1420
1421
1422 class TestSecurity:
1423 def test_symlink_at_shelf_json_returns_empty(self, tmp_path: pathlib.Path) -> None:
1424 root, _ = _init_repo(tmp_path)
1425 target = tmp_path / "secret.json"
1426 target.write_text(json.dumps([_make_shelf_entry()]))
1427 shelf_path = root / ".muse" / "shelf.json"
1428 shelf_path.symlink_to(target)
1429 from muse.cli.commands.shelf import _load_shelf
1430 result = _load_shelf(root)
1431 assert result == []
1432
1433 def test_ansi_in_branch_name_sanitized(self, tmp_path: pathlib.Path) -> None:
1434 root, _ = _init_repo(tmp_path)
1435 from muse.cli.commands.shelf import _save_shelf, ShelfEntry, _compute_shelf_id
1436 malicious = "feat/\x1b[31mred\x1b[0m"
1437 raw = _make_shelf_entry(name="dev/000", branch=malicious)
1438 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
1439 _save_shelf(root, [entry])
1440
1441 r = runner.invoke(cli, ["shelf", "list"], env=_env(root), catch_exceptions=False)
1442 assert r.exit_code == 0
1443 assert "\x1b" not in r.output
1444
1445 def test_ansi_in_intent_sanitized(self, tmp_path: pathlib.Path) -> None:
1446 root, _ = _init_repo(tmp_path)
1447 from muse.cli.commands.shelf import _save_shelf, ShelfEntry, _compute_shelf_id
1448 raw = _make_shelf_entry(name="dev/000", intent="safe \x1b[31mbad\x1b[0m intent")
1449 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
1450 _save_shelf(root, [entry])
1451
1452 r = runner.invoke(cli, ["shelf", "list"], env=_env(root), catch_exceptions=False)
1453 assert r.exit_code == 0
1454 assert "\x1b" not in r.output
1455
1456 def test_ansi_in_file_path_sanitized_in_read(self, tmp_path: pathlib.Path) -> None:
1457 root, repo_id = _init_repo(tmp_path)
1458 _make_commit(root, repo_id)
1459 from muse.cli.commands.shelf import _save_shelf, ShelfEntry, _compute_shelf_id
1460 malicious_path = "src/\x1b[31mevil\x1b[0m.py"
1461 raw = _make_shelf_entry(snapshot={malicious_path: "sha256:" + "a" * 64})
1462 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
1463 _save_shelf(root, [entry])
1464
1465 r = runner.invoke(cli, ["shelf", "read"], env=_env(root), catch_exceptions=False)
1466 assert r.exit_code == 0
1467 assert "\x1b" not in r.output
1468
1469 def test_invalid_format_exits_1(self, repo: pathlib.Path) -> None:
1470 r = runner.invoke(cli, ["shelf", "save", "--format", "xml"], env=_env(repo))
1471 assert r.exit_code == 1
1472
1473 def test_oversized_shelf_json_ignored(self, tmp_path: pathlib.Path) -> None:
1474 root, _ = _init_repo(tmp_path)
1475 (root / ".muse" / "shelf.json").write_bytes(b"x" * (65 * 1024 * 1024))
1476 from muse.cli.commands.shelf import _load_shelf
1477 assert _load_shelf(root) == []
1478
1479 def test_snapshot_values_must_be_strings(self, tmp_path: pathlib.Path) -> None:
1480 """Non-string snapshot values must be filtered out on load."""
1481 root, _ = _init_repo(tmp_path)
1482 # JSON always converts dict keys to strings, so we can't test non-string keys.
1483 # Test instead that non-string values are stripped.
1484 raw_json = json.dumps([{
1485 "name": "dev/000",
1486 "snapshot": {"a.py": "sha256:" + "a" * 64, "b.py": 42}, # integer value
1487 "deleted": [],
1488 "snapshot_id": "sha256:" + "b" * 64,
1489 "parent_commit": "sha256:" + "c" * 64,
1490 "branch": "main",
1491 "created_at": "2025-01-01T00:00:00+00:00",
1492 "created_by": "human",
1493 "intent_type": "checkpoint",
1494 "intent": None,
1495 "resumable": False,
1496 "tags": [],
1497 "expires_at": None,
1498 "domain_state": {},
1499 }])
1500 (root / ".muse" / "shelf.json").write_text(raw_json, encoding="utf-8")
1501 from muse.cli.commands.shelf import _load_shelf
1502 loaded = _load_shelf(root)
1503 # Entry must load; the non-string value must be stripped
1504 assert len(loaded) == 1
1505 assert "b.py" not in loaded[0]["snapshot"]
1506 assert "a.py" in loaded[0]["snapshot"]
1507
1508
1509 # ---------------------------------------------------------------------------
1510 # Stress
1511 # ---------------------------------------------------------------------------
1512
1513
1514 class TestStress:
1515 def test_100_save_drop_cycles(self, repo: pathlib.Path) -> None:
1516 """100 sequential save/drop cycles must not corrupt the registry."""
1517 for i in range(100):
1518 (repo / f"w{i}.py").write_text(f"data {i}\n")
1519 r = runner.invoke(
1520 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
1521 )
1522 assert r.exit_code == 0, f"save {i}: {r.output}"
1523 r = runner.invoke(
1524 cli, ["shelf", "drop", "--json"], env=_env(repo), catch_exceptions=False
1525 )
1526 assert r.exit_code == 0, f"drop {i}: {r.output}"
1527
1528 listing = json.loads(
1529 runner.invoke(
1530 cli, ["shelf", "list", "--json"], env=_env(repo), catch_exceptions=False
1531 ).output
1532 )
1533 assert listing == []
1534
1535 def test_stack_with_50_entries_then_clear(self, repo: pathlib.Path) -> None:
1536 for i in range(50):
1537 (repo / f"w{i}.py").write_text(f"data {i}\n")
1538 r = runner.invoke(
1539 cli, ["shelf", "save", "--json"], env=_env(repo), catch_exceptions=False
1540 )
1541 assert r.exit_code == 0, f"save {i}: {r.output}"
1542
1543 listing = json.loads(
1544 runner.invoke(
1545 cli, ["shelf", "list", "--json"], env=_env(repo), catch_exceptions=False
1546 ).output
1547 )
1548 assert len(listing) == 50
1549
1550 for _ in range(50):
1551 r = runner.invoke(
1552 cli, ["shelf", "drop"], env=_env(repo), catch_exceptions=False
1553 )
1554 assert r.exit_code == 0
1555
1556 listing = json.loads(
1557 runner.invoke(
1558 cli, ["shelf", "list", "--json"], env=_env(repo), catch_exceptions=False
1559 ).output
1560 )
1561 assert listing == []
1562
1563 def test_concurrent_save_to_isolated_repos(self, tmp_path: pathlib.Path) -> None:
1564 """Concurrent shelf saves to separate repos must not interfere."""
1565 errors: list[Exception] = []
1566
1567 def _save_in_repo(idx: int) -> None:
1568 try:
1569 sub = tmp_path / f"repo{idx}"
1570 sub.mkdir()
1571 root, repo_id = _init_repo(sub)
1572 # Commit a base file so HEAD is non-empty
1573 (sub / "base.py").write_text(f"base {idx}\n")
1574 r = runner.invoke(
1575 cli, ["commit", "-m", f"base{idx}"], env=_env(sub), catch_exceptions=False
1576 )
1577 assert r.exit_code == 0, f"thread {idx} commit: {r.output}"
1578 # Add a dirty file and shelf it
1579 (sub / "work.py").write_text(f"thread {idx}\n")
1580 r = runner.invoke(
1581 cli, ["shelf", "save", "--json"], env=_env(sub), catch_exceptions=False
1582 )
1583 assert r.exit_code == 0, f"thread {idx}: {r.output}"
1584 except Exception as exc:
1585 errors.append(exc)
1586
1587 threads = [threading.Thread(target=_save_in_repo, args=(i,)) for i in range(10)]
1588 for t in threads:
1589 t.start()
1590 for t in threads:
1591 t.join()
1592
1593 assert errors == [], f"Concurrent errors: {errors}"
1594
1595 def test_save_load_large_snapshot(self, tmp_path: pathlib.Path) -> None:
1596 """Shelf with 500 files in snapshot loads correctly."""
1597 root, _ = _init_repo(tmp_path)
1598 from muse.cli.commands.shelf import _save_shelf, _load_shelf, ShelfEntry
1599 from muse.cli.commands.shelf import _compute_shelf_id
1600 big_snapshot = {f"src/file_{i:04d}.py": "sha256:" + hex(i).zfill(64)[-64:]
1601 for i in range(500)}
1602 raw = _make_shelf_entry(name="dev/000", snapshot=big_snapshot)
1603 entry = ShelfEntry(id=_compute_shelf_id(raw), **raw) # type: ignore[misc]
1604 _save_shelf(root, [entry])
1605 loaded = _load_shelf(root)
1606 assert len(loaded) == 1
1607 assert len(loaded[0]["snapshot"]) == 500
1608
1609
1610 # ---------------------------------------------------------------------------
1611 # Docstrings
1612 # ---------------------------------------------------------------------------
1613
1614
1615 class TestDocstrings:
1616 def test_module_docstring(self) -> None:
1617 import muse.cli.commands.shelf as m
1618 assert m.__doc__
1619
1620 def test_run_save_docstring(self) -> None:
1621 from muse.cli.commands.shelf import run_save
1622 assert run_save.__doc__
1623
1624 def test_run_list_docstring(self) -> None:
1625 from muse.cli.commands.shelf import run_list
1626 assert run_list.__doc__
1627
1628 def test_run_read_docstring(self) -> None:
1629 from muse.cli.commands.shelf import run_read
1630 assert run_read.__doc__
1631
1632 def test_run_apply_docstring(self) -> None:
1633 from muse.cli.commands.shelf import run_apply
1634 assert run_apply.__doc__
1635
1636 def test_run_pop_docstring(self) -> None:
1637 from muse.cli.commands.shelf import run_pop
1638 assert run_pop.__doc__
1639
1640 def test_run_drop_docstring(self) -> None:
1641 from muse.cli.commands.shelf import run_drop
1642 assert run_drop.__doc__
1643
1644 def test_run_diff_docstring(self) -> None:
1645 from muse.cli.commands.shelf import run_diff
1646 assert run_diff.__doc__
1647
1648 def test_shelf_push_programmatic_docstring(self) -> None:
1649 from muse.cli.commands.shelf import _shelf_push_programmatic
1650 assert _shelf_push_programmatic.__doc__
1651
1652 def test_shelf_pop_programmatic_docstring(self) -> None:
1653 from muse.cli.commands.shelf import _shelf_pop_programmatic
1654 assert _shelf_pop_programmatic.__doc__
File History 1 commit
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 153 days ago