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