"""Section XX — No Legacy Flat-Store: regression guard for the per-repo migration. Every test in this file asserts the ABSENCE of legacy flat-store code. They all start RED and turn GREEN as the legacy code is deleted. Their permanent role is to prevent regressions — if any of these tests starts failing, legacy code was accidentally reintroduced. What "legacy flat-store" means: - LocalBackend._safe_id() — converts "sha256:abc" → "sha256_abc" for flat naming - LocalBackend._root — the /data/musehub global objects directory - LocalBackend._path(oid, repo_root=None) falling back to the flat store - put/get/exists/delete without repo_root silently using the flat store - settings.musehub_objects_dir (/data/musehub) — superseded by musehub_repos_dir - musehub_sync._object_disk_path() — built old / paths - musehub/api/routes/musehub/objects.py validating disk paths against musehub_objects_dir instead of musehub_repos_dir Per-repo store (current canonical layout): /data/repos///objects/sha256/<2-hex>/<62-hex> This file intentionally does NOT test the positive behaviour (that is already covered by test_storage_backends.py and test_storage_tiers.py). It tests only the negative: that the old code paths are gone. """ from __future__ import annotations import ast import inspect import textwrap from pathlib import Path import pytest from muse.core.types import long_id # ── helpers ─────────────────────────────────────────────────────────────────── _REPO_ROOT = Path(__file__).parent.parent # ~/ecosystem/musehub def _source(rel: str) -> str: return (_REPO_ROOT / rel).read_text() def _ast_names(rel: str) -> set[str]: """Return all function/class names defined at module level.""" tree = ast.parse(_source(rel)) return {n.name for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef))} # ═══════════════════════════════════════════════════════════════════════════════ # 1. LocalBackend — no _safe_id, no _root, no flat-store fallback in _path # ═══════════════════════════════════════════════════════════════════════════════ class TestLocalBackendNoLegacy: def test_no_safe_id_method(self) -> None: """_safe_id() must be deleted — it exists only to sanitise flat-store filenames.""" from musehub.storage.backends import LocalBackend assert not hasattr(LocalBackend, "_safe_id"), ( "LocalBackend._safe_id still exists. Delete it — it was only needed " "for the legacy flat-store naming scheme (sha256:abc → sha256_abc)." ) def test_no_root_attribute(self, tmp_path: Path) -> None: """_root was the global /data/musehub objects directory — must be gone.""" from musehub.storage.backends import LocalBackend b = LocalBackend() assert not hasattr(b, "_root"), ( "LocalBackend._root still exists. It pointed at the legacy global " "flat-store directory. Remove it and its __init__ assignment." ) def test_init_takes_no_objects_dir_arg(self, tmp_path: Path) -> None: """__init__(objects_dir=...) was for the flat store — no longer needed.""" from musehub.storage.backends import LocalBackend import inspect sig = inspect.signature(LocalBackend.__init__) assert "objects_dir" not in sig.parameters, ( "LocalBackend.__init__ still accepts 'objects_dir'. That parameter " "was used to set self._root for the flat store. Remove it." ) def test_path_without_repo_root_raises(self, tmp_path: Path) -> None: """_path(oid, repo_root=None) must raise — no silent flat-store fallback.""" from musehub.storage.backends import LocalBackend b = LocalBackend() with pytest.raises((ValueError, TypeError, AssertionError)): b._path(long_id("a" * 64), repo_root=None) def test_path_with_repo_root_uses_algo_shard_layout(self, tmp_path: Path) -> None: """Per-repo path must be /objects/sha256/<2hex>/<62hex>.""" from musehub.storage.backends import LocalBackend b = LocalBackend() repo_root = tmp_path / "repos" / "gabriel" / "muse" hex64 = "ab" + "c" * 62 oid = long_id(hex64) p = b._path(oid, repo_root=repo_root) expected = repo_root / "objects" / "sha256" / "ab" / ("c" * 62) assert p == expected, ( f"Expected per-repo path {expected} but got {p}. " "LocalBackend._path must produce algo/shard/rest layout." ) async def test_put_without_repo_root_raises(self, tmp_path: Path) -> None: """put() without repo_root must raise — no silent flat-store write.""" from musehub.storage.backends import LocalBackend b = LocalBackend() with pytest.raises((ValueError, TypeError, AssertionError)): await b.put(long_id("a" * 64), b"data", repo_root=None) async def test_get_without_repo_root_raises(self, tmp_path: Path) -> None: """get() without repo_root must raise — no silent flat-store read.""" from musehub.storage.backends import LocalBackend b = LocalBackend() with pytest.raises((ValueError, TypeError, AssertionError)): await b.get(long_id("a" * 64), repo_root=None) async def test_exists_without_repo_root_raises(self, tmp_path: Path) -> None: """exists() without repo_root must raise — no silent flat-store check.""" from musehub.storage.backends import LocalBackend b = LocalBackend() with pytest.raises((ValueError, TypeError, AssertionError)): await b.exists(long_id("a" * 64), repo_root=None) async def test_per_repo_roundtrip(self, tmp_path: Path) -> None: """Full put/get/exists/delete cycle with explicit repo_root works.""" from musehub.storage.backends import LocalBackend b = LocalBackend() repo_root = tmp_path / "repos" / "gabriel" / "muse" hex64 = "ab" + "c" * 62 oid = long_id(hex64) data = b"per-repo content" uri = await b.put(oid, data, repo_root=repo_root) assert uri.startswith("local://") assert await b.exists(oid, repo_root=repo_root) is True assert await b.get(oid, repo_root=repo_root) == data await b.delete(oid, repo_root=repo_root) assert await b.exists(oid, repo_root=repo_root) is False async def test_per_repo_object_on_disk_at_correct_path(self, tmp_path: Path) -> None: """Object bytes land at objects/sha256/<2hex>/<62hex>, not at a flat path.""" from musehub.storage.backends import LocalBackend b = LocalBackend() repo_root = tmp_path / "repos" / "gabriel" / "muse" hex64 = "de" + "f" * 62 oid = long_id(hex64) await b.put(oid, b"content", repo_root=repo_root) expected = repo_root / "objects" / "sha256" / "de" / ("f" * 62) assert expected.exists(), ( f"Object not found at expected per-repo path {expected}. " "Check LocalBackend._path algo/shard layout." ) def test_no_flat_path_in_source(self) -> None: """backends.py must not reference the flat legacy path patterns.""" src = _source("musehub/storage/backends.py") forbidden = [ "_safe_id", "musehub_objects_dir", "self._root", ] for pattern in forbidden: assert pattern not in src, ( f"Found '{pattern}' in backends.py — legacy flat-store remnant. " "Delete it." ) # ═══════════════════════════════════════════════════════════════════════════════ # 2. Config — musehub_objects_dir removed, musehub_repos_dir present # ═══════════════════════════════════════════════════════════════════════════════ class TestConfigNoLegacy: def test_no_musehub_objects_dir_in_settings(self) -> None: """settings.musehub_objects_dir must be removed — superseded by musehub_repos_dir.""" from musehub.config import Settings assert "musehub_objects_dir" not in Settings.model_fields, ( "Settings.musehub_objects_dir still exists. Remove it — the flat-store " "root (/data/musehub) is no longer used. Only musehub_repos_dir remains." ) def test_musehub_repos_dir_present(self) -> None: """settings.musehub_repos_dir must exist — this is the canonical store root.""" from musehub.config import Settings assert "musehub_repos_dir" in Settings.model_fields, ( "Settings.musehub_repos_dir is missing. It must be present and default " "to '/data/repos'." ) def test_no_objects_dir_in_config_source(self) -> None: """config.py must not define musehub_objects_dir.""" src = _source("musehub/config.py") assert "musehub_objects_dir" not in src, ( "Found 'musehub_objects_dir' in config.py. Delete it." ) # ═══════════════════════════════════════════════════════════════════════════════ # 3. musehub_sync — no _object_disk_path, no flat-store writes # ═══════════════════════════════════════════════════════════════════════════════ class TestMusehubSyncNoLegacy: def test_no_object_disk_path_function(self) -> None: """_object_disk_path() built the old / flat path — must be gone.""" from musehub.services import musehub_sync assert not hasattr(musehub_sync, "_object_disk_path"), ( "musehub_sync._object_disk_path still exists. Delete it — it computed " "the legacy /data/musehub// path." ) def test_no_musehub_objects_dir_in_sync_source(self) -> None: """musehub_sync.py must not import or reference musehub_objects_dir.""" src = _source("musehub/services/musehub_sync.py") assert "musehub_objects_dir" not in src, ( "Found 'musehub_objects_dir' in musehub_sync.py. Remove it — objects " "are now written via LocalBackend with an explicit repo_root." ) def test_no_safe_id_in_sync_source(self) -> None: """No manual safe_id conversion in sync — that was for flat-store filenames.""" src = _source("musehub/services/musehub_sync.py") # Only flag the colon-replacement pattern used for flat filenames assert 'replace(":", "-")' not in src and 'replace(":", "_")' not in src, ( "Found legacy colon-replacement in musehub_sync.py. Delete it — " "_object_disk_path was the only caller of this pattern." ) # ═══════════════════════════════════════════════════════════════════════════════ # 4. objects route — validates disk path against musehub_repos_dir, not objects_dir # ═══════════════════════════════════════════════════════════════════════════════ class TestObjectsRouteNoLegacy: def test_no_musehub_objects_dir_in_objects_route(self) -> None: """The objects download route must not reference musehub_objects_dir.""" src = _source("musehub/api/routes/musehub/objects.py") assert "musehub_objects_dir" not in src, ( "Found 'musehub_objects_dir' in objects.py. The security check must " "validate disk paths against musehub_repos_dir instead." ) def test_objects_route_validates_against_repos_dir(self) -> None: """The path-traversal guard in the objects route must use musehub_repos_dir.""" src = _source("musehub/api/routes/musehub/objects.py") assert "musehub_repos_dir" in src, ( "objects.py does not reference musehub_repos_dir. The disk_path " "security check must validate that paths stay inside musehub_repos_dir " "(/data/repos), not the old musehub_objects_dir (/data/musehub)." ) # ═══════════════════════════════════════════════════════════════════════════════ # 5. storage/__init__.py docstring — no flat-store description # ═══════════════════════════════════════════════════════════════════════════════ class TestStorageInitNoLegacy: def test_no_flat_store_description_in_init(self) -> None: """storage/__init__.py docstring must not describe the legacy flat layout.""" src = _source("musehub/storage/__init__.py") assert "musehub_objects_dir" not in src, ( "Found 'musehub_objects_dir' in storage/__init__.py. Update the docstring." )