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