gabriel / muse public
test_core_paths_init_repo_dirs.py python
259 lines 9.6 KB
Raw
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ breaking 30 days ago
1 """Tests for ``muse.core.paths.init_repo_dirs``.
2
3 Seven tiers:
4 Tier 1 — Unit: each directory helper, return value, idempotency
5 Tier 2 — Integration: dirs usable by store/commit/snapshot layer
6 Tier 3 — End-to-end: full CLI init flow uses init_repo_dirs layout
7 Tier 4 — Stress: concurrent calls, large number of calls
8 Tier 5 — State integrity: no files written, no extra dirs created
9 Tier 6 — Security: symlink attack, path traversal, permission edge cases
10 Tier 7 — Performance: single call completes well under threshold
11 """
12
13 from __future__ import annotations
14
15 import concurrent.futures
16 import pathlib
17 import threading
18 import time
19 from collections.abc import Callable
20
21 import pytest
22
23 from muse.core.types import long_id
24 from muse.core.paths import (
25 commits_dir,
26 heads_dir,
27 init_repo_dirs,
28 logs_dir,
29 muse_dir,
30 objects_dir,
31 remotes_dir,
32 shelf_dir,
33 snapshots_dir,
34 )
35
36
37 # ---------------------------------------------------------------------------
38 # Tier 1 — Unit
39 # ---------------------------------------------------------------------------
40
41
42 class TestInitRepoDirsUnit:
43 def test_returns_root(self, tmp_path: pathlib.Path) -> None:
44 assert init_repo_dirs(tmp_path) is tmp_path
45
46 def test_muse_dir_created(self, tmp_path: pathlib.Path) -> None:
47 init_repo_dirs(tmp_path)
48 assert muse_dir(tmp_path).is_dir()
49
50 @pytest.mark.parametrize("dir_fn", [
51 objects_dir, commits_dir, snapshots_dir,
52 heads_dir, remotes_dir, logs_dir, shelf_dir,
53 ])
54 def test_each_subdirectory_created(self, tmp_path: pathlib.Path, dir_fn: Callable[[pathlib.Path], pathlib.Path]) -> None:
55 init_repo_dirs(tmp_path)
56 assert dir_fn(tmp_path).is_dir()
57
58 def test_idempotent_second_call_no_raise(self, tmp_path: pathlib.Path) -> None:
59 init_repo_dirs(tmp_path)
60 init_repo_dirs(tmp_path) # must not raise FileExistsError
61
62 def test_idempotent_dirs_still_present(self, tmp_path: pathlib.Path) -> None:
63 init_repo_dirs(tmp_path)
64 init_repo_dirs(tmp_path)
65 assert muse_dir(tmp_path).is_dir()
66 assert objects_dir(tmp_path).is_dir()
67
68 def test_creates_root_if_missing(self, tmp_path: pathlib.Path) -> None:
69 root = tmp_path / "new_repo"
70 assert not root.exists()
71 init_repo_dirs(root)
72 assert muse_dir(root).is_dir()
73
74 def test_creates_nested_missing_root(self, tmp_path: pathlib.Path) -> None:
75 root = tmp_path / "a" / "b" / "c"
76 init_repo_dirs(root)
77 assert muse_dir(root).is_dir()
78
79
80 # ---------------------------------------------------------------------------
81 # Tier 2 — Integration: dirs are usable by the object/commit/snapshot layers
82 # ---------------------------------------------------------------------------
83
84
85 class TestInitRepoDirsIntegration:
86 def test_objects_dir_accepts_write(self, tmp_path: pathlib.Path) -> None:
87 init_repo_dirs(tmp_path)
88 obj = objects_dir(tmp_path) / "test_blob"
89 obj.write_bytes(b"hello")
90 assert obj.read_bytes() == b"hello"
91
92 def test_commits_dir_accepts_write(self, tmp_path: pathlib.Path) -> None:
93 init_repo_dirs(tmp_path)
94 f = commits_dir(tmp_path) / "sha256" / "ab"
95 f.mkdir(parents=True)
96 assert f.is_dir()
97
98 def test_heads_dir_accepts_ref_file(self, tmp_path: pathlib.Path) -> None:
99 init_repo_dirs(tmp_path)
100 ref = heads_dir(tmp_path) / "main"
101 ref.write_text(long_id("a" * 64))
102 assert ref.read_text().startswith("sha256:")
103
104 def test_logs_dir_accepts_subdirs(self, tmp_path: pathlib.Path) -> None:
105 init_repo_dirs(tmp_path)
106 sub = logs_dir(tmp_path) / "refs" / "heads"
107 sub.mkdir(parents=True)
108 assert sub.is_dir()
109
110 def test_shelf_dir_accepts_msgpack_file(self, tmp_path: pathlib.Path) -> None:
111 init_repo_dirs(tmp_path)
112 entry = shelf_dir(tmp_path) / "entry.msgpack"
113 entry.write_bytes(b"\x82\xa3key\xa5value")
114 assert entry.exists()
115
116
117 # ---------------------------------------------------------------------------
118 # Tier 3 — End-to-end: CLI init uses same layout
119 # ---------------------------------------------------------------------------
120
121
122 class TestInitRepoDirsE2E:
123 def test_cli_init_produces_same_dirs(self, muse_repo: pathlib.Path) -> None:
124 """muse init creates at minimum the dirs that init_repo_dirs promises."""
125 expected = [
126 muse_dir, objects_dir, commits_dir, snapshots_dir,
127 heads_dir, remotes_dir, logs_dir, shelf_dir,
128 ]
129 for dir_fn in expected:
130 assert dir_fn(muse_repo).is_dir(), f"{dir_fn.__name__} missing after muse init"
131
132 def test_bare_repo_dirs_subset_of_full_init(self, tmp_path: pathlib.Path, muse_repo: pathlib.Path) -> None:
133 """Every dir init_repo_dirs creates also exists in a CLI-init repo."""
134 init_repo_dirs(tmp_path)
135 for p in muse_dir(tmp_path).rglob("*"):
136 if p.is_dir():
137 rel = p.relative_to(tmp_path)
138 assert (muse_repo / rel).is_dir(), f"{rel} missing in CLI repo"
139
140
141 # ---------------------------------------------------------------------------
142 # Tier 4 — Stress
143 # ---------------------------------------------------------------------------
144
145
146 class TestInitRepoDirsStress:
147 def test_100_sequential_calls_idempotent(self, tmp_path: pathlib.Path) -> None:
148 for _ in range(100):
149 init_repo_dirs(tmp_path)
150 assert muse_dir(tmp_path).is_dir()
151
152 def test_concurrent_calls_safe(self, tmp_path: pathlib.Path) -> None:
153 errors: list[Exception] = []
154
155 def call() -> None:
156 try:
157 init_repo_dirs(tmp_path)
158 except Exception as exc:
159 errors.append(exc)
160
161 threads = [threading.Thread(target=call) for _ in range(20)]
162 for t in threads:
163 t.start()
164 for t in threads:
165 t.join()
166
167 assert errors == [], f"concurrent calls raised: {errors}"
168 assert muse_dir(tmp_path).is_dir()
169
170 def test_threadpool_concurrent_different_roots(self, tmp_path: pathlib.Path) -> None:
171 roots = [tmp_path / f"repo_{i}" for i in range(10)]
172 with concurrent.futures.ThreadPoolExecutor(max_workers=10) as ex:
173 list(ex.map(init_repo_dirs, roots))
174 for root in roots:
175 assert muse_dir(root).is_dir()
176
177
178 # ---------------------------------------------------------------------------
179 # Tier 5 — State integrity
180 # ---------------------------------------------------------------------------
181
182
183 class TestInitRepoDirsStateIntegrity:
184 def test_no_files_written(self, tmp_path: pathlib.Path) -> None:
185 init_repo_dirs(tmp_path)
186 files = [p for p in muse_dir(tmp_path).rglob("*") if p.is_file()]
187 assert files == [], f"Unexpected files written: {files}"
188
189 def test_no_head_written(self, tmp_path: pathlib.Path) -> None:
190 init_repo_dirs(tmp_path)
191 assert not (muse_dir(tmp_path) / "HEAD").exists()
192
193 def test_no_repo_json_written(self, tmp_path: pathlib.Path) -> None:
194 init_repo_dirs(tmp_path)
195 assert not (muse_dir(tmp_path) / "repo.json").exists()
196
197 def test_no_extra_toplevel_dirs(self, tmp_path: pathlib.Path) -> None:
198 init_repo_dirs(tmp_path)
199 expected_names = {
200 "objects", "commits", "snapshots", "refs", "remotes", "logs", "shelf",
201 }
202 actual = {p.name for p in muse_dir(tmp_path).iterdir() if p.is_dir()}
203 unexpected = actual - expected_names
204 assert unexpected == set(), f"Unexpected dirs: {unexpected}"
205
206 def test_existing_files_not_touched(self, tmp_path: pathlib.Path) -> None:
207 init_repo_dirs(tmp_path)
208 sentinel = objects_dir(tmp_path) / "sentinel"
209 sentinel.write_bytes(b"preserved")
210 init_repo_dirs(tmp_path)
211 assert sentinel.read_bytes() == b"preserved"
212
213
214 # ---------------------------------------------------------------------------
215 # Tier 6 — Security
216 # ---------------------------------------------------------------------------
217
218
219 class TestInitRepoDirsSecurity:
220 def test_symlink_root_does_not_escape(self, tmp_path: pathlib.Path) -> None:
221 """init_repo_dirs on a symlinked root creates dirs under the real target."""
222 real = tmp_path / "real_repo"
223 real.mkdir()
224 link = tmp_path / "link_repo"
225 link.symlink_to(real)
226 init_repo_dirs(link)
227 assert muse_dir(real).is_dir()
228 assert not muse_dir(link).is_symlink()
229
230 def test_root_with_spaces_in_path(self, tmp_path: pathlib.Path) -> None:
231 root = tmp_path / "my repo with spaces"
232 init_repo_dirs(root)
233 assert muse_dir(root).is_dir()
234
235 def test_root_with_unicode_path(self, tmp_path: pathlib.Path) -> None:
236 root = tmp_path / "répertoire_音楽"
237 init_repo_dirs(root)
238 assert muse_dir(root).is_dir()
239
240
241 # ---------------------------------------------------------------------------
242 # Tier 7 — Performance
243 # ---------------------------------------------------------------------------
244
245
246 class TestInitRepoDirsPerformance:
247 def test_single_call_under_100ms(self, tmp_path: pathlib.Path) -> None:
248 root = tmp_path / "perf_repo"
249 start = time.perf_counter()
250 init_repo_dirs(root)
251 elapsed = time.perf_counter() - start
252 assert elapsed < 0.1, f"init_repo_dirs took {elapsed:.3f}s — expected < 0.1s"
253
254 def test_idempotent_call_under_50ms(self, tmp_path: pathlib.Path) -> None:
255 init_repo_dirs(tmp_path)
256 start = time.perf_counter()
257 init_repo_dirs(tmp_path)
258 elapsed = time.perf_counter() - start
259 assert elapsed < 0.05, f"second call took {elapsed:.3f}s — expected < 0.05s"
File History 1 commit
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ 30 days ago