gabriel / muse public
test_core_coverage_gaps.py python
502 lines 19.7 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Tests targeting coverage gaps in core modules: object_store, repo, store, merge_engine."""
2
3 import hashlib
4 import json
5 import os
6 import pathlib
7
8 import pytest
9
10
11 def _sha256(data: bytes) -> str:
12 return hashlib.sha256(data).hexdigest()
13
14 from muse.core.object_store import (
15 has_object,
16 object_path,
17 objects_dir,
18 read_object,
19 restore_object,
20 write_object,
21 write_object_from_path,
22 )
23 from muse.core.repo import find_repo_root, require_repo
24 from muse.core.store import (
25 CommitRecord,
26 SnapshotRecord,
27 get_commits_for_branch,
28 get_head_commit_id,
29 get_head_snapshot_id,
30 get_head_snapshot_manifest,
31 get_tags_for_commit,
32 read_commit,
33 read_snapshot,
34 resolve_commit_ref,
35 update_commit_metadata,
36 write_commit,
37 write_snapshot,
38 )
39 from muse.core.merge_engine import apply_resolution, clear_merge_state, read_merge_state, write_merge_state
40 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
41 from muse.core._types import Manifest
42
43 import datetime
44
45
46 # ---------------------------------------------------------------------------
47 # object_store
48 # ---------------------------------------------------------------------------
49
50
51 class TestObjectStore:
52 def test_objects_dir_path(self, tmp_path: pathlib.Path) -> None:
53 d = objects_dir(tmp_path)
54 assert d == tmp_path / ".muse" / "objects"
55
56 def test_object_path_sharding(self, tmp_path: pathlib.Path) -> None:
57 oid = "ab" + "c" * 62
58 p = object_path(tmp_path, oid)
59 assert p.parent.name == "ab"
60 assert p.name == "c" * 62
61
62 def test_has_object_false_when_absent(self, tmp_path: pathlib.Path) -> None:
63 assert not has_object(tmp_path, "a" * 64)
64
65 def test_has_object_true_after_write(self, tmp_path: pathlib.Path) -> None:
66 content = b"hello"
67 oid = _sha256(content)
68 write_object(tmp_path, oid, content)
69 assert has_object(tmp_path, oid)
70
71 def test_write_object_idempotent_returns_false(self, tmp_path: pathlib.Path) -> None:
72 content = b"first"
73 oid = _sha256(content)
74 assert write_object(tmp_path, oid, content) is True
75 # Second write with correct hash but same ID — idempotent
76 assert write_object(tmp_path, oid, content) is False
77 # content should not change
78 assert read_object(tmp_path, oid) == content
79
80 def test_write_object_from_path_idempotent(self, tmp_path: pathlib.Path) -> None:
81 content = b"content"
82 src = tmp_path / "src.bin"
83 src.write_bytes(content)
84 oid = _sha256(content)
85 assert write_object_from_path(tmp_path, oid, src) is True
86 assert write_object_from_path(tmp_path, oid, src) is False
87
88 def test_write_object_from_path_stores_content(self, tmp_path: pathlib.Path) -> None:
89 content = b"my bytes"
90 src = tmp_path / "file.bin"
91 src.write_bytes(content)
92 oid = _sha256(content)
93 write_object_from_path(tmp_path, oid, src)
94 assert read_object(tmp_path, oid) == content
95
96 def test_read_object_returns_none_when_absent(self, tmp_path: pathlib.Path) -> None:
97 assert read_object(tmp_path, "e" * 64) is None
98
99 def test_read_object_returns_bytes(self, tmp_path: pathlib.Path) -> None:
100 content = b"data"
101 oid = _sha256(content)
102 write_object(tmp_path, oid, content)
103 assert read_object(tmp_path, oid) == content
104
105 def test_restore_object_returns_false_when_absent(self, tmp_path: pathlib.Path) -> None:
106 dest = tmp_path / "out.bin"
107 result = restore_object(tmp_path, "0" * 64, dest)
108 assert result is False
109 assert not dest.exists()
110
111 def test_restore_object_creates_dest(self, tmp_path: pathlib.Path) -> None:
112 content = b"restored"
113 oid = _sha256(content)
114 write_object(tmp_path, oid, content)
115 dest = tmp_path / "sub" / "out.bin"
116 result = restore_object(tmp_path, oid, dest)
117 assert result is True
118 assert dest.read_bytes() == content
119
120 def test_restore_object_creates_parent_dirs(self, tmp_path: pathlib.Path) -> None:
121 content = b"nested"
122 oid = _sha256(content)
123 write_object(tmp_path, oid, content)
124 dest = tmp_path / "a" / "b" / "c" / "file.bin"
125 restore_object(tmp_path, oid, dest)
126 assert dest.exists()
127
128
129 class TestRestoreObjectIdempotency:
130 """restore_object must preserve the destination inode when content matches.
131
132 The ``os.replace`` rename syscall always produces a new inode. Editors
133 (Cursor, VS Code, Vim, …) use inode-based filesystem-event watchers; a
134 spurious rename blinds them to subsequent changes, leaving permanently stale
135 buffers. The fix: hash-check dest before writing — if bytes already match
136 the requested object_id, return without touching the file.
137
138 These tests are the regression gate for that fix. They prove:
139
140 1. When dest already has the correct content the inode is preserved.
141 2. When dest has *different* content the file is replaced (inode changes).
142 3. When dest does not yet exist the write proceeds normally.
143 4. A checkout-style simulation: many files, only changed ones get new inodes.
144 """
145
146 def test_inode_preserved_when_content_matches(self, tmp_path: pathlib.Path) -> None:
147 """Core regression: restore_object must NOT rename when content is correct."""
148 content = b"editor-watching-this-file"
149 oid = _sha256(content)
150 write_object(tmp_path, oid, content)
151
152 dest = tmp_path / "file.txt"
153 dest.write_bytes(content)
154 inode_before = dest.stat().st_ino
155
156 result = restore_object(tmp_path, oid, dest)
157
158 assert result is True
159 assert dest.read_bytes() == content
160 # The inode must not change — a rename would produce a new inode and
161 # blind any editor that was watching the original file descriptor.
162 assert dest.stat().st_ino == inode_before, (
163 "restore_object issued a spurious rename even though dest already "
164 "contained the correct content — this blinds inode-watching editors"
165 )
166
167 def test_mtime_preserved_when_content_matches(self, tmp_path: pathlib.Path) -> None:
168 """mtime stability: no spurious write means no mtime bump."""
169 content = b"stable-mtime-check"
170 oid = _sha256(content)
171 write_object(tmp_path, oid, content)
172
173 dest = tmp_path / "file.txt"
174 dest.write_bytes(content)
175 mtime_ns_before = dest.stat().st_mtime_ns
176
177 restore_object(tmp_path, oid, dest)
178
179 assert dest.stat().st_mtime_ns == mtime_ns_before, (
180 "restore_object bumped mtime even though content was already correct"
181 )
182
183 def test_inode_changes_when_content_differs(self, tmp_path: pathlib.Path) -> None:
184 """When dest has wrong content the file must be replaced."""
185 correct = b"correct-content"
186 wrong = b"wrong-content-different-bytes"
187 oid = _sha256(correct)
188 write_object(tmp_path, oid, correct)
189
190 dest = tmp_path / "file.txt"
191 dest.write_bytes(wrong)
192 inode_before = dest.stat().st_ino
193
194 result = restore_object(tmp_path, oid, dest)
195
196 assert result is True
197 assert dest.read_bytes() == correct
198 # Content changed — a rename is expected; the inode must be different.
199 assert dest.stat().st_ino != inode_before
200
201 def test_idempotent_on_fresh_file(self, tmp_path: pathlib.Path) -> None:
202 """When dest does not yet exist the write proceeds normally."""
203 content = b"brand-new-file"
204 oid = _sha256(content)
205 write_object(tmp_path, oid, content)
206
207 dest = tmp_path / "new.txt"
208 assert not dest.exists()
209
210 result = restore_object(tmp_path, oid, dest)
211
212 assert result is True
213 assert dest.read_bytes() == content
214
215 def test_second_restore_is_truly_noop(self, tmp_path: pathlib.Path) -> None:
216 """Calling restore_object twice leaves the file and inode unchanged."""
217 content = b"idempotent-restore"
218 oid = _sha256(content)
219 write_object(tmp_path, oid, content)
220
221 dest = tmp_path / "file.txt"
222 restore_object(tmp_path, oid, dest) # first call — writes the file
223 inode_first = dest.stat().st_ino
224 mtime_first = dest.stat().st_mtime_ns
225
226 restore_object(tmp_path, oid, dest) # second call — must be a no-op
227
228 assert dest.stat().st_ino == inode_first
229 assert dest.stat().st_mtime_ns == mtime_first
230 assert dest.read_bytes() == content
231
232 def test_checkout_simulation_only_changed_files_renamed(
233 self, tmp_path: pathlib.Path
234 ) -> None:
235 """Simulate a branch checkout: only files that changed get new inodes.
236
237 This is the end-to-end scenario that caused the Cursor stale-buffer bug:
238 a ``muse checkout`` that touches N files would rename ALL of them even
239 when most were identical on both branches. After the fix, only the
240 genuinely changed file gets a new inode.
241 """
242 unchanged_content = b"I am the same on both branches"
243 changed_old = b"old branch content"
244 changed_new = b"new branch content"
245
246 unchanged_oid = _sha256(unchanged_content)
247 changed_oid = _sha256(changed_new)
248
249 write_object(tmp_path, unchanged_oid, unchanged_content)
250 write_object(tmp_path, changed_oid, changed_new)
251
252 unchanged_dest = tmp_path / "unchanged.py"
253 changed_dest = tmp_path / "changed.py"
254
255 # Simulate working tree before checkout
256 unchanged_dest.write_bytes(unchanged_content)
257 changed_dest.write_bytes(changed_old)
258
259 inode_unchanged_before = unchanged_dest.stat().st_ino
260 inode_changed_before = changed_dest.stat().st_ino
261
262 # Simulate _checkout_snapshot restoring both files
263 restore_object(tmp_path, unchanged_oid, unchanged_dest)
264 restore_object(tmp_path, changed_oid, changed_dest)
265
266 # unchanged file: inode must be preserved (no rename)
267 assert unchanged_dest.stat().st_ino == inode_unchanged_before, (
268 "unchanged file got a new inode — editor watching it would go blind"
269 )
270 # changed file: inode should differ (content replaced)
271 assert changed_dest.stat().st_ino != inode_changed_before
272 assert changed_dest.read_bytes() == changed_new
273
274
275 # ---------------------------------------------------------------------------
276 # repo
277 # ---------------------------------------------------------------------------
278
279
280 class TestFindRepoRoot:
281 def test_finds_muse_dir_in_cwd(self, tmp_path: pathlib.Path) -> None:
282 (tmp_path / ".muse").mkdir()
283 result = find_repo_root(tmp_path)
284 assert result == tmp_path
285
286 def test_finds_muse_dir_in_parent(self, tmp_path: pathlib.Path) -> None:
287 (tmp_path / ".muse").mkdir()
288 subdir = tmp_path / "a" / "b"
289 subdir.mkdir(parents=True)
290 result = find_repo_root(subdir)
291 assert result == tmp_path
292
293 def test_returns_none_when_no_repo(self, tmp_path: pathlib.Path) -> None:
294 result = find_repo_root(tmp_path)
295 assert result is None
296
297 def test_env_override_returns_path(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
298 (tmp_path / ".muse").mkdir()
299 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
300 result = find_repo_root()
301 assert result == tmp_path
302
303 def test_env_override_returns_none_when_not_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
304 # tmp_path exists but has no .muse/
305 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
306 result = find_repo_root()
307 assert result is None
308
309 def test_require_repo_exits_when_no_repo(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
310 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
311 monkeypatch.chdir(tmp_path)
312 with pytest.raises(SystemExit):
313 require_repo()
314
315
316 # ---------------------------------------------------------------------------
317 # store coverage gaps
318 # ---------------------------------------------------------------------------
319
320
321 class TestStoreGaps:
322 def _make_repo(self, tmp_path: pathlib.Path) -> pathlib.Path:
323 muse = tmp_path / ".muse"
324 for d in ("commits", "snapshots", "objects", "refs/heads"):
325 (muse / d).mkdir(parents=True)
326 (muse / "HEAD").write_text("ref: refs/heads/main\n")
327 (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo"}))
328 (muse / "refs" / "heads" / "main").write_text("")
329 return tmp_path
330
331 def test_get_head_commit_id_empty_branch(self, tmp_path: pathlib.Path) -> None:
332 root = self._make_repo(tmp_path)
333 assert get_head_commit_id(root, "main") is None
334
335 def test_get_head_snapshot_id_no_commits(self, tmp_path: pathlib.Path) -> None:
336 root = self._make_repo(tmp_path)
337 assert get_head_snapshot_id(root, "test-repo", "main") is None
338
339 def test_get_head_snapshot_manifest_no_commits(self, tmp_path: pathlib.Path) -> None:
340 root = self._make_repo(tmp_path)
341 assert get_head_snapshot_manifest(root, "test-repo", "main") is None
342
343 def test_get_commits_for_branch_empty(self, tmp_path: pathlib.Path) -> None:
344 root = self._make_repo(tmp_path)
345 commits = get_commits_for_branch(root, "test-repo", "main")
346 assert commits == []
347
348 def _seed_chain(self, root: pathlib.Path, n: int) -> list[str]:
349 """Write a linear chain of *n* commits on ``main`` and return their IDs (newest first)."""
350 ids: list[str] = []
351 parent_id: str | None = None
352 manifest: Manifest = {}
353 snap_id = compute_snapshot_id(manifest)
354 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
355 for i in range(n):
356 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) + datetime.timedelta(hours=i)
357 message = f"commit {i}"
358 parent_ids = [parent_id] if parent_id else []
359 commit_id = compute_commit_id(parent_ids, snap_id, message, committed_at.isoformat())
360 commit = CommitRecord(
361 commit_id=commit_id,
362 repo_id="test-repo",
363 branch="main",
364 snapshot_id=snap_id,
365 message=message,
366 committed_at=committed_at,
367 parent_commit_id=parent_id,
368 )
369 write_commit(root, commit)
370 ids.append(commit_id)
371 parent_id = commit_id
372 # HEAD points at the last (newest) commit
373 (root / ".muse" / "refs" / "heads" / "main").write_text(ids[-1])
374 ids.reverse() # newest first, matching get_commits_for_branch order
375 return ids
376
377 def test_get_commits_for_branch_max_count_stops_early(
378 self, tmp_path: pathlib.Path
379 ) -> None:
380 """max_count caps the walk — only that many commits are returned."""
381 root = self._make_repo(tmp_path)
382 all_ids = self._seed_chain(root, 5)
383
384 result = get_commits_for_branch(root, "test-repo", "main", max_count=2)
385 assert len(result) == 2
386 assert result[0].commit_id == all_ids[0]
387 assert result[1].commit_id == all_ids[1]
388
389 def test_get_commits_for_branch_max_count_zero_returns_all(
390 self, tmp_path: pathlib.Path
391 ) -> None:
392 """max_count=0 (the default) returns the full chain."""
393 root = self._make_repo(tmp_path)
394 all_ids = self._seed_chain(root, 5)
395
396 result = get_commits_for_branch(root, "test-repo", "main", max_count=0)
397 assert len(result) == 5
398 assert [c.commit_id for c in result] == all_ids
399
400 def test_get_commits_for_branch_max_count_larger_than_chain(
401 self, tmp_path: pathlib.Path
402 ) -> None:
403 """max_count larger than the chain length returns every commit without error."""
404 root = self._make_repo(tmp_path)
405 all_ids = self._seed_chain(root, 3)
406
407 result = get_commits_for_branch(root, "test-repo", "main", max_count=100)
408 assert len(result) == 3
409 assert [c.commit_id for c in result] == all_ids
410
411 def test_resolve_commit_ref_with_none_returns_head(self, tmp_path: pathlib.Path) -> None:
412 root = self._make_repo(tmp_path)
413 manifest: Manifest = {"a.mid": "h" * 64}
414 snap_id = compute_snapshot_id(manifest)
415 snap = SnapshotRecord(snapshot_id=snap_id, manifest=manifest)
416 write_snapshot(root, snap)
417 committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
418 commit_id = compute_commit_id([], snap_id, "test", committed_at.isoformat())
419 commit = CommitRecord(
420 commit_id=commit_id,
421 repo_id="test-repo",
422 branch="main",
423 snapshot_id=snap_id,
424 message="test",
425 committed_at=committed_at,
426 )
427 write_commit(root, commit)
428 (root / ".muse" / "refs" / "heads" / "main").write_text(commit_id)
429
430 result = resolve_commit_ref(root, "test-repo", "main", None)
431 assert result is not None
432 assert result.commit_id == commit_id
433
434 def test_read_commit_returns_none_for_unknown(self, tmp_path: pathlib.Path) -> None:
435 root = self._make_repo(tmp_path)
436 assert read_commit(root, "unknown") is None
437
438 def test_read_snapshot_returns_none_for_unknown(self, tmp_path: pathlib.Path) -> None:
439 root = self._make_repo(tmp_path)
440 assert read_snapshot(root, "unknown") is None
441
442 def test_update_commit_metadata_false_for_unknown(self, tmp_path: pathlib.Path) -> None:
443 root = self._make_repo(tmp_path)
444 assert update_commit_metadata(root, "unknown", "key", "val") is False
445
446 def test_get_tags_for_commit_empty(self, tmp_path: pathlib.Path) -> None:
447 root = self._make_repo(tmp_path)
448 tags = get_tags_for_commit(root, "test-repo", "c" * 64)
449 assert tags == []
450
451
452 # ---------------------------------------------------------------------------
453 # merge_engine coverage gaps
454 # ---------------------------------------------------------------------------
455
456
457 class TestMergeEngineCoverageGaps:
458 def _make_repo(self, tmp_path: pathlib.Path) -> pathlib.Path:
459 muse = tmp_path / ".muse"
460 muse.mkdir(parents=True)
461 return tmp_path
462
463 def test_clear_merge_state_no_file(self, tmp_path: pathlib.Path) -> None:
464 root = self._make_repo(tmp_path)
465 # Should not raise even if MERGE_STATE.json is absent
466 clear_merge_state(root)
467
468 def test_apply_resolution_copies_object(self, tmp_path: pathlib.Path) -> None:
469 root = self._make_repo(tmp_path)
470 # Write a real object to the store — oid must be the SHA-256 of the content.
471 content = b"resolved content"
472 oid = _sha256(content)
473 write_object(root, oid, content)
474
475 apply_resolution(root, "track.mid", oid)
476 dest = root / "track.mid"
477 assert dest.exists()
478 assert dest.read_bytes() == b"resolved content"
479
480 def test_apply_resolution_raises_when_object_absent(self, tmp_path: pathlib.Path) -> None:
481 root = self._make_repo(tmp_path)
482 with pytest.raises(FileNotFoundError):
483 apply_resolution(root, "track.mid", "0" * 64)
484
485 def test_read_merge_state_invalid_json_returns_none(self, tmp_path: pathlib.Path) -> None:
486 root = self._make_repo(tmp_path)
487 (root / ".muse" / "MERGE_STATE.json").write_text("not json {{")
488 result = read_merge_state(root)
489 assert result is None
490
491 def test_write_then_clear_merge_state(self, tmp_path: pathlib.Path) -> None:
492 root = self._make_repo(tmp_path)
493 write_merge_state(
494 root,
495 base_commit="b" * 64,
496 ours_commit="o" * 64,
497 theirs_commit="t" * 64,
498 conflict_paths=["a.mid"],
499 )
500 assert (root / ".muse" / "MERGE_STATE.json").exists()
501 clear_merge_state(root)
502 assert not (root / ".muse" / "MERGE_STATE.json").exists()
File History 5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago