"""Tests for ``muse.core.paths.init_repo_dirs``. Seven tiers: Tier 1 — Unit: each directory helper, return value, idempotency Tier 2 — Integration: dirs usable by store/commit/snapshot layer Tier 3 — End-to-end: full CLI init flow uses init_repo_dirs layout Tier 4 — Stress: concurrent calls, large number of calls Tier 5 — State integrity: no files written, no extra dirs created Tier 6 — Security: symlink attack, path traversal, permission edge cases Tier 7 — Performance: single call completes well under threshold """ from __future__ import annotations import concurrent.futures import pathlib import threading import time from collections.abc import Callable import pytest from muse.core.types import long_id from muse.core.paths import ( commits_dir, heads_dir, init_repo_dirs, logs_dir, muse_dir, objects_dir, remotes_dir, shelf_dir, snapshots_dir, ) # --------------------------------------------------------------------------- # Tier 1 — Unit # --------------------------------------------------------------------------- class TestInitRepoDirsUnit: def test_returns_root(self, tmp_path: pathlib.Path) -> None: assert init_repo_dirs(tmp_path) is tmp_path def test_muse_dir_created(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) assert muse_dir(tmp_path).is_dir() @pytest.mark.parametrize("dir_fn", [ objects_dir, heads_dir, remotes_dir, logs_dir, shelf_dir, ]) def test_each_subdirectory_created(self, tmp_path: pathlib.Path, dir_fn: Callable[[pathlib.Path], pathlib.Path]) -> None: init_repo_dirs(tmp_path) assert dir_fn(tmp_path).is_dir() def test_idempotent_second_call_no_raise(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) init_repo_dirs(tmp_path) # must not raise FileExistsError def test_idempotent_dirs_still_present(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) init_repo_dirs(tmp_path) assert muse_dir(tmp_path).is_dir() assert objects_dir(tmp_path).is_dir() def test_creates_root_if_missing(self, tmp_path: pathlib.Path) -> None: root = tmp_path / "new_repo" assert not root.exists() init_repo_dirs(root) assert muse_dir(root).is_dir() def test_creates_nested_missing_root(self, tmp_path: pathlib.Path) -> None: root = tmp_path / "a" / "b" / "c" init_repo_dirs(root) assert muse_dir(root).is_dir() # --------------------------------------------------------------------------- # Tier 2 — Integration: dirs are usable by the object/commit/snapshot layers # --------------------------------------------------------------------------- class TestInitRepoDirsIntegration: def test_objects_dir_accepts_write(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) obj = objects_dir(tmp_path) / "test_blob" obj.write_bytes(b"hello") assert obj.read_bytes() == b"hello" def test_commits_dir_accepts_write(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) f = commits_dir(tmp_path) / "sha256" / "ab" f.mkdir(parents=True) assert f.is_dir() def test_heads_dir_accepts_ref_file(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) ref = heads_dir(tmp_path) / "main" ref.write_text(long_id("a" * 64)) assert ref.read_text().startswith("sha256:") def test_logs_dir_accepts_subdirs(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) sub = logs_dir(tmp_path) / "refs" / "heads" sub.mkdir(parents=True) assert sub.is_dir() def test_shelf_dir_accepts_msgpack_file(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) entry = shelf_dir(tmp_path) / "entry.msgpack" entry.write_bytes(b"\x82\xa3key\xa5value") assert entry.exists() # --------------------------------------------------------------------------- # Tier 3 — End-to-end: CLI init uses same layout # --------------------------------------------------------------------------- class TestInitRepoDirsE2E: def test_cli_init_produces_same_dirs(self, muse_repo: pathlib.Path) -> None: """muse init creates at minimum the dirs that init_repo_dirs promises.""" expected = [ muse_dir, objects_dir, heads_dir, remotes_dir, logs_dir, shelf_dir, ] for dir_fn in expected: assert dir_fn(muse_repo).is_dir(), f"{dir_fn.__name__} missing after muse init" def test_bare_repo_dirs_subset_of_full_init(self, tmp_path: pathlib.Path, muse_repo: pathlib.Path) -> None: """Every dir init_repo_dirs creates also exists in a CLI-init repo.""" init_repo_dirs(tmp_path) for p in muse_dir(tmp_path).rglob("*"): if p.is_dir(): rel = p.relative_to(tmp_path) assert (muse_repo / rel).is_dir(), f"{rel} missing in CLI repo" # --------------------------------------------------------------------------- # Tier 4 — Stress # --------------------------------------------------------------------------- class TestInitRepoDirsStress: def test_100_sequential_calls_idempotent(self, tmp_path: pathlib.Path) -> None: for _ in range(100): init_repo_dirs(tmp_path) assert muse_dir(tmp_path).is_dir() def test_concurrent_calls_safe(self, tmp_path: pathlib.Path) -> None: errors: list[Exception] = [] def call() -> None: try: init_repo_dirs(tmp_path) except Exception as exc: errors.append(exc) threads = [threading.Thread(target=call) for _ in range(20)] for t in threads: t.start() for t in threads: t.join() assert errors == [], f"concurrent calls raised: {errors}" assert muse_dir(tmp_path).is_dir() def test_threadpool_concurrent_different_roots(self, tmp_path: pathlib.Path) -> None: roots = [tmp_path / f"repo_{i}" for i in range(10)] with concurrent.futures.ThreadPoolExecutor(max_workers=10) as ex: list(ex.map(init_repo_dirs, roots)) for root in roots: assert muse_dir(root).is_dir() # --------------------------------------------------------------------------- # Tier 5 — State integrity # --------------------------------------------------------------------------- class TestInitRepoDirsStateIntegrity: def test_no_files_written(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) files = [p for p in muse_dir(tmp_path).rglob("*") if p.is_file()] assert files == [], f"Unexpected files written: {files}" def test_no_head_written(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) assert not (muse_dir(tmp_path) / "HEAD").exists() def test_no_repo_json_written(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) assert not (muse_dir(tmp_path) / "repo.json").exists() def test_no_extra_toplevel_dirs(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) expected_names = { "objects", "refs", "remotes", "logs", "shelf", } actual = {p.name for p in muse_dir(tmp_path).iterdir() if p.is_dir()} unexpected = actual - expected_names assert unexpected == set(), f"Unexpected dirs: {unexpected}" def test_existing_files_not_touched(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) sentinel = objects_dir(tmp_path) / "sentinel" sentinel.write_bytes(b"preserved") init_repo_dirs(tmp_path) assert sentinel.read_bytes() == b"preserved" # --------------------------------------------------------------------------- # Tier 6 — Security # --------------------------------------------------------------------------- class TestInitRepoDirsSecurity: def test_symlink_root_does_not_escape(self, tmp_path: pathlib.Path) -> None: """init_repo_dirs on a symlinked root creates dirs under the real target.""" real = tmp_path / "real_repo" real.mkdir() link = tmp_path / "link_repo" link.symlink_to(real) init_repo_dirs(link) assert muse_dir(real).is_dir() assert not muse_dir(link).is_symlink() def test_root_with_spaces_in_path(self, tmp_path: pathlib.Path) -> None: root = tmp_path / "my repo with spaces" init_repo_dirs(root) assert muse_dir(root).is_dir() def test_root_with_unicode_path(self, tmp_path: pathlib.Path) -> None: root = tmp_path / "répertoire_音楽" init_repo_dirs(root) assert muse_dir(root).is_dir() # --------------------------------------------------------------------------- # Tier 7 — Performance # --------------------------------------------------------------------------- class TestInitRepoDirsPerformance: def test_single_call_under_100ms(self, tmp_path: pathlib.Path) -> None: root = tmp_path / "perf_repo" start = time.perf_counter() init_repo_dirs(root) elapsed = time.perf_counter() - start assert elapsed < 0.1, f"init_repo_dirs took {elapsed:.3f}s — expected < 0.1s" def test_idempotent_call_under_50ms(self, tmp_path: pathlib.Path) -> None: init_repo_dirs(tmp_path) start = time.perf_counter() init_repo_dirs(tmp_path) elapsed = time.perf_counter() - start assert elapsed < 0.05, f"second call took {elapsed:.3f}s — expected < 0.05s"