"""TDD — Phase 7: msgpack confined to wire-protocol files only. Phase 7 requirements (issue #12): - `import msgpack` must appear ONLY in wire-protocol files (transport, object store, gc, migration, bundle, pack/unpack commands). - Every file in the allowlist must exist (guard against stale allowlist). - No file under muse/core/, muse/plugins/, or muse/cli/commands/ outside the allowlist may import the msgpack library. """ from __future__ import annotations import ast import pathlib # --------------------------------------------------------------------------- # Wire-only allowlist — files that legitimately import msgpack # --------------------------------------------------------------------------- _REPO_ROOT = pathlib.Path(__file__).parent.parent _WIRE_ALLOWLIST: frozenset[pathlib.Path] = frozenset({ _REPO_ROOT / "muse" / "core" / "io.py", _REPO_ROOT / "muse" / "core" / "gc.py", _REPO_ROOT / "muse" / "core" / "transport.py", _REPO_ROOT / "muse" / "core" / "migrate.py", _REPO_ROOT / "muse" / "core" / "mpack.py", _REPO_ROOT / "muse" / "cli" / "commands" / "bundle.py", _REPO_ROOT / "muse" / "cli" / "commands" / "pack_objects.py", _REPO_ROOT / "muse" / "cli" / "commands" / "unpack_objects.py", _REPO_ROOT / "muse" / "cli" / "commands" / "verify_pack.py", }) def _imports_msgpack(path: pathlib.Path) -> bool: """Return True if *path* contains any import of the msgpack library.""" try: tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) except SyntaxError: return False for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: if alias.name == "msgpack" or alias.name.startswith("msgpack."): return True elif isinstance(node, ast.ImportFrom): if node.module and ( node.module == "msgpack" or node.module.startswith("msgpack.") ): return True return False # --------------------------------------------------------------------------- # Test: allowlist files all exist # --------------------------------------------------------------------------- class TestWireAllowlistFilesExist: def test_all_allowlist_files_exist(self) -> None: """Every file in the wire allowlist must actually exist on disk.""" missing = [p for p in _WIRE_ALLOWLIST if not p.is_file()] assert missing == [], ( f"Stale allowlist — these files no longer exist: " + ", ".join(str(p.relative_to(_REPO_ROOT)) for p in missing) ) # --------------------------------------------------------------------------- # Test: no msgpack in muse/core/ (except allowlist) # --------------------------------------------------------------------------- class TestNoMsgpackInCore: def test_no_msgpack_import_in_core_storage(self) -> None: """No file in muse/core/ outside the allowlist may import msgpack.""" violations: list[str] = [] for path in (_REPO_ROOT / "muse" / "core").rglob("*.py"): if path in _WIRE_ALLOWLIST: continue if _imports_msgpack(path): violations.append(str(path.relative_to(_REPO_ROOT))) assert violations == [], ( "Non-wire muse/core/ files import msgpack:\n " + "\n ".join(sorted(violations)) ) def test_no_msgpack_import_in_plugins(self) -> None: """No file in muse/plugins/ may import msgpack.""" violations: list[str] = [] for path in (_REPO_ROOT / "muse" / "plugins").rglob("*.py"): if path in _WIRE_ALLOWLIST: continue if _imports_msgpack(path): violations.append(str(path.relative_to(_REPO_ROOT))) assert violations == [], ( "Plugin files import msgpack:\n " + "\n ".join(sorted(violations)) ) # --------------------------------------------------------------------------- # Test: no msgpack in muse/cli/commands/ (except allowlist) # --------------------------------------------------------------------------- class TestNoMsgpackInCliCommands: def test_no_msgpack_import_in_cli_storage(self) -> None: """No cli/commands/ file outside the wire allowlist may import msgpack.""" violations: list[str] = [] for path in (_REPO_ROOT / "muse" / "cli" / "commands").rglob("*.py"): if path in _WIRE_ALLOWLIST: continue if _imports_msgpack(path): violations.append(str(path.relative_to(_REPO_ROOT))) assert violations == [], ( "Non-wire cli/commands/ files import msgpack:\n " + "\n ".join(sorted(violations)) )