gabriel / muse public
test_directories_feature.py python
1,166 lines 47.0 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 120 days ago
1 """Comprehensive tests for the "directories as first-class objects" feature.
2
3 Covers every changed surface:
4 - directories_from_manifest (unit)
5 - walk_workdir_with_dirs (unit + integration)
6 - compute_snapshot_id with directories parameter (unit)
7 - detect_directory_renames (unit + property-style)
8 - diff_workdir_vs_snapshot 6-tuple (unit + integration)
9 - SnapshotRecord.directories serialisation round-trip (unit)
10 - write_snapshot / read_snapshot with directories (integration)
11 - CodePlugin.diff directory rename detection (integration)
12 - delta_summary directory rename counting (unit)
13 - replay_one propagates directories to new SnapshotRecord (integration)
14 - Full commit → branch → rename → commit → merge E2E workflow (e2e)
15 - Stress / performance (stress)
16 - Security: path traversal, symlinks, adversarial inputs (security)
17 """
18
19 from __future__ import annotations
20
21 import datetime
22 import hashlib
23 import json
24 import os
25 import pathlib
26 import subprocess
27 import sys
28 import time
29 import pytest
30
31 from muse.core.snapshot import (
32 compute_commit_id,
33 compute_snapshot_id,
34 detect_directory_renames,
35 diff_workdir_vs_snapshot,
36 directories_from_manifest,
37 hash_file,
38 walk_workdir_with_dirs,
39 )
40 from muse.core.store import (
41 CommitRecord,
42 SnapshotRecord,
43 read_commit,
44 read_snapshot,
45 write_commit,
46 write_snapshot,
47 )
48 from muse.domain import DirectoryRenameOp, SnapshotManifest
49 from muse.core.types import Manifest, MsgpackDict, blob_id, fake_id, now_utc_iso, split_id
50 from muse.plugins.code.plugin import CodePlugin
51 from muse.core.paths import ref_path, muse_dir
52
53
54 # ---------------------------------------------------------------------------
55 # Shared helpers
56 # ---------------------------------------------------------------------------
57
58 _REPO_ID = "test-repo-dirs"
59 _counter = 0
60
61
62
63
64 def _init_store(root: pathlib.Path) -> None:
65 dot_muse = muse_dir(root)
66 for d in ("commits", "snapshots", "objects", "refs/heads"):
67 (dot_muse / d).mkdir(parents=True, exist_ok=True)
68 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
69 (dot_muse / "repo.json").write_text(
70 json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8"
71 )
72
73
74 def _make_snap(root: pathlib.Path, manifest: Manifest, dirs: list[str] | None = None) -> SnapshotRecord:
75 dirs = dirs if dirs is not None else directories_from_manifest(manifest)
76 sid = compute_snapshot_id(manifest, dirs)
77 rec = SnapshotRecord(snapshot_id=sid, manifest=manifest, directories=dirs)
78 write_snapshot(root, rec)
79 return rec
80
81
82 def _make_commit_rec(
83 root: pathlib.Path,
84 snap: SnapshotRecord,
85 branch: str = "main",
86 parent_id: str | None = None,
87 message: str = "test commit",
88 ) -> CommitRecord:
89 global _counter
90 _counter += 1
91 committed_at = datetime.datetime.now(datetime.timezone.utc)
92 cid = compute_commit_id(
93 [parent_id] if parent_id else [],
94 snap.snapshot_id,
95 message,
96 committed_at.isoformat(),
97 )
98 rec = CommitRecord(
99 repo_id=_REPO_ID,
100 commit_id=cid,
101 branch=branch,
102 snapshot_id=snap.snapshot_id,
103 message=message,
104 committed_at=committed_at,
105 parent_commit_id=parent_id,
106 )
107 write_commit(root, rec)
108 (ref_path(root, branch)).write_text(cid, encoding="utf-8")
109 return rec
110
111
112 @pytest.fixture()
113 def store(tmp_path: pathlib.Path) -> pathlib.Path:
114 _init_store(tmp_path)
115 return tmp_path
116
117
118 @pytest.fixture()
119 def workdir(tmp_path: pathlib.Path) -> pathlib.Path:
120 return tmp_path
121
122
123 # ===========================================================================
124 # 1. directories_from_manifest — unit
125 # ===========================================================================
126
127 class TestDirectoriesFromManifest:
128 def test_empty_manifest_returns_empty(self) -> None:
129 assert directories_from_manifest({}) == []
130
131 def test_flat_files_no_dirs(self) -> None:
132 result = directories_from_manifest({"a.py": "h1", "b.py": "h2"})
133 assert result == []
134
135 def test_single_nested_file(self) -> None:
136 result = directories_from_manifest({"src/main.py": "h1"})
137 assert result == ["src"]
138
139 def test_deeply_nested(self) -> None:
140 result = directories_from_manifest({"a/b/c/d.py": "h1"})
141 assert result == ["a", "a/b", "a/b/c"]
142
143 def test_multiple_files_same_dir_deduped(self) -> None:
144 result = directories_from_manifest({"src/a.py": "h1", "src/b.py": "h2"})
145 assert result == ["src"]
146
147 def test_sibling_dirs(self) -> None:
148 result = directories_from_manifest({
149 "src/foo.py": "h1",
150 "tests/bar.py": "h2",
151 })
152 assert result == ["src", "tests"]
153
154 def test_mixed_flat_and_nested(self) -> None:
155 result = directories_from_manifest({
156 "root.py": "h0",
157 "src/main.py": "h1",
158 "src/lib/util.py": "h2",
159 })
160 assert result == ["src", "src/lib"]
161
162 def test_result_is_sorted(self) -> None:
163 result = directories_from_manifest({
164 "z/file.py": "h1",
165 "a/file.py": "h2",
166 "m/sub/file.py": "h3",
167 })
168 assert result == sorted(result)
169
170 def test_result_is_deduplicated(self) -> None:
171 result = directories_from_manifest({
172 "pkg/a.py": "h1",
173 "pkg/b.py": "h2",
174 "pkg/c.py": "h3",
175 })
176 assert result.count("pkg") == 1
177
178 def test_large_flat_tree_no_dirs(self) -> None:
179 manifest = {f"file_{i}.txt": f"hash{i}" for i in range(200)}
180 assert directories_from_manifest(manifest) == []
181
182 def test_preserves_posix_separators(self) -> None:
183 result = directories_from_manifest({"foo/bar/baz.py": "h"})
184 assert all("/" in d or d == "foo" for d in result)
185 assert "\\" not in "".join(result)
186
187
188 # ===========================================================================
189 # 2. compute_snapshot_id with directories — unit
190 # ===========================================================================
191
192 class TestComputeSnapshotIdWithDirectories:
193 def test_no_dirs_matches_legacy_behaviour(self) -> None:
194 m = {"a.py": fake_id("h1")}
195 assert compute_snapshot_id(m) == compute_snapshot_id(m, None)
196 assert compute_snapshot_id(m) == compute_snapshot_id(m, [])
197
198 def test_dirs_change_the_id(self) -> None:
199 m = {"a.py": fake_id("h1")}
200 without = compute_snapshot_id(m, [])
201 with_dir = compute_snapshot_id(m, ["src"])
202 assert without != with_dir
203
204 def test_different_dirs_different_id(self) -> None:
205 m = {"a.py": fake_id("h1")}
206 id1 = compute_snapshot_id(m, ["src"])
207 id2 = compute_snapshot_id(m, ["lib"])
208 assert id1 != id2
209
210 def test_same_files_same_dirs_deterministic(self) -> None:
211 m = {"a/b.py": fake_id("h1"), "c/d.py": fake_id("h2")}
212 dirs = ["a", "c"]
213 assert compute_snapshot_id(m, dirs) == compute_snapshot_id(m, dirs)
214
215 def test_dir_order_independent(self) -> None:
216 m = {"a.py": fake_id("h1")}
217 id1 = compute_snapshot_id(m, ["src", "lib"])
218 id2 = compute_snapshot_id(m, ["lib", "src"])
219 assert id1 == id2
220
221 def test_file_rename_changes_id_even_with_same_dirs(self) -> None:
222 dirs = ["src"]
223 id1 = compute_snapshot_id({"src/a.py": fake_id("h1")}, dirs)
224 id2 = compute_snapshot_id({"src/b.py": fake_id("h1")}, dirs)
225 assert id1 != id2
226
227 def test_dir_rename_changes_id_same_file_content(self) -> None:
228 manifest = {"f.py": fake_id("h1")}
229 id_old = compute_snapshot_id(manifest, ["old_name"])
230 id_new = compute_snapshot_id(manifest, ["new_name"])
231 assert id_old != id_new
232
233 def test_result_is_64_hex_chars(self) -> None:
234 sid = compute_snapshot_id({"a.py": fake_id("h")}, ["src"])
235 assert len(sid) == 71
236 assert all(c in "0123456789abcdef" for c in split_id(sid)[1])
237
238
239 # ===========================================================================
240 # 3. detect_directory_renames — unit
241 # ===========================================================================
242
243 class TestDetectDirectoryRenames:
244 def test_clean_single_rename(self) -> None:
245 last = {"old/a.py": "h1", "old/b.py": "h2"}
246 current = {"new/a.py": "h1", "new/b.py": "h2"}
247 renames = detect_directory_renames({"old"}, {"new"}, last, current)
248 assert renames == [("old", "new")]
249
250 def test_no_rename_content_changed(self) -> None:
251 last = {"old/a.py": "h1"}
252 current = {"new/a.py": "DIFFERENT"}
253 renames = detect_directory_renames({"old"}, {"new"}, last, current)
254 assert renames == []
255
256 def test_no_rename_empty_old_dir(self) -> None:
257 # old dir has no files in last_manifest → can't match
258 last: Manifest = {}
259 current = {"new/a.py": "h1"}
260 renames = detect_directory_renames({"old"}, {"new"}, last, current)
261 assert renames == []
262
263 def test_multiple_independent_renames(self) -> None:
264 last = {"foo/x.py": "h1", "bar/y.py": "h2"}
265 current = {"baz/x.py": "h1", "qux/y.py": "h2"}
266 renames = detect_directory_renames({"foo", "bar"}, {"baz", "qux"}, last, current)
267 assert set(renames) == {("foo", "baz"), ("bar", "qux")}
268
269 def test_ambiguous_candidates_not_renamed(self) -> None:
270 # Two added dirs have identical file sets → ambiguous, none matched
271 last = {"old/f.py": "h1"}
272 current = {"new1/f.py": "h1", "new2/f.py": "h1"}
273 renames = detect_directory_renames({"old"}, {"new1", "new2"}, last, current)
274 # Should match exactly one (first sorted candidate wins)
275 assert len(renames) == 1
276
277 def test_partial_match_not_renamed(self) -> None:
278 last = {"old/a.py": "h1", "old/b.py": "h2"}
279 current = {"new/a.py": "h1"} # b.py missing
280 renames = detect_directory_renames({"old"}, {"new"}, last, current)
281 assert renames == []
282
283 def test_extra_file_in_new_dir_not_renamed(self) -> None:
284 last = {"old/a.py": "h1"}
285 current = {"new/a.py": "h1", "new/extra.py": "h2"}
286 renames = detect_directory_renames({"old"}, {"new"}, last, current)
287 assert renames == []
288
289 def test_returns_list_of_tuples(self) -> None:
290 last = {"src/main.py": "abc"}
291 current = {"lib/main.py": "abc"}
292 result = detect_directory_renames({"src"}, {"lib"}, last, current)
293 assert isinstance(result, list)
294 assert all(isinstance(r, tuple) and len(r) == 2 for r in result)
295
296 def test_empty_sets_returns_empty(self) -> None:
297 assert detect_directory_renames(set(), set(), {}, {}) == []
298
299 def test_single_file_dir_rename(self) -> None:
300 last = {"pkg/module.py": "cafebabe"}
301 current = {"renamed_pkg/module.py": "cafebabe"}
302 renames = detect_directory_renames({"pkg"}, {"renamed_pkg"}, last, current)
303 assert renames == [("pkg", "renamed_pkg")]
304
305
306 # ===========================================================================
307 # 4. diff_workdir_vs_snapshot — 6-tuple (fix broken existing + new)
308 # ===========================================================================
309
310 class TestDiffWorkdirVsSnapshot6Tuple:
311 def test_returns_6_tuple(self, workdir: pathlib.Path) -> None:
312 result = diff_workdir_vs_snapshot(workdir, {})
313 assert len(result) == 6
314
315 def test_untracked_first_commit(self, workdir: pathlib.Path) -> None:
316 (workdir / "f.py").write_bytes(b"x")
317 added, modified, deleted, untracked, added_dirs, deleted_dirs = \
318 diff_workdir_vs_snapshot(workdir, {})
319 assert added == set()
320 assert "f.py" in untracked
321
322 def test_added_file_detected(self, workdir: pathlib.Path) -> None:
323 (workdir / "f.py").write_bytes(b"x")
324 added, modified, deleted, untracked, added_dirs, deleted_dirs = \
325 diff_workdir_vs_snapshot(workdir, {"other.py": "abc"})
326 assert "f.py" in added
327 assert "other.py" in deleted
328
329 def test_modified_file_detected(self, workdir: pathlib.Path) -> None:
330 f = workdir / "f.py"
331 f.write_bytes(b"new content")
332 added, modified, deleted, untracked, added_dirs, deleted_dirs = \
333 diff_workdir_vs_snapshot(workdir, {"f.py": "oldhash"})
334 assert "f.py" in modified
335
336 def test_clean_workdir_all_empty(self, workdir: pathlib.Path) -> None:
337 f = workdir / "f.py"
338 f.write_bytes(b"content")
339 h = hash_file(f)
340 added, modified, deleted, untracked, added_dirs, deleted_dirs = \
341 diff_workdir_vs_snapshot(workdir, {"f.py": h})
342 assert not added and not modified and not deleted and not untracked
343
344 def test_added_dir_detected(self, workdir: pathlib.Path) -> None:
345 (workdir / "src").mkdir()
346 (workdir / "src" / "main.py").write_bytes(b"x")
347 added, modified, deleted, untracked, added_dirs, deleted_dirs = \
348 diff_workdir_vs_snapshot(workdir, {"root.py": "abc"}, last_directories=["lib"])
349 assert "src" in added_dirs
350 assert "lib" in deleted_dirs
351
352 def test_deleted_dir_detected(self, workdir: pathlib.Path) -> None:
353 (workdir / "f.py").write_bytes(b"x")
354 h = hash_file(workdir / "f.py")
355 added, modified, deleted, untracked, added_dirs, deleted_dirs = \
356 diff_workdir_vs_snapshot(workdir, {"f.py": h}, last_directories=["old_dir"])
357 assert "old_dir" in deleted_dirs
358
359 def test_unchanged_dirs_not_in_delta(self, workdir: pathlib.Path) -> None:
360 (workdir / "src").mkdir()
361 (workdir / "src" / "main.py").write_bytes(b"x")
362 h = hash_file(workdir / "src" / "main.py")
363 added, modified, deleted, untracked, added_dirs, deleted_dirs = \
364 diff_workdir_vs_snapshot(workdir, {"src/main.py": h}, last_directories=["src"])
365 assert "src" not in added_dirs
366 assert "src" not in deleted_dirs
367
368 def test_nonexistent_workdir_returns_all_deleted(self, tmp_path: pathlib.Path) -> None:
369 missing = tmp_path / "gone"
370 added, modified, deleted, untracked, added_dirs, deleted_dirs = \
371 diff_workdir_vs_snapshot(missing, {"f.py": "h"}, last_directories=["src"])
372 assert "f.py" in deleted
373 assert "src" in deleted_dirs
374 assert not added
375
376 def test_pruned_dirs_not_tracked(self, workdir: pathlib.Path) -> None:
377 (workdir / "node_modules").mkdir()
378 (workdir / "node_modules" / "pkg.js").write_bytes(b"x")
379 (workdir / "src").mkdir()
380 (workdir / "src" / "app.py").write_bytes(b"y")
381 added, modified, deleted, untracked, added_dirs, deleted_dirs = \
382 diff_workdir_vs_snapshot(workdir, {})
383 assert "node_modules" not in added_dirs
384 assert "src" in added_dirs or "src/app.py" in untracked
385
386
387 # ===========================================================================
388 # 5. walk_workdir_with_dirs — unit
389 # ===========================================================================
390
391 class TestWalkWorkdirWithDirs:
392 def test_empty_dir_returns_empty(self, workdir: pathlib.Path) -> None:
393 files, dirs = walk_workdir_with_dirs(workdir)
394 assert files == {}
395 assert dirs == []
396
397 def test_flat_files_no_dirs(self, workdir: pathlib.Path) -> None:
398 (workdir / "a.py").write_bytes(b"x")
399 files, dirs = walk_workdir_with_dirs(workdir)
400 assert "a.py" in files
401 assert dirs == []
402
403 def test_nested_file_dir_tracked(self, workdir: pathlib.Path) -> None:
404 (workdir / "src").mkdir()
405 (workdir / "src" / "main.py").write_bytes(b"x")
406 files, dirs = walk_workdir_with_dirs(workdir)
407 assert "src/main.py" in files
408 assert "src" in dirs
409
410 def test_deeply_nested_dirs_all_tracked(self, workdir: pathlib.Path) -> None:
411 deep = workdir / "a" / "b" / "c"
412 deep.mkdir(parents=True)
413 (deep / "f.py").write_bytes(b"x")
414 files, dirs = walk_workdir_with_dirs(workdir)
415 assert "a" in dirs
416 assert "a/b" in dirs
417 assert "a/b/c" in dirs
418
419 def test_dirs_sorted(self, workdir: pathlib.Path) -> None:
420 for name in ("zzz", "aaa", "mmm"):
421 (workdir / name).mkdir()
422 (workdir / name / "f.py").write_bytes(b"x")
423 _, dirs = walk_workdir_with_dirs(workdir)
424 assert dirs == sorted(dirs)
425
426 def test_pruned_dirs_excluded(self, workdir: pathlib.Path) -> None:
427 (workdir / "node_modules").mkdir()
428 (workdir / "node_modules" / "lib.js").write_bytes(b"x")
429 (workdir / "__pycache__").mkdir()
430 (workdir / "__pycache__" / "mod.pyc").write_bytes(b"x")
431 _, dirs = walk_workdir_with_dirs(workdir)
432 assert "node_modules" not in dirs
433 assert "__pycache__" not in dirs
434
435 def test_symlinks_not_followed(self, workdir: pathlib.Path) -> None:
436 real = workdir / "real_dir"
437 real.mkdir()
438 (real / "secret.py").write_bytes(b"secret")
439 link = workdir / "link_dir"
440 link.symlink_to(real)
441 files, dirs = walk_workdir_with_dirs(workdir)
442 # symlink directory should not be descended (followlinks=False)
443 assert "link_dir/secret.py" not in files
444
445
446 # ===========================================================================
447 # 6. SnapshotRecord.directories serialisation — unit
448 # ===========================================================================
449
450 class TestSnapshotRecordDirectories:
451 def test_default_directories_is_empty_list(self) -> None:
452 rec = SnapshotRecord(snapshot_id="abc", manifest={})
453 assert rec.directories == []
454
455 def test_to_dict_includes_directories(self) -> None:
456 rec = SnapshotRecord(snapshot_id="abc", manifest={}, directories=["src", "lib"])
457 d = rec.to_dict()
458 assert d["directories"] == ["src", "lib"]
459
460 def test_from_dict_roundtrip(self) -> None:
461 rec = SnapshotRecord(snapshot_id="abc", manifest={"f.py": "h"}, directories=["pkg"])
462 loaded = SnapshotRecord.from_dict(rec.to_dict())
463 assert loaded.directories == ["pkg"]
464
465 def test_from_msgpack_roundtrip(self) -> None:
466 rec = SnapshotRecord(snapshot_id="xyz", manifest={}, directories=["a", "b"])
467 d: MsgpackDict = dict(rec.to_dict())
468 loaded = SnapshotRecord.from_msgpack(d)
469 assert loaded.directories == ["a", "b"]
470
471 def test_from_msgpack_missing_field_defaults_empty(self) -> None:
472 d: MsgpackDict = {
473 "snapshot_id": "abc",
474 "manifest": {},
475 "created_at": now_utc_iso(),
476 "note": "",
477 }
478 rec = SnapshotRecord.from_msgpack(d)
479 assert rec.directories == []
480
481 def test_from_msgpack_filters_non_string_items(self) -> None:
482 d: MsgpackDict = {
483 "snapshot_id": "abc",
484 "manifest": {},
485 "directories": ["valid", 42, None, "also_valid"],
486 "created_at": now_utc_iso(),
487 "note": "",
488 }
489 rec = SnapshotRecord.from_msgpack(d)
490 assert rec.directories == ["valid", "also_valid"]
491
492 def test_from_msgpack_non_list_directories_defaults_empty(self) -> None:
493 d: MsgpackDict = {
494 "snapshot_id": "abc",
495 "manifest": {},
496 "directories": "not-a-list",
497 "created_at": now_utc_iso(),
498 "note": "",
499 }
500 rec = SnapshotRecord.from_msgpack(d)
501 assert rec.directories == []
502
503 def test_to_dict_returns_copy_not_reference(self) -> None:
504 dirs = ["src"]
505 rec = SnapshotRecord(snapshot_id="abc", manifest={}, directories=dirs)
506 d = rec.to_dict()
507 d["directories"].append("mutated")
508 assert rec.directories == ["src"]
509
510
511 # ===========================================================================
512 # 7. write_snapshot / read_snapshot roundtrip with directories — integration
513 # ===========================================================================
514
515 class TestWriteReadSnapshotWithDirectories:
516 def test_roundtrip_preserves_directories(self, store: pathlib.Path) -> None:
517 manifest = {"src/main.py": fake_id("h1"), "src/util.py": fake_id("h2")}
518 dirs = ["src"]
519 sid = compute_snapshot_id(manifest, dirs)
520 rec = SnapshotRecord(snapshot_id=sid, manifest=manifest, directories=dirs)
521 write_snapshot(store, rec)
522
523 loaded = read_snapshot(store, sid)
524 assert loaded is not None
525 assert loaded.directories == ["src"]
526
527 def test_roundtrip_empty_directories(self, store: pathlib.Path) -> None:
528 manifest = {"f.py": fake_id("h1")}
529 sid = compute_snapshot_id(manifest, [])
530 rec = SnapshotRecord(snapshot_id=sid, manifest=manifest, directories=[])
531 write_snapshot(store, rec)
532 loaded = read_snapshot(store, sid)
533 assert loaded is not None
534 assert loaded.directories == []
535
536 def test_roundtrip_deeply_nested_dirs(self, store: pathlib.Path) -> None:
537 manifest = {"a/b/c/d.py": fake_id("h1")}
538 dirs = directories_from_manifest(manifest)
539 sid = compute_snapshot_id(manifest, dirs)
540 rec = SnapshotRecord(snapshot_id=sid, manifest=manifest, directories=dirs)
541 write_snapshot(store, rec)
542 loaded = read_snapshot(store, sid)
543 assert loaded is not None
544 assert loaded.directories == ["a", "a/b", "a/b/c"]
545
546 def test_snapshot_id_includes_dirs_in_verification(self, store: pathlib.Path) -> None:
547 """read_snapshot verifies the stored ID — tampering with dirs must fail."""
548 manifest = {"f.py": fake_id("h1")}
549 dirs = ["src"]
550 sid = compute_snapshot_id(manifest, dirs)
551 rec = SnapshotRecord(snapshot_id=sid, manifest=manifest, directories=dirs)
552 write_snapshot(store, rec)
553
554 # Compute ID without dirs — must be different
555 sid_no_dirs = compute_snapshot_id(manifest, [])
556 assert sid != sid_no_dirs
557
558 def test_directory_rename_produces_different_snapshot_id(self, store: pathlib.Path) -> None:
559 manifest = {"f.py": fake_id("h1")}
560 id_old = compute_snapshot_id(manifest, ["old_name"])
561 id_new = compute_snapshot_id(manifest, ["new_name"])
562 assert id_old != id_new
563
564
565 # ===========================================================================
566 # 8. DirectoryRenameOp TypedDict — unit
567 # ===========================================================================
568
569 class TestDirectoryRenameOp:
570 def test_construct_fields(self) -> None:
571 op = DirectoryRenameOp(
572 op="directory_rename",
573 address="new/path",
574 from_address="old/path",
575 file_count=5,
576 )
577 assert op["op"] == "directory_rename"
578 assert op["address"] == "new/path"
579 assert op["from_address"] == "old/path"
580 assert op["file_count"] == 5
581
582 def test_zero_file_count_allowed(self) -> None:
583 op = DirectoryRenameOp(
584 op="directory_rename",
585 address="a",
586 from_address="b",
587 file_count=0,
588 )
589 assert op["file_count"] == 0
590
591
592 # ===========================================================================
593 # 9. CodePlugin.diff directory rename detection — integration
594 # ===========================================================================
595
596 class TestCodePluginDiffDirectories:
597 @pytest.fixture()
598 def plugin(self) -> CodePlugin:
599 from muse.plugins.code.plugin import CodePlugin
600 return CodePlugin()
601
602 def _snap(self, files: Manifest, dirs: list[str] | None = None) -> SnapshotManifest:
603 d = dirs if dirs is not None else directories_from_manifest(files)
604 return SnapshotManifest(files=files, domain="code", directories=d)
605
606 def test_directory_rename_emits_directory_rename_op(self, plugin: CodePlugin) -> None:
607 base = self._snap({"src/a.py": "h1", "src/b.py": "h2"}, ["src"])
608 target = self._snap({"lib/a.py": "h1", "lib/b.py": "h2"}, ["lib"])
609 delta = plugin.diff(base, target)
610 ops = delta["ops"]
611 dir_rename_ops = [o for o in ops if o["op"] == "directory_rename"]
612 assert len(dir_rename_ops) == 1
613 assert dir_rename_ops[0]["from_address"] == "src"
614 assert dir_rename_ops[0]["address"] == "lib"
615 assert dir_rename_ops[0]["file_count"] == 2
616
617 def test_directory_rename_suppresses_file_level_ops(self, plugin: CodePlugin) -> None:
618 base = self._snap({"src/a.py": "h1"}, ["src"])
619 target = self._snap({"lib/a.py": "h1"}, ["lib"])
620 delta = plugin.diff(base, target)
621 ops = delta["ops"]
622 # No plain insert/delete for the covered file paths
623 file_ops = [o for o in ops if o["op"] in ("insert", "delete") and "/" in o["address"]]
624 assert not any(o["address"] in ("src/a.py", "lib/a.py") for o in file_ops)
625
626 def test_plain_added_dir_emits_insert_op(self, plugin: CodePlugin) -> None:
627 base = self._snap({}, [])
628 target = self._snap({"new/f.py": "h1"}, ["new"])
629 delta = plugin.diff(base, target)
630 ops = delta["ops"]
631 insert_dir_ops = [o for o in ops if o["op"] == "insert" and o["address"] == "new"]
632 assert len(insert_dir_ops) == 1
633
634 def test_plain_deleted_dir_emits_delete_op(self, plugin: CodePlugin) -> None:
635 base = self._snap({"old/f.py": "h1"}, ["old"])
636 target = self._snap({}, [])
637 delta = plugin.diff(base, target)
638 ops = delta["ops"]
639 delete_dir_ops = [o for o in ops if o["op"] == "delete" and o["address"] == "old"]
640 assert len(delete_dir_ops) == 1
641
642 def test_no_dir_changes_no_dir_ops(self, plugin: CodePlugin) -> None:
643 base = self._snap({"src/a.py": "h1"}, ["src"])
644 target = self._snap({"src/a.py": "h2"}, ["src"])
645 delta = plugin.diff(base, target)
646 ops = delta["ops"]
647 dir_ops = [o for o in ops if o["op"] in ("directory_rename",) or
648 (o["op"] in ("insert", "delete") and "::" not in o["address"] and "/" not in o["address"])]
649 assert not any(o["op"] == "directory_rename" for o in dir_ops)
650
651 def test_no_directories_field_no_crash(self, plugin: CodePlugin) -> None:
652 # Snapshots without the directories key should not crash
653 base = SnapshotManifest(files={"a.py": "h1"}, domain="code", directories=[])
654 target = SnapshotManifest(files={"b.py": "h1"}, domain="code", directories=[])
655 delta = plugin.diff(base, target)
656 assert "ops" in delta
657
658
659 # ===========================================================================
660 # 10. delta_summary directory rename counting — unit
661 # ===========================================================================
662
663 class TestDeltaSummaryDirectories:
664 def _make_dir_rename_op(self, old: str, new: str, file_count: int = 1) -> DirectoryRenameOp:
665 return DirectoryRenameOp(
666 op="directory_rename",
667 address=new,
668 from_address=old,
669 file_count=file_count,
670 )
671
672 def test_no_changes_returns_no_changes(self) -> None:
673 from muse.plugins.code.symbol_diff import delta_summary
674 assert delta_summary([]) == "no changes"
675
676 def test_single_directory_rename(self) -> None:
677 from muse.plugins.code.symbol_diff import delta_summary
678 ops = [self._make_dir_rename_op("old", "new")]
679 result = delta_summary(ops)
680 assert "1 directory renamed" in result
681
682 def test_two_directory_renames_plural(self) -> None:
683 from muse.plugins.code.symbol_diff import delta_summary
684 ops = [
685 self._make_dir_rename_op("a", "x"),
686 self._make_dir_rename_op("b", "y"),
687 ]
688 result = delta_summary(ops)
689 assert "2 directories renamed" in result
690
691 def test_directory_rename_combined_with_file_ops(self) -> None:
692 from muse.plugins.code.symbol_diff import delta_summary
693 from muse.domain import InsertOp
694 insert_op = InsertOp(
695 op="insert", address="new_file.py",
696 position=None, content_id="h1", content_summary="",
697 )
698 rename_op = self._make_dir_rename_op("src", "lib")
699 result = delta_summary([insert_op, rename_op])
700 assert "added" in result
701 assert "directory renamed" in result
702
703 def test_directory_rename_not_counted_as_file(self) -> None:
704 from muse.plugins.code.symbol_diff import delta_summary
705 ops = [self._make_dir_rename_op("old", "new")]
706 result = delta_summary(ops)
707 assert "file" not in result
708
709
710 # ===========================================================================
711 # 11. replay_one propagates directories — integration
712 # ===========================================================================
713
714 class TestReplayOneWithDirectories:
715 def _write_obj(self, store: pathlib.Path, content: bytes) -> str:
716 from muse.core.object_store import write_object
717 oid = blob_id(content)
718 write_object(store, oid, content)
719 return oid
720
721 def test_clean_merge_produces_snapshot_with_dirs(self, store: pathlib.Path) -> None:
722 from muse.core.rebase import replay_one
723 from muse.plugins.code.plugin import CodePlugin
724
725 plugin = CodePlugin()
726 domain = "code"
727
728 # Write actual file objects so apply_manifest can restore them
729 oid_a = self._write_obj(store, b"# a.py content\n")
730 oid_b = self._write_obj(store, b"# b.py content\n")
731
732 # Create a base commit: one file in src/
733 base_manifest = {"src/a.py": oid_a}
734 base_dirs = directories_from_manifest(base_manifest)
735 base_snap = _make_snap(store, base_manifest, base_dirs)
736 base_commit = _make_commit_rec(store, base_snap, message="base")
737
738 # Create "theirs" commit: adds src/b.py (same parent = base)
739 theirs_manifest = {"src/a.py": oid_a, "src/b.py": oid_b}
740 theirs_dirs = directories_from_manifest(theirs_manifest)
741 theirs_snap = _make_snap(store, theirs_manifest, theirs_dirs)
742 theirs_commit = _make_commit_rec(
743 store, theirs_snap, parent_id=base_commit.commit_id, message="theirs"
744 )
745
746 # replay theirs_commit on top of base_commit (onto = base)
747 result = replay_one(
748 root=store,
749 commit=theirs_commit,
750 parent_id=base_commit.commit_id,
751 plugin=plugin,
752 domain=domain,
753 repo_id=_REPO_ID,
754 branch="main",
755 )
756
757 assert isinstance(result, CommitRecord), f"Expected CommitRecord, got: {result}"
758 replayed_snap = read_snapshot(store, result.snapshot_id)
759 assert replayed_snap is not None
760 assert replayed_snap.directories == ["src"]
761
762 def test_conflict_returns_path_list_not_commit(self, store: pathlib.Path) -> None:
763 from muse.core.rebase import replay_one
764 from muse.plugins.code.plugin import CodePlugin
765
766 plugin = CodePlugin()
767
768 oid_v1 = self._write_obj(store, b"version 1\n")
769 oid_v2 = self._write_obj(store, b"version 2\n")
770 oid_v3 = self._write_obj(store, b"version 3\n")
771
772 # base: file a.py = v1
773 base_manifest = {"a.py": oid_v1}
774 base_snap = _make_snap(store, base_manifest)
775 base_commit = _make_commit_rec(store, base_snap, message="base")
776
777 # "theirs" modifies a.py from v1 → v2
778 theirs_manifest = {"a.py": oid_v2}
779 theirs_snap = _make_snap(store, theirs_manifest)
780 theirs_commit = _make_commit_rec(
781 store, theirs_snap, parent_id=base_commit.commit_id, message="theirs"
782 )
783
784 # "ours" (parent_id in replay) also modified a.py from v1 → v3 (conflict)
785 ours_manifest = {"a.py": oid_v3}
786 ours_snap = _make_snap(store, ours_manifest)
787 ours_commit = _make_commit_rec(store, ours_snap, message="ours")
788
789 result = replay_one(
790 root=store,
791 commit=theirs_commit,
792 parent_id=ours_commit.commit_id,
793 plugin=plugin,
794 domain="code",
795 repo_id=_REPO_ID,
796 branch="main",
797 )
798
799 # Should return conflict paths, not a CommitRecord
800 assert isinstance(result, list)
801
802
803 # ===========================================================================
804 # 12. E2E workflow — full CLI commit/branch/rename/merge cycle
805 # ===========================================================================
806
807 def _muse(repo: pathlib.Path, *args: str) -> subprocess.CompletedProcess[str]:
808 import shutil
809 import sys
810 # Prefer a .venv installation (development mode), fall back to the
811 # interpreter's sibling or the PATH-resolved binary.
812 venv_muse = pathlib.Path(__file__).parent.parent / ".venv" / "bin" / "muse"
813 if venv_muse.exists():
814 muse_bin = str(venv_muse)
815 else:
816 sibling = pathlib.Path(sys.executable).parent / "muse"
817 muse_bin = str(sibling) if sibling.exists() else (shutil.which("muse") or "muse")
818 return subprocess.run(
819 [muse_bin, *args],
820 cwd=str(repo),
821 capture_output=True,
822 text=True,
823 )
824
825
826 class TestDirectoriesE2EWorkflow:
827 @pytest.fixture()
828 def repo(self, tmp_path: pathlib.Path) -> pathlib.Path:
829 result = _muse(tmp_path, "init")
830 assert result.returncode == 0, result.stderr
831 return tmp_path
832
833 def test_commit_records_directories(self, repo: pathlib.Path) -> None:
834 (repo / "src").mkdir()
835 (repo / "src" / "main.py").write_text("x = 1\n")
836 r = _muse(repo, "commit", "-m", "add src/main.py")
837 assert r.returncode == 0, r.stderr
838
839 # Read the snapshot from store and confirm directories is populated
840 from muse.core.store import read_commit, get_head_snapshot_id
841 from muse.core.repo import read_repo_id
842 from muse.core.store import read_current_branch
843 repo_id = read_repo_id(repo)
844 branch = read_current_branch(repo)
845 snap_id = get_head_snapshot_id(repo, repo_id, branch)
846 assert snap_id is not None
847 snap = read_snapshot(repo, snap_id)
848 assert snap is not None
849 assert "src" in snap.directories
850
851 def test_snapshot_id_changes_on_dir_rename(self, repo: pathlib.Path) -> None:
852 (repo / "old_name").mkdir()
853 (repo / "old_name" / "f.py").write_text("x = 1\n")
854 _muse(repo, "commit", "-m", "add old_name/")
855
856 from muse.core.store import get_head_snapshot_id, read_current_branch
857 from muse.core.repo import read_repo_id
858 repo_id = read_repo_id(repo)
859 branch = read_current_branch(repo)
860 sid_before = get_head_snapshot_id(repo, repo_id, branch)
861
862 # Simulate rename: remove old dir, create new dir with same content
863 import shutil
864 shutil.move(str(repo / "old_name"), str(repo / "new_name"))
865 _muse(repo, "commit", "-m", "rename dir")
866
867 sid_after = get_head_snapshot_id(repo, repo_id, branch)
868 assert sid_before != sid_after
869
870 def test_status_handles_directory_rename_op(self, repo: pathlib.Path) -> None:
871 (repo / "src").mkdir()
872 (repo / "src" / "app.py").write_text("app = True\n")
873 _muse(repo, "commit", "-m", "initial")
874
875 import shutil
876 shutil.move(str(repo / "src"), str(repo / "lib"))
877
878 r = _muse(repo, "status")
879 assert r.returncode == 0, r.stderr
880
881 def test_nested_directories_tracked_through_commit(self, repo: pathlib.Path) -> None:
882 deep = repo / "a" / "b" / "c"
883 deep.mkdir(parents=True)
884 (deep / "f.py").write_text("pass\n")
885 r = _muse(repo, "commit", "-m", "deep nest")
886 assert r.returncode == 0, r.stderr
887
888 from muse.core.store import get_head_snapshot_id, read_current_branch
889 from muse.core.repo import read_repo_id
890 repo_id = read_repo_id(repo)
891 branch = read_current_branch(repo)
892 snap = read_snapshot(repo, get_head_snapshot_id(repo, repo_id, branch))
893 assert snap is not None
894 assert "a" in snap.directories
895 assert "a/b" in snap.directories
896 assert "a/b/c" in snap.directories
897
898
899 # ===========================================================================
900 # 13. Empty directory ghost bug
901 #
902 # Regression tests for: empty directories left on disk after their files are
903 # deleted and committed must NOT appear in `muse status --json` `added`.
904 #
905 # Root cause: CodePlugin.snapshot() recorded every directory visited by
906 # os.walk() into `dirs`, including empty ones. These empty dirs had no
907 # counterpart in HEAD, so diff() produced InsertOp entries for them,
908 # and status --json reported them as `added`.
909 # ===========================================================================
910
911 class TestEmptyDirectoryGhost:
912 """Empty orphan directories must not appear as 'added' in muse status."""
913
914 @pytest.fixture()
915 def repo(self, tmp_path: pathlib.Path) -> pathlib.Path:
916 result = _muse(tmp_path, "init")
917 assert result.returncode == 0, result.stderr
918 return tmp_path
919
920 # ── Unit: snapshot() must not include empty dirs ──────────────────────────
921
922 def test_snapshot_excludes_empty_directory(self, tmp_path: pathlib.Path) -> None:
923 """CodePlugin.snapshot() must not list a directory that has no files."""
924 from muse.plugins.code.plugin import CodePlugin
925 _muse(tmp_path, "init")
926 plugin = CodePlugin()
927
928 # Empty nested directory — no files, no .musekeep
929 (tmp_path / "empty_pkg" / "sub").mkdir(parents=True)
930
931 snap = plugin.snapshot(tmp_path)
932 assert "empty_pkg" not in snap["directories"], (
933 "Empty directory 'empty_pkg' must not appear in snapshot directories"
934 )
935 assert "empty_pkg/sub" not in snap["directories"], (
936 "Empty nested directory 'empty_pkg/sub' must not appear in snapshot directories"
937 )
938
939 def test_snapshot_includes_dir_with_files(self, tmp_path: pathlib.Path) -> None:
940 """Directories containing files must still appear in the snapshot."""
941 from muse.plugins.code.plugin import CodePlugin
942 _muse(tmp_path, "init")
943 plugin = CodePlugin()
944
945 (tmp_path / "pkg").mkdir()
946 (tmp_path / "pkg" / "mod.py").write_text("x = 1\n")
947
948 snap = plugin.snapshot(tmp_path)
949 assert "pkg" in snap["directories"]
950
951 def test_snapshot_includes_musekeep_empty_dir(self, tmp_path: pathlib.Path) -> None:
952 """An empty directory with a .musekeep marker must be tracked."""
953 from muse.plugins.code.plugin import CodePlugin
954 _muse(tmp_path, "init")
955 plugin = CodePlugin()
956
957 (tmp_path / "intentionally_empty").mkdir()
958 (tmp_path / "intentionally_empty" / ".musekeep").write_text("")
959
960 snap = plugin.snapshot(tmp_path)
961 assert "intentionally_empty" in snap["directories"]
962
963 # ── Integration: status --json must not list orphan empty dirs ────────────
964
965 def test_status_clean_after_deleting_only_file_in_dir(self, repo: pathlib.Path) -> None:
966 """After deleting the last file in a directory and committing,
967 muse status must report clean — not list the empty leftover dir as added."""
968 pkg = repo / "tourdeforce" / "clients"
969 pkg.mkdir(parents=True)
970 (pkg / "module.py").write_text("x = 1\n")
971 _muse(repo, "commit", "-m", "add tourdeforce")
972
973 # Delete the file — leave the empty directories on disk
974 (pkg / "module.py").unlink()
975 _muse(repo, "commit", "-m", "delete module.py")
976
977 r = _muse(repo, "status", "--json")
978 assert r.returncode == 0, r.stderr
979 data = json.loads(r.stdout)
980
981 assert data["clean"] is True, (
982 f"Expected clean status after deleting all files; added={data['added']}"
983 )
984 assert "tourdeforce" not in data["added"], (
985 "Empty leftover directory 'tourdeforce' must not appear as added"
986 )
987 assert "tourdeforce/clients" not in data["added"], (
988 "Empty leftover directory 'tourdeforce/clients' must not appear as added"
989 )
990
991 def test_status_does_not_report_never_committed_empty_dir(self, repo: pathlib.Path) -> None:
992 """An empty directory that was never committed must not appear in added."""
993 (repo / "orphan" / "nested").mkdir(parents=True)
994 # No files, never committed
995
996 r = _muse(repo, "status", "--json")
997 assert r.returncode == 0, r.stderr
998 data = json.loads(r.stdout)
999
1000 assert "orphan" not in data["added"], (
1001 "Untracked empty directory 'orphan' must not appear as added"
1002 )
1003 assert "orphan/nested" not in data["added"], (
1004 "Untracked empty nested directory must not appear as added"
1005 )
1006
1007 def test_status_reports_added_for_dir_with_new_file(self, repo: pathlib.Path) -> None:
1008 """A new directory containing a real file must still appear as added."""
1009 (repo / "new_pkg").mkdir()
1010 (repo / "new_pkg" / "api.py").write_text("pass\n")
1011
1012 r = _muse(repo, "status", "--json")
1013 assert r.returncode == 0, r.stderr
1014 data = json.loads(r.stdout)
1015
1016 # The file should be added (the directory entry itself may or may not be
1017 # in added — what matters is the file is visible and dirs without files are not)
1018 all_visible = data["added"] + data["untracked"]
1019 assert any("new_pkg" in p for p in all_visible), (
1020 "New directory with a file should be reflected in added or untracked"
1021 )
1022
1023
1024 # ===========================================================================
1025 # 13. Stress / performance
1026 # ===========================================================================
1027
1028 class TestDirectoriesStress:
1029 def test_directories_from_manifest_1000_files(self) -> None:
1030 manifest = {
1031 f"pkg_{i}/sub_{j}/file_{k}.py": f"hash{i}{j}{k}"
1032 for i in range(10)
1033 for j in range(10)
1034 for k in range(10)
1035 }
1036 assert len(manifest) == 1000
1037 start = time.monotonic()
1038 dirs = directories_from_manifest(manifest)
1039 elapsed = time.monotonic() - start
1040 # Should complete in under 1 second for 1000 files
1041 assert elapsed < 1.0, f"directories_from_manifest took {elapsed:.3f}s for 1000 files"
1042 # 10 top-level dirs (pkg_0..9) + 100 second-level dirs (pkg_N/sub_M) = 110
1043 assert len(dirs) == 110
1044
1045 def test_detect_directory_renames_50_dirs(self) -> None:
1046 # 50 dirs each with 5 files, all renamed old_N → new_N
1047 last: Manifest = {}
1048 current: Manifest = {}
1049 for i in range(50):
1050 for j in range(5):
1051 h = blob_id(f"content_{i}_{j}".encode())
1052 last[f"old_{i}/file_{j}.py"] = h
1053 current[f"new_{i}/file_{j}.py"] = h
1054
1055 deleted = {f"old_{i}" for i in range(50)}
1056 added = {f"new_{i}" for i in range(50)}
1057
1058 start = time.monotonic()
1059 renames = detect_directory_renames(deleted, added, last, current)
1060 elapsed = time.monotonic() - start
1061
1062 assert elapsed < 2.0, f"detect_directory_renames took {elapsed:.3f}s for 50 dirs"
1063 assert len(renames) == 50
1064
1065 def test_compute_snapshot_id_large_dir_list(self) -> None:
1066 manifest = {f"f_{i}.py": fake_id(f"h{i}") for i in range(500)}
1067 dirs = [f"dir_{i}" for i in range(500)]
1068 start = time.monotonic()
1069 sid = compute_snapshot_id(manifest, dirs)
1070 elapsed = time.monotonic() - start
1071 assert elapsed < 1.0, f"compute_snapshot_id took {elapsed:.3f}s for 500 dirs"
1072 assert len(sid) == 71
1073
1074 def test_walk_workdir_with_dirs_deep_tree(self, tmp_path: pathlib.Path) -> None:
1075 # 20 levels of nesting
1076 deep = tmp_path
1077 for level in range(20):
1078 deep = deep / f"level_{level}"
1079 deep.mkdir()
1080 (deep / "leaf.py").write_bytes(b"x")
1081
1082 start = time.monotonic()
1083 files, dirs = walk_workdir_with_dirs(tmp_path)
1084 elapsed = time.monotonic() - start
1085
1086 assert elapsed < 2.0, f"walk_workdir_with_dirs took {elapsed:.3f}s on 20-level tree"
1087 assert "leaf.py" in "".join(files.keys())
1088 assert len(dirs) == 20
1089
1090
1091 # ===========================================================================
1092 # 14. Security
1093 # ===========================================================================
1094
1095 class TestDirectoriesSecurity:
1096 def test_path_traversal_in_directory_address_not_resolved(self) -> None:
1097 # directories_from_manifest should treat path components literally
1098 manifest = {"../../etc/shadow": "h1"}
1099 dirs = directories_from_manifest(manifest)
1100 # The result should contain "../.." and "../../etc" literally, not resolve them
1101 # The important thing: no OS path resolution happens
1102 for d in dirs:
1103 assert not pathlib.Path(d).is_absolute()
1104
1105 def test_null_byte_in_directory_path_handled(self) -> None:
1106 # Null bytes in paths are unusual but should not crash
1107 manifest = {"src\x00/malicious.py": fake_id("h1")}
1108 try:
1109 dirs = directories_from_manifest(manifest)
1110 sid = compute_snapshot_id(manifest, dirs)
1111 assert len(sid) == 71
1112 except (ValueError, TypeError):
1113 pass # rejecting is also acceptable
1114
1115 def test_very_long_directory_path(self) -> None:
1116 long_name = "a" * 4096
1117 manifest = {f"{long_name}/f.py": fake_id("h1")}
1118 dirs = directories_from_manifest(manifest)
1119 assert dirs == [long_name]
1120 sid = compute_snapshot_id(manifest, dirs)
1121 assert len(sid) == 71
1122
1123 def test_symlinked_dir_not_followed_during_walk(self, tmp_path: pathlib.Path) -> None:
1124 sensitive = tmp_path / "sensitive"
1125 sensitive.mkdir()
1126 (sensitive / "secret.txt").write_bytes(b"SECRET")
1127
1128 repo_root = tmp_path / "repo"
1129 repo_root.mkdir()
1130 link = repo_root / "malicious_link"
1131 link.symlink_to(sensitive)
1132
1133 files, dirs = walk_workdir_with_dirs(repo_root)
1134 assert "malicious_link/secret.txt" not in files
1135
1136 def test_snapshot_record_with_adversarial_dirs_survives_roundtrip(self, store: pathlib.Path) -> None:
1137 # Adversarial: dirs containing special characters
1138 dirs = ["src", "src/sub dir", "a-b_c.d"]
1139 manifest = {"src/f.py": fake_id("h1")}
1140 sid = compute_snapshot_id(manifest, dirs)
1141 rec = SnapshotRecord(snapshot_id=sid, manifest=manifest, directories=dirs)
1142 # to_dict / from_dict roundtrip
1143 loaded = SnapshotRecord.from_dict(rec.to_dict())
1144 assert loaded.directories == dirs
1145
1146 def test_detect_directory_renames_no_prefix_confusion(self) -> None:
1147 # "a" should not confuse files under "ab/" as being under "a/"
1148 # because the prefix check uses "a/" (with trailing slash)
1149 last = {"a/f.py": "h1"}
1150 current = {"ab/f.py": "h1"}
1151 # "ab/f.py" does NOT start with "a/" so old_files under "a/" = {"f.py": "h1"}
1152 # but new_files under "ab/" = {"f.py": "h1"} — these DO match, so rename is detected
1153 # (which is correct: the file genuinely moved from a/ to ab/)
1154 renames = detect_directory_renames({"a"}, {"ab"}, last, current)
1155 assert renames == [("a", "ab")]
1156
1157 def test_detect_directory_renames_prefix_does_not_bleed_across_siblings(self) -> None:
1158 # "models" should never absorb files from "models_v2" in the source manifest
1159 # when looking at what files belong to "models/"
1160 last = {"models/user.py": "h1", "models_v2/user.py": "h2"}
1161 # Both dirs deleted, one new dir added with only models_v2's content
1162 current = {"new_home/user.py": "h2"}
1163 renames = detect_directory_renames({"models", "models_v2"}, {"new_home"}, last, current)
1164 # "new_home" has {"user.py": "h2"} which matches "models_v2/" not "models/"
1165 assert ("models_v2", "new_home") in renames
1166 assert ("models", "new_home") not in renames
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 120 days ago