228 lines
13 KiB
Python
228 lines
13 KiB
Python
"""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.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_uses_absolute_framework_generator_and_writes_when_validating(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)
|
|
command = result["structuredContent"]["command"]
|
|
self.assertEqual(command[1], str(self.framework / "scripts/generate-graph.py"))
|
|
self.assertEqual(command[command.index("--vault-root") + 1], str(self.vault))
|
|
self.assertTrue((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("graph", "install-hooks")
|
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
|
hook = self.vault / ".git/hooks/pre-commit"
|
|
self.assertTrue(hook.exists())
|
|
self.git(self.vault, "add", "index/graph.json")
|
|
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")
|
|
proc = subprocess.run([str(hook)], cwd=self.other, env=self.env, capture_output=True, text=True)
|
|
self.assertNotEqual(proc.returncode, 0)
|
|
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()
|