test_core_paths_init_repo_dirs.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 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, heads_dir, remotes_dir, logs_dir, shelf_dir, |
| 52 | ]) |
| 53 | def test_each_subdirectory_created(self, tmp_path: pathlib.Path, dir_fn: Callable[[pathlib.Path], pathlib.Path]) -> None: |
| 54 | init_repo_dirs(tmp_path) |
| 55 | assert dir_fn(tmp_path).is_dir() |
| 56 | |
| 57 | def test_idempotent_second_call_no_raise(self, tmp_path: pathlib.Path) -> None: |
| 58 | init_repo_dirs(tmp_path) |
| 59 | init_repo_dirs(tmp_path) # must not raise FileExistsError |
| 60 | |
| 61 | def test_idempotent_dirs_still_present(self, tmp_path: pathlib.Path) -> None: |
| 62 | init_repo_dirs(tmp_path) |
| 63 | init_repo_dirs(tmp_path) |
| 64 | assert muse_dir(tmp_path).is_dir() |
| 65 | assert objects_dir(tmp_path).is_dir() |
| 66 | |
| 67 | def test_creates_root_if_missing(self, tmp_path: pathlib.Path) -> None: |
| 68 | root = tmp_path / "new_repo" |
| 69 | assert not root.exists() |
| 70 | init_repo_dirs(root) |
| 71 | assert muse_dir(root).is_dir() |
| 72 | |
| 73 | def test_creates_nested_missing_root(self, tmp_path: pathlib.Path) -> None: |
| 74 | root = tmp_path / "a" / "b" / "c" |
| 75 | init_repo_dirs(root) |
| 76 | assert muse_dir(root).is_dir() |
| 77 | |
| 78 | |
| 79 | # --------------------------------------------------------------------------- |
| 80 | # Tier 2 — Integration: dirs are usable by the object/commit/snapshot layers |
| 81 | # --------------------------------------------------------------------------- |
| 82 | |
| 83 | |
| 84 | class TestInitRepoDirsIntegration: |
| 85 | def test_objects_dir_accepts_write(self, tmp_path: pathlib.Path) -> None: |
| 86 | init_repo_dirs(tmp_path) |
| 87 | obj = objects_dir(tmp_path) / "test_blob" |
| 88 | obj.write_bytes(b"hello") |
| 89 | assert obj.read_bytes() == b"hello" |
| 90 | |
| 91 | def test_commits_dir_accepts_write(self, tmp_path: pathlib.Path) -> None: |
| 92 | init_repo_dirs(tmp_path) |
| 93 | f = commits_dir(tmp_path) / "sha256" / "ab" |
| 94 | f.mkdir(parents=True) |
| 95 | assert f.is_dir() |
| 96 | |
| 97 | def test_heads_dir_accepts_ref_file(self, tmp_path: pathlib.Path) -> None: |
| 98 | init_repo_dirs(tmp_path) |
| 99 | ref = heads_dir(tmp_path) / "main" |
| 100 | ref.write_text(long_id("a" * 64)) |
| 101 | assert ref.read_text().startswith("sha256:") |
| 102 | |
| 103 | def test_logs_dir_accepts_subdirs(self, tmp_path: pathlib.Path) -> None: |
| 104 | init_repo_dirs(tmp_path) |
| 105 | sub = logs_dir(tmp_path) / "refs" / "heads" |
| 106 | sub.mkdir(parents=True) |
| 107 | assert sub.is_dir() |
| 108 | |
| 109 | def test_shelf_dir_accepts_msgpack_file(self, tmp_path: pathlib.Path) -> None: |
| 110 | init_repo_dirs(tmp_path) |
| 111 | entry = shelf_dir(tmp_path) / "entry.msgpack" |
| 112 | entry.write_bytes(b"\x82\xa3key\xa5value") |
| 113 | assert entry.exists() |
| 114 | |
| 115 | |
| 116 | # --------------------------------------------------------------------------- |
| 117 | # Tier 3 — End-to-end: CLI init uses same layout |
| 118 | # --------------------------------------------------------------------------- |
| 119 | |
| 120 | |
| 121 | class TestInitRepoDirsE2E: |
| 122 | def test_cli_init_produces_same_dirs(self, muse_repo: pathlib.Path) -> None: |
| 123 | """muse init creates at minimum the dirs that init_repo_dirs promises.""" |
| 124 | expected = [ |
| 125 | muse_dir, objects_dir, heads_dir, remotes_dir, logs_dir, shelf_dir, |
| 126 | ] |
| 127 | for dir_fn in expected: |
| 128 | assert dir_fn(muse_repo).is_dir(), f"{dir_fn.__name__} missing after muse init" |
| 129 | |
| 130 | def test_bare_repo_dirs_subset_of_full_init(self, tmp_path: pathlib.Path, muse_repo: pathlib.Path) -> None: |
| 131 | """Every dir init_repo_dirs creates also exists in a CLI-init repo.""" |
| 132 | init_repo_dirs(tmp_path) |
| 133 | for p in muse_dir(tmp_path).rglob("*"): |
| 134 | if p.is_dir(): |
| 135 | rel = p.relative_to(tmp_path) |
| 136 | assert (muse_repo / rel).is_dir(), f"{rel} missing in CLI repo" |
| 137 | |
| 138 | |
| 139 | # --------------------------------------------------------------------------- |
| 140 | # Tier 4 — Stress |
| 141 | # --------------------------------------------------------------------------- |
| 142 | |
| 143 | |
| 144 | class TestInitRepoDirsStress: |
| 145 | def test_100_sequential_calls_idempotent(self, tmp_path: pathlib.Path) -> None: |
| 146 | for _ in range(100): |
| 147 | init_repo_dirs(tmp_path) |
| 148 | assert muse_dir(tmp_path).is_dir() |
| 149 | |
| 150 | def test_concurrent_calls_safe(self, tmp_path: pathlib.Path) -> None: |
| 151 | errors: list[Exception] = [] |
| 152 | |
| 153 | def call() -> None: |
| 154 | try: |
| 155 | init_repo_dirs(tmp_path) |
| 156 | except Exception as exc: |
| 157 | errors.append(exc) |
| 158 | |
| 159 | threads = [threading.Thread(target=call) for _ in range(20)] |
| 160 | for t in threads: |
| 161 | t.start() |
| 162 | for t in threads: |
| 163 | t.join() |
| 164 | |
| 165 | assert errors == [], f"concurrent calls raised: {errors}" |
| 166 | assert muse_dir(tmp_path).is_dir() |
| 167 | |
| 168 | def test_threadpool_concurrent_different_roots(self, tmp_path: pathlib.Path) -> None: |
| 169 | roots = [tmp_path / f"repo_{i}" for i in range(10)] |
| 170 | with concurrent.futures.ThreadPoolExecutor(max_workers=10) as ex: |
| 171 | list(ex.map(init_repo_dirs, roots)) |
| 172 | for root in roots: |
| 173 | assert muse_dir(root).is_dir() |
| 174 | |
| 175 | |
| 176 | # --------------------------------------------------------------------------- |
| 177 | # Tier 5 — State integrity |
| 178 | # --------------------------------------------------------------------------- |
| 179 | |
| 180 | |
| 181 | class TestInitRepoDirsStateIntegrity: |
| 182 | def test_no_files_written(self, tmp_path: pathlib.Path) -> None: |
| 183 | init_repo_dirs(tmp_path) |
| 184 | files = [p for p in muse_dir(tmp_path).rglob("*") if p.is_file()] |
| 185 | assert files == [], f"Unexpected files written: {files}" |
| 186 | |
| 187 | def test_no_head_written(self, tmp_path: pathlib.Path) -> None: |
| 188 | init_repo_dirs(tmp_path) |
| 189 | assert not (muse_dir(tmp_path) / "HEAD").exists() |
| 190 | |
| 191 | def test_no_repo_json_written(self, tmp_path: pathlib.Path) -> None: |
| 192 | init_repo_dirs(tmp_path) |
| 193 | assert not (muse_dir(tmp_path) / "repo.json").exists() |
| 194 | |
| 195 | def test_no_extra_toplevel_dirs(self, tmp_path: pathlib.Path) -> None: |
| 196 | init_repo_dirs(tmp_path) |
| 197 | expected_names = { |
| 198 | "objects", "refs", "remotes", "logs", "shelf", |
| 199 | } |
| 200 | actual = {p.name for p in muse_dir(tmp_path).iterdir() if p.is_dir()} |
| 201 | unexpected = actual - expected_names |
| 202 | assert unexpected == set(), f"Unexpected dirs: {unexpected}" |
| 203 | |
| 204 | def test_existing_files_not_touched(self, tmp_path: pathlib.Path) -> None: |
| 205 | init_repo_dirs(tmp_path) |
| 206 | sentinel = objects_dir(tmp_path) / "sentinel" |
| 207 | sentinel.write_bytes(b"preserved") |
| 208 | init_repo_dirs(tmp_path) |
| 209 | assert sentinel.read_bytes() == b"preserved" |
| 210 | |
| 211 | |
| 212 | # --------------------------------------------------------------------------- |
| 213 | # Tier 6 — Security |
| 214 | # --------------------------------------------------------------------------- |
| 215 | |
| 216 | |
| 217 | class TestInitRepoDirsSecurity: |
| 218 | def test_symlink_root_does_not_escape(self, tmp_path: pathlib.Path) -> None: |
| 219 | """init_repo_dirs on a symlinked root creates dirs under the real target.""" |
| 220 | real = tmp_path / "real_repo" |
| 221 | real.mkdir() |
| 222 | link = tmp_path / "link_repo" |
| 223 | link.symlink_to(real) |
| 224 | init_repo_dirs(link) |
| 225 | assert muse_dir(real).is_dir() |
| 226 | assert not muse_dir(link).is_symlink() |
| 227 | |
| 228 | def test_root_with_spaces_in_path(self, tmp_path: pathlib.Path) -> None: |
| 229 | root = tmp_path / "my repo with spaces" |
| 230 | init_repo_dirs(root) |
| 231 | assert muse_dir(root).is_dir() |
| 232 | |
| 233 | def test_root_with_unicode_path(self, tmp_path: pathlib.Path) -> None: |
| 234 | root = tmp_path / "répertoire_音楽" |
| 235 | init_repo_dirs(root) |
| 236 | assert muse_dir(root).is_dir() |
| 237 | |
| 238 | |
| 239 | # --------------------------------------------------------------------------- |
| 240 | # Tier 7 — Performance |
| 241 | # --------------------------------------------------------------------------- |
| 242 | |
| 243 | |
| 244 | class TestInitRepoDirsPerformance: |
| 245 | def test_single_call_under_100ms(self, tmp_path: pathlib.Path) -> None: |
| 246 | root = tmp_path / "perf_repo" |
| 247 | start = time.perf_counter() |
| 248 | init_repo_dirs(root) |
| 249 | elapsed = time.perf_counter() - start |
| 250 | assert elapsed < 0.1, f"init_repo_dirs took {elapsed:.3f}s — expected < 0.1s" |
| 251 | |
| 252 | def test_idempotent_call_under_50ms(self, tmp_path: pathlib.Path) -> None: |
| 253 | init_repo_dirs(tmp_path) |
| 254 | start = time.perf_counter() |
| 255 | init_repo_dirs(tmp_path) |
| 256 | elapsed = time.perf_counter() - start |
| 257 | assert elapsed < 0.05, f"second call took {elapsed:.3f}s — expected < 0.05s" |