"""Exercise framework commands against separate vaults to prevent cross-repo writes.""" import json import os import shutil import subprocess import sys import tempfile import unittest from pathlib import Path FRAMEWORK = Path(__file__).resolve().parents[1] class ExternalVaultTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.base = Path(self.tmp.name) self.framework = self.base / "framework space" shutil.copytree(FRAMEWORK / "scripts", self.framework / "scripts") shutil.copytree(FRAMEWORK / "kb_framework", self.framework / "kb_framework", ignore=shutil.ignore_patterns("__pycache__")) shutil.copy2(FRAMEWORK / "Makefile", self.framework / "Makefile") self.vault = self.base / "vault space" self.other = self.base / "other" self.env = {k: v for k, v in os.environ.items() if k != "KNOWLEDGE_BASE_ROOT"} for root, name in [(self.framework, "framework"), (self.vault, "vault"), (self.other, "other")]: (root / "03_resources").mkdir(parents=True, exist_ok=True) (root / "03_resources" / "index.md").write_text( f"---\nid: {name}\ntitle: {name}\ntype: hub\n---\n# {name}\n" ) self.git(root, "init", "-q", "-b", "main") self.git(root, "config", "user.email", "test@example.com") self.git(root, "config", "user.name", "Test") self.git(root, "config", "commit.gpgsign", "false") self.git(root, "add", "03_resources/index.md") self.git(root, "commit", "-qm", "init") def git(self, root, *args): return subprocess.run(["git", *args], cwd=root, env=self.env, capture_output=True, text=True, check=True).stdout.strip() def run_script(self, name, *args, input="", env=None): return subprocess.run([sys.executable, str(self.framework / "scripts" / name), *map(str, args)], cwd=self.other, env=env or self.env, input=input, capture_output=True, text=True, timeout=15) def rpc(self, calls, *args, env=None): requests = [{"jsonrpc": "2.0", "id": i, "method": method, "params": params} for i, (method, params) in enumerate(calls)] proc = self.run_script("mcp-server.py", *args, env=env, input="".join(json.dumps(r) + "\n" for r in requests)) self.assertEqual(proc.returncode, 0, proc.stderr) responses = [json.loads(line) for line in proc.stdout.splitlines()] self.assertEqual(len(responses), len(calls), proc.stdout) for response in responses: self.assertNotIn("error", response, response) return [r["result"] for r in responses] def tool(self, name, arguments=None, root=None): return self.rpc([("tools/call", {"name": name, "arguments": arguments or {}})], "--vault-root", root or self.vault)[0] def test_generator_writes_only_external_vault(self): proc = self.run_script("generate-graph.py", "--vault-root", self.vault, "--write", "--validate") self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) graph = json.loads((self.vault / "index/graph.json").read_text()) self.assertEqual([n["id"] for n in graph["nodes"]], ["vault"]) self.assertFalse((self.framework / "index/graph.json").exists()) self.assertFalse((self.other / "index/graph.json").exists()) def test_instructions_are_discoverable_and_come_from_framework_without_graph(self): expected = {} for name in ["AGENTS.md", "AGENT_QUERY_GUIDE.md"]: expected[name] = f"# Framework {name}\nFramework-owned instructions.\n" (self.framework / name).write_text(expected[name]) (self.vault / name).write_text("# Unrelated vault file\n") listed = self.rpc([("tools/list", {})], "--vault-root", self.vault)[0] self.assertIn("read_instructions", {t["name"] for t in listed["tools"]}) result = self.tool("read_instructions") self.assertFalse(result.get("isError"), result) documents = result["structuredContent"]["documents"] self.assertEqual({d["path"]: d["content"] for d in documents}, expected) self.assertEqual(result["structuredContent"]["framework_root"], str(self.framework)) self.assertFalse((self.vault / "index/graph.json").exists()) def test_missing_framework_instructions_fail_explicitly_without_vault_fallback(self): (self.framework / "AGENTS.md").write_text("Read AGENT_QUERY_GUIDE.md\n") (self.vault / "AGENT_QUERY_GUIDE.md").write_text("Wrong instructions\n") result = self.tool("read_instructions") self.assertTrue(result.get("isError"), result) message = result["content"][0]["text"] self.assertIn(str(self.framework / "AGENT_QUERY_GUIDE.md"), message) def test_wiki_inventory_uses_external_vault(self): for root, target in [(self.vault, "vault-target"), (self.framework, "framework-target")]: (root / "03_resources/wiki.md").write_text(f"[[{target}]]\n") proc = self.run_script("list-wiki-links.py", "--vault-root", self.vault) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("vault-target", proc.stdout) self.assertNotIn("framework-target", proc.stdout) def test_mcp_reads_writes_and_commits_only_vault(self): framework_head = self.git(self.framework, "rev-parse", "HEAD") result = self.tool("write_note", {"path": "03_resources/new.md", "content": "# New\n"}) self.assertFalse(result.get("isError"), result) read = self.tool("read_note", {"path": "03_resources/new.md"}) self.assertEqual(read["structuredContent"]["content"], "# New\n") result = self.tool("git_commit", {"message": "add test note", "paths": ["03_resources/new.md"]}) self.assertFalse(result.get("isError"), result) self.assertEqual(self.git(self.vault, "log", "-1", "--format=%s"), "add test note") self.assertEqual(self.git(self.framework, "rev-parse", "HEAD"), framework_head) self.assertFalse((self.framework / "03_resources/new.md").exists()) for tool, args in [("read_note", {"path": "../framework space/03_resources/index.md"}), ("write_note", {"path": "../escape.md", "content": "escape"})]: self.assertTrue(self.tool(tool, args).get("isError")) def test_refresh_validates_in_memory_and_never_runs_vault_tools(self): (self.vault / "scripts").mkdir() (self.vault / "scripts/generate-graph.py").write_text("raise RuntimeError('vault tool used')\n") result = self.tool("refresh_graph", {"validate": True}) self.assertFalse(result.get("isError"), result) self.assertEqual(result["structuredContent"]["vault_root"], str(self.vault)) self.assertEqual(result["structuredContent"]["node_count"], 1) self.assertFalse((self.vault / "index/graph.json").exists()) summary = self.tool("graph_summary") self.assertFalse(summary.get("isError"), summary) def test_invalid_roots_fail_clearly_for_all_commands(self): file = self.base / "file" file.write_text("not a directory") for script in ["generate-graph.py", "list-wiki-links.py", "mcp-server.py"]: for root in [self.base / "missing", file, ""]: with self.subTest(script=script, root=root): proc = self.run_script(script, "--vault-root", root) self.assertNotEqual(proc.returncode, 0) self.assertIn("vault root", proc.stderr.lower()) self.assertNotIn("Traceback", proc.stderr) def test_root_precedence_and_standalone_for_all_commands(self): for root, name in [(self.framework, "framework"), (self.vault, "vault"), (self.other, "other")]: (root / "03_resources/wiki.md").write_text(f"[[{name}-target]]\n") for args, env, root, name in [([], self.env, self.framework, "framework"), ([], dict(self.env, KNOWLEDGE_BASE_ROOT=str(self.other)), self.other, "other"), (["--vault-root", str(self.vault)], dict(self.env, KNOWLEDGE_BASE_ROOT=str(self.other)), self.vault, "vault")]: with self.subTest(name=name): proc = self.run_script("generate-graph.py", *args, env=env) self.assertEqual(proc.returncode, 0, proc.stderr) graph = json.loads((root / "index/graph.json").read_text()) self.assertIn(name, [n["id"] for n in graph["nodes"]]) proc = self.run_script("list-wiki-links.py", *args, env=env) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn(f"{name}-target", proc.stdout) result = self.rpc([("tools/call", {"name": "read_note", "arguments": {"path": "03_resources/index.md"}})], *args, env=env)[0] self.assertIn(f"# {name}", result["structuredContent"]["content"]) def test_mcp_protocol_and_optional_readme_resource(self): self.tool("refresh_graph") results = self.rpc([("initialize", {}), ("tools/list", {}), ("resources/list", {})], "--vault-root", self.vault) self.assertEqual(results[0]["serverInfo"]["name"], "knowledge-base-mcp") self.assertIn("graph_summary", {t["name"] for t in results[1]["tools"]}) self.assertNotIn("repo://readme", {r["uri"] for r in results[2]["resources"]}) (self.vault / "README.md").write_text("# Optional hub\n") result = self.rpc([("resources/list", {})], "--vault-root", self.vault)[0] self.assertIn("repo://readme", {r["uri"] for r in result["resources"]}) def test_git_refuses_parent_repository_for_nested_non_repo_vault(self): nested = self.framework / "nested" nested.mkdir() (nested / "note.md").write_text("# note\n") before = self.git(self.framework, "rev-parse", "HEAD") result = self.tool("git_commit", {"message": "must fail", "paths": ["note.md"]}, root=nested) self.assertTrue(result.get("isError"), result) self.assertEqual(self.git(self.framework, "rev-parse", "HEAD"), before) def test_make_and_installed_hook_use_selected_vault(self): def make(*args): return subprocess.run(["make", "-C", str(self.framework), f"VAULT_ROOT={self.vault}", *args], env=self.env, capture_output=True, text=True) proc = make("validate-graph", "install-hooks") self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) hook = self.vault / ".git/hooks/pre-commit" self.assertTrue(hook.exists()) proc = subprocess.run([str(hook)], cwd=self.other, env=self.env, capture_output=True, text=True) self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr) (self.vault / "03_resources/new.md").write_text("# new\n[gone](nowhere)\n") proc = subprocess.run([str(hook)], cwd=self.other, env=self.env, capture_output=True, text=True) self.assertNotEqual(proc.returncode, 0) self.assertIn("nowhere", proc.stdout + proc.stderr) self.assertFalse((self.vault / "index/graph.json").exists()) self.assertFalse((self.framework / "index/graph.json").exists()) def test_installer_preserves_unmanaged_hooks_and_reinstalls_managed_hook(self): hook = self.vault / ".git/hooks/pre-commit" custom = "#!/bin/sh\n# user's hook\nexit 0\n" hook.write_text(custom) proc = self.run_script("install-hooks.py", "--vault-root", self.vault) self.assertNotEqual(proc.returncode, 0) self.assertIn("unmanaged hook", proc.stderr) self.assertEqual(hook.read_text(), custom) hook.unlink() (self.vault / "scripts").mkdir() target = self.vault / "scripts/pre-commit" target.write_text(custom) hook.symlink_to("../../scripts/pre-commit") for _ in range(2): proc = self.run_script("install-hooks.py", "--vault-root", self.vault) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertFalse(hook.is_symlink()) self.assertEqual(target.read_text(), custom) def test_installer_honors_worktree_and_custom_hooks_directory(self): worktree = self.base / "worktree" self.git(self.vault, "worktree", "add", "-qb", "worktree", str(worktree)) proc = self.run_script("install-hooks.py", "--vault-root", worktree) self.assertEqual(proc.returncode, 0, proc.stderr) hook = Path(self.git(worktree, "rev-parse", "--path-format=absolute", "--git-path", "hooks/pre-commit")) self.assertIn(str(worktree), hook.read_text()) custom = self.base / "custom hooks" self.git(self.vault, "config", "core.hooksPath", str(custom)) proc = self.run_script("install-hooks.py", "--vault-root", self.vault) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn(str(self.vault), (custom / "pre-commit").read_text()) if __name__ == "__main__": unittest.main()