267 lines
13 KiB
Python
267 lines
13 KiB
Python
from __future__ import annotations
|
|
|
|
from io import StringIO
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import unittest
|
|
import uuid
|
|
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from labyricorn_mcp.domain import LabyricornDomain
|
|
from labyricorn_mcp.errors import LabyricornMcpError
|
|
from labyricorn_mcp.gitops import GitOperations
|
|
from labyricorn_mcp.safety import RepositoryGuard
|
|
from labyricorn_mcp.server import StdioMcpServer
|
|
from labyricorn_mcp.tools import ToolRegistry
|
|
from labyricorn_mcp.validation import ValidationOperations
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class McpRepositoryFixture(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.base = ROOT / ".cache" / f"mcp-test-{uuid.uuid4().hex}"
|
|
self.site = self.base / "site"
|
|
for section, model in (("articles", "section"), ("blog", "section"), ("projects", "section")):
|
|
directory = self.site / "content" / section
|
|
directory.mkdir(parents=True)
|
|
(directory / "contents.lr").write_text(
|
|
f"_model: {model}\n---\ntitle: {section.title()}\n", encoding="utf-8"
|
|
)
|
|
tags = self.site / "content" / "tags"
|
|
(tags / "writing").mkdir(parents=True)
|
|
(tags / "contents.lr").write_text("_model: tag-index\n---\ntitle: Tags\n", encoding="utf-8")
|
|
(tags / "writing" / "contents.lr").write_text(
|
|
"_model: tag\n---\ntitle: Writing\n---\nsummary: Writing and editing.\n", encoding="utf-8"
|
|
)
|
|
(self.site / "models").mkdir()
|
|
for name in ("entry.ini", "tag.ini", "project.ini", "devlog.ini", "devlog-entry.ini", "section.ini", "tag-index.ini"):
|
|
shutil.copy2(ROOT / "models" / name, self.site / "models" / name)
|
|
(self.site / "configs").mkdir()
|
|
(self.site / "configs" / "project-sources.ini").write_text(
|
|
"[example]\nrepository = https://example.test/example.git\nweb_url = https://example.test/example\napi_url = https://example.test/api/example\nbranch = main\n",
|
|
encoding="utf-8",
|
|
)
|
|
(self.site / "scripts").mkdir()
|
|
for name in (
|
|
"article_editor.py",
|
|
"blog_editor.py",
|
|
"git_repo.py",
|
|
"narration_import.py",
|
|
"project_editor.py",
|
|
"project_providers.py",
|
|
"tag_editor.py",
|
|
):
|
|
shutil.copy2(ROOT / "scripts" / name, self.site / "scripts" / name)
|
|
(self.site / "scripts" / "build_with_projects.py").write_text("raise SystemExit(0)\n", encoding="utf-8")
|
|
(self.site / "templates").mkdir()
|
|
(self.site / "templates" / "base.html").write_text("<main>Test</main>\n", encoding="utf-8")
|
|
(self.site / "assets" / "static").mkdir(parents=True)
|
|
(self.site / "assets" / "static" / "style.css").write_text("body {}\n", encoding="utf-8")
|
|
(self.site / "Labyricorn.lektorproject").write_text(
|
|
"[project]\nname = Fixture\nurl = https://example.test\n", encoding="utf-8"
|
|
)
|
|
self.git("init", "--initial-branch=main")
|
|
self.git("config", "user.name", "MCP Test")
|
|
self.git("config", "user.email", "[email protected]")
|
|
self.git("add", ".")
|
|
self.git("commit", "-m", "Initial fixture")
|
|
self.guard = RepositoryGuard(self.site)
|
|
self.domain = LabyricornDomain(self.guard)
|
|
|
|
def tearDown(self) -> None:
|
|
shutil.rmtree(self.base, onexc=self._make_writable)
|
|
|
|
@staticmethod
|
|
def _make_writable(function: object, path: str, _error: BaseException) -> None:
|
|
os.chmod(path, 0o700)
|
|
function(path)
|
|
|
|
def git(self, *args: str) -> subprocess.CompletedProcess[str]:
|
|
result = subprocess.run(
|
|
["git", *args], cwd=self.site, capture_output=True, text=True, encoding="utf-8", check=False
|
|
)
|
|
if result.returncode:
|
|
raise AssertionError(result.stderr or result.stdout)
|
|
return result
|
|
|
|
@staticmethod
|
|
def article_fields(**changes: object) -> dict[str, object]:
|
|
fields: dict[str, object] = {
|
|
"title": "MCP Test Article",
|
|
"date": "2026-08-23",
|
|
"author": "Test",
|
|
"tags": ["writing", "Local Experiment"],
|
|
"summary": "A test article.",
|
|
"body": "Test body.",
|
|
}
|
|
fields.update(changes)
|
|
return fields
|
|
|
|
|
|
class RepositorySafetyTests(McpRepositoryFixture):
|
|
def test_repository_detection_rejects_wrong_directory(self) -> None:
|
|
wrong = self.base / "wrong"
|
|
wrong.mkdir()
|
|
with self.assertRaisesRegex(LabyricornMcpError, "compatible Labyricorn"):
|
|
RepositoryGuard(wrong)
|
|
|
|
def test_path_traversal_and_absolute_paths_are_rejected(self) -> None:
|
|
for value in ("../outside.txt", str((self.base / "outside.txt").resolve())):
|
|
with self.subTest(value=value), self.assertRaisesRegex(LabyricornMcpError, "repository-relative"):
|
|
self.guard.resolve(value)
|
|
|
|
def test_symlink_escape_is_rejected(self) -> None:
|
|
outside = self.base / "outside.txt"
|
|
outside.write_text("outside", encoding="utf-8")
|
|
link = self.site / "templates" / "escape.html"
|
|
try:
|
|
link.symlink_to(outside)
|
|
except OSError as exc:
|
|
self.skipTest(f"symlink creation is unavailable: {exc}")
|
|
with self.assertRaisesRegex(LabyricornMcpError, "outside the repository"):
|
|
self.guard.resolve("templates/escape.html", allowed_roots=("templates",), must_exist=True)
|
|
|
|
|
|
class ContentDomainTests(McpRepositoryFixture):
|
|
def test_content_create_read_update_and_conflict(self) -> None:
|
|
created = self.domain.create_article("mcp-test", self.article_fields())
|
|
self.assertEqual("writing, local-experiment", created["fields"]["tags"])
|
|
loaded = self.domain.get_content("article", "mcp-test")
|
|
updated = self.domain.update_article(
|
|
"mcp-test", {"summary": "Updated safely."}, loaded["sha256"]
|
|
)
|
|
self.assertEqual("Updated safely.", updated["fields"]["summary"])
|
|
with self.assertRaisesRegex(LabyricornMcpError, "changed after it was read"):
|
|
self.domain.update_article("mcp-test", {"summary": "Stale"}, loaded["sha256"])
|
|
|
|
def test_malformed_content_is_rejected(self) -> None:
|
|
created = self.domain.create_article("mcp-test", self.article_fields())
|
|
with (self.site / created["path"]).open("a", encoding="utf-8") as output:
|
|
output.write("---\ntitle: Duplicate\n")
|
|
with self.assertRaises(LabyricornMcpError):
|
|
self.domain.get_content("article", "mcp-test")
|
|
|
|
def test_model_description_is_derived_and_invalid_model_rejected(self) -> None:
|
|
model = self.domain.describe_model("entry")
|
|
self.assertIn("title", {field["name"] for field in model["fields"]})
|
|
self.assertEqual("models/entry.ini", model["source"])
|
|
with self.assertRaisesRegex(LabyricornMcpError, "Model names"):
|
|
self.domain.describe_model("../entry")
|
|
|
|
def test_design_update_requires_matching_hash(self) -> None:
|
|
loaded = self.domain.get_design_file("templates/base.html")
|
|
updated = self.domain.update_design_file(
|
|
"templates/base.html", "<main>Updated</main>\n", loaded["sha256"]
|
|
)
|
|
self.assertNotEqual(loaded["sha256"], updated["sha256"])
|
|
with self.assertRaisesRegex(LabyricornMcpError, "changed after it was read"):
|
|
self.domain.update_design_file("templates/base.html", "stale", loaded["sha256"])
|
|
|
|
|
|
class GitSafetyTests(McpRepositoryFixture):
|
|
def test_dirty_state_hash_detects_intervening_change(self) -> None:
|
|
operations = GitOperations(self.guard)
|
|
first = self.site / "first.txt"
|
|
first.write_text("first\n", encoding="utf-8")
|
|
status = operations.status()
|
|
(self.site / "second.txt").write_text("second\n", encoding="utf-8")
|
|
with self.assertRaisesRegex(LabyricornMcpError, "changed after it was reviewed"):
|
|
operations.commit("Test", ["first.txt"], status["status_sha256"])
|
|
|
|
def test_commit_is_limited_to_reviewed_paths(self) -> None:
|
|
operations = GitOperations(self.guard)
|
|
(self.site / "new").mkdir()
|
|
(self.site / "new" / "first.txt").write_text("first\n", encoding="utf-8")
|
|
(self.site / "second.txt").write_text("second\n", encoding="utf-8")
|
|
status = operations.status()
|
|
result = operations.commit("Commit one file", ["new/first.txt"], status["status_sha256"])
|
|
self.assertEqual(["new/first.txt"], result["paths"])
|
|
self.assertIn("second.txt", {item["path"] for item in result["status"]["changes"]})
|
|
self.assertEqual("Commit one file", self.git("log", "-1", "--format=%s").stdout.strip())
|
|
|
|
def test_sensitive_file_cannot_be_committed(self) -> None:
|
|
operations = GitOperations(self.guard)
|
|
(self.site / ".env").write_text("TOKEN=secret\n", encoding="utf-8")
|
|
status = operations.status()
|
|
with self.assertRaisesRegex(LabyricornMcpError, "not eligible"):
|
|
operations.commit("Do not commit", [".env"], status["status_sha256"])
|
|
|
|
|
|
class ProtocolAndValidationTests(McpRepositoryFixture):
|
|
def test_tool_registration_has_focused_surface_and_no_shell_or_deploy(self) -> None:
|
|
names = {tool["name"] for tool in ToolRegistry(self.guard).tools}
|
|
self.assertTrue({"site_info", "create_article", "validate_changes", "push_changes"} <= names)
|
|
self.assertFalse({"shell", "exec", "deploy", "publish_site"} & names)
|
|
|
|
def test_mutation_reports_unrelated_worktree_context(self) -> None:
|
|
(self.site / "unrelated.txt").write_text("keep me\n", encoding="utf-8")
|
|
result = ToolRegistry(self.guard).call(
|
|
"create_article", {"slug": "mcp-test", "fields": self.article_fields()}
|
|
)
|
|
context = result["worktree_context"]
|
|
self.assertIn("unrelated.txt", context["before_changed_paths"])
|
|
self.assertIn("content/articles/mcp-test/", context["after_changed_paths"])
|
|
|
|
def test_stdio_protocol_initializes_lists_and_calls_tools(self) -> None:
|
|
messages = [
|
|
{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "test", "version": "1"}}},
|
|
{"jsonrpc": "2.0", "method": "notifications/initialized"},
|
|
{"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}},
|
|
{"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "site_structure", "arguments": {}}},
|
|
]
|
|
input_stream = StringIO("".join(json.dumps(message) + "\n" for message in messages))
|
|
output_stream = StringIO()
|
|
server = StdioMcpServer(self.guard, input_stream=input_stream, output_stream=output_stream)
|
|
self.assertEqual(0, server.serve_forever())
|
|
responses = [json.loads(line) for line in output_stream.getvalue().splitlines()]
|
|
self.assertEqual([1, 2, 3], [response["id"] for response in responses])
|
|
self.assertEqual("2025-06-18", responses[0]["result"]["protocolVersion"])
|
|
self.assertGreaterEqual(len(responses[1]["result"]["tools"]), 30)
|
|
self.assertFalse(responses[2]["result"]["isError"])
|
|
|
|
def test_validation_failure_is_structured(self) -> None:
|
|
git = GitOperations(self.guard)
|
|
validation = ValidationOperations(self.guard, git)
|
|
validation._lektor_executable = lambda: "lektor" # type: ignore[method-assign]
|
|
validation._run = lambda _command, timeout: { # type: ignore[method-assign]
|
|
"passed": False, "returncode": 1, "stdout": "", "stderr": "build failed", "duration_seconds": 0.1
|
|
}
|
|
result = validation.build_site()
|
|
self.assertFalse(result["passed"])
|
|
self.assertEqual("build failed", result["stderr"])
|
|
self.assertFalse(result["output_retained"])
|
|
|
|
|
|
class RealStdioStartupTests(unittest.TestCase):
|
|
def test_module_starts_over_stdio_without_protocol_noise(self) -> None:
|
|
request = json.dumps(
|
|
{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "test", "version": "1"}}}
|
|
) + "\n"
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "labyricorn_mcp", "--repository", str(ROOT)],
|
|
cwd=ROOT,
|
|
input=request,
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
lines = result.stdout.splitlines()
|
|
self.assertEqual(1, len(lines))
|
|
response = json.loads(lines[0])
|
|
self.assertEqual("labyricorn-mcp", response["result"]["serverInfo"]["name"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|