Codex
d3bd7f0c08
Add short-lived single-use HTTP upload tickets that authenticated MCP clients can request before uploading files to /mcp-ingest/uploads. This lets remote agents use curl without needing to read or expose the main MCP bearer token, avoiding oversized base64 tool arguments for larger files. The upload endpoint now accepts either the normal bearer token or X-Obsidian-Wiki-Upload-Token. Ticket defaults can carry project, doc_type_hint, note, start_job, and file-count limits while uploads still stage outside the vault. Validation: python3 -m unittest discover -v; python3 -m py_compile obsidian_wiki/mcp_ingest.py obsidian_wiki/mcp_server.py tests/test_mcp_ingest.py tests/test_mcp_server_remote.py; git diff --check
377 lines
16 KiB
Python
377 lines
16 KiB
Python
import base64
|
|
import json
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from obsidian_wiki.mcp_ingest import (
|
|
IngestError,
|
|
IngestManager,
|
|
parse_bool,
|
|
)
|
|
|
|
|
|
class McpIngestTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tempdir = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tempdir.name)
|
|
self.vault = self.root / "vault"
|
|
self.project = self.root / "project"
|
|
self.runtime = self.project / ".runtime" / "mcp-ingest"
|
|
self.vault.mkdir()
|
|
self.project.mkdir()
|
|
self._write_opencode_skill_fixture(self.project)
|
|
self.manager = IngestManager(
|
|
vault_path=self.vault,
|
|
project_root=self.project,
|
|
runtime_root=self.runtime,
|
|
opencode_path=Path("/usr/bin/opencode"),
|
|
max_file_bytes=100,
|
|
max_chunk_bytes=8,
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.tempdir.cleanup()
|
|
|
|
def test_parse_bool_accepts_enabled_values_only(self):
|
|
self.assertTrue(parse_bool("1"))
|
|
self.assertTrue(parse_bool("true"))
|
|
self.assertTrue(parse_bool("YES"))
|
|
self.assertFalse(parse_bool(""))
|
|
self.assertFalse(parse_bool("0"))
|
|
self.assertFalse(parse_bool("false"))
|
|
|
|
def test_uploads_are_staged_in_runtime_not_vault(self):
|
|
upload = self.manager.create_upload(file_name="drawing.pdf", size_bytes=0, sha256=self._sha256(b""))
|
|
upload_path = Path(upload["path"]).resolve()
|
|
|
|
self.assertTrue(upload_path.is_relative_to((self.runtime / "uploads").resolve()))
|
|
self.assertFalse(upload_path.is_relative_to(self.vault.resolve()))
|
|
|
|
def test_runtime_root_must_not_be_inside_vault(self):
|
|
with self.assertRaises(IngestError):
|
|
IngestManager(
|
|
vault_path=self.vault,
|
|
project_root=self.project,
|
|
runtime_root=self.vault / ".runtime" / "mcp-ingest",
|
|
)
|
|
|
|
def test_check_opencode_readiness_static_success(self):
|
|
with patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run) as run_mock:
|
|
result = self.manager.check_opencode_readiness()
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertTrue(result["static_preflight"])
|
|
self.assertEqual(result["runtime_smoke_test"], "skipped")
|
|
self.assertEqual(result["opencode_version"], "1.17.13")
|
|
self.assertFalse(result["errors"])
|
|
self.assertTrue(any(path.endswith(".agents/skills/wiki-ingest/SKILL.md") for path in result["checked_paths"]))
|
|
self.assertEqual(run_mock.call_count, 2)
|
|
|
|
def test_check_opencode_readiness_reports_missing_skill_files(self):
|
|
missing_project = self.root / "missing-project"
|
|
missing_project.mkdir()
|
|
(missing_project / "AGENTS.md").write_text("# Agent rules\n", encoding="utf-8")
|
|
manager = IngestManager(
|
|
vault_path=self.vault,
|
|
project_root=missing_project,
|
|
runtime_root=self.root / "runtime-missing",
|
|
opencode_path=Path("/usr/bin/opencode"),
|
|
)
|
|
|
|
with patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run):
|
|
result = manager.check_opencode_readiness()
|
|
|
|
self.assertFalse(result["ok"])
|
|
self.assertFalse(result["static_preflight"])
|
|
self.assertIn(".agents/skills/wiki-ingest/SKILL.md", "\n".join(result["errors"]))
|
|
self.assertIn(".agents/skills/llm-wiki/SKILL.md", "\n".join(result["errors"]))
|
|
|
|
def test_check_opencode_readiness_smoke_test_success(self):
|
|
with patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_smoke_success_run) as run_mock:
|
|
result = self.manager.check_opencode_readiness(run_smoke_test=True)
|
|
|
|
self.assertTrue(result["ok"])
|
|
self.assertEqual(result["runtime_smoke_test"], "passed")
|
|
self.assertEqual(run_mock.call_count, 3)
|
|
smoke_args = run_mock.call_args_list[-1].args[0]
|
|
self.assertEqual(smoke_args[:4], ["/usr/bin/opencode", "run", "--format", "json"])
|
|
self.assertIn("--auto", smoke_args)
|
|
self.assertIn("--dir", smoke_args)
|
|
self.assertIn("MCP ingest readiness smoke test", smoke_args)
|
|
|
|
def test_check_opencode_readiness_smoke_test_failure_and_timeout(self):
|
|
with patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_smoke_failure_run):
|
|
failed = self.manager.check_opencode_readiness(run_smoke_test=True)
|
|
self.assertFalse(failed["ok"])
|
|
self.assertEqual(failed["runtime_smoke_test"], "failed")
|
|
self.assertIn("SKILL_LOAD_FAILED", "\n".join(failed["errors"]))
|
|
|
|
with patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_smoke_timeout_run):
|
|
timed_out = self.manager.check_opencode_readiness(run_smoke_test=True)
|
|
self.assertFalse(timed_out["ok"])
|
|
self.assertEqual(timed_out["runtime_smoke_test"], "failed")
|
|
self.assertIn("timed out", "\n".join(timed_out["errors"]))
|
|
|
|
def test_start_ingest_job_fails_before_opencode_when_static_readiness_fails(self):
|
|
payload = b"source data"
|
|
upload = self._complete_upload("source.pdf", payload)
|
|
(self.project / ".agents" / "skills" / "wiki-ingest" / "SKILL.md").unlink()
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen") as popen_mock,
|
|
):
|
|
job = self.manager.start_ingest_job([upload["upload_id"]])
|
|
|
|
self.assertEqual(job["state"], "failed")
|
|
self.assertIn("OpenCode readiness failed", job["error"])
|
|
popen_mock.assert_not_called()
|
|
|
|
def test_upload_chunk_lifecycle_verifies_offset_size_and_sha(self):
|
|
payload = b"hello ingest"
|
|
upload = self.manager.create_upload(
|
|
file_name="drawing.pdf",
|
|
size_bytes=len(payload),
|
|
sha256="sha256:" + self._sha256(payload),
|
|
project="bjd",
|
|
doc_type_hint="drawing",
|
|
)
|
|
|
|
first = self.manager.append_upload_chunk(
|
|
upload["upload_id"],
|
|
offset=0,
|
|
data_base64=base64.b64encode(payload[:5]).decode("ascii"),
|
|
)
|
|
self.assertEqual(first["received_bytes"], 5)
|
|
|
|
with self.assertRaises(IngestError):
|
|
self.manager.append_upload_chunk(
|
|
upload["upload_id"],
|
|
offset=3,
|
|
data_base64=base64.b64encode(payload[5:]).decode("ascii"),
|
|
)
|
|
|
|
second = self.manager.append_upload_chunk(
|
|
upload["upload_id"],
|
|
offset=5,
|
|
data_base64=base64.b64encode(payload[5:]).decode("ascii"),
|
|
)
|
|
self.assertEqual(second["received_bytes"], len(payload))
|
|
|
|
complete = self.manager.complete_upload(upload["upload_id"])
|
|
|
|
self.assertEqual(complete["state"], "complete")
|
|
self.assertEqual(Path(complete["path"]).read_bytes(), payload)
|
|
|
|
def test_create_completed_upload_accepts_bytes_without_agent_chunking(self):
|
|
payload = b"direct upload payload"
|
|
|
|
upload = self.manager.create_completed_upload(
|
|
file_name="drawing.pdf",
|
|
payload=payload,
|
|
project="bjd",
|
|
doc_type_hint="drawing",
|
|
)
|
|
|
|
self.assertEqual(upload["state"], "complete")
|
|
self.assertEqual(upload["received_bytes"], len(payload))
|
|
self.assertEqual(upload["size_bytes"], len(payload))
|
|
self.assertEqual(upload["sha256"], "sha256:" + self._sha256(payload))
|
|
self.assertEqual(upload["project"], "bjd")
|
|
self.assertEqual(upload["doc_type_hint"], "drawing")
|
|
self.assertEqual(Path(upload["path"]).read_bytes(), payload)
|
|
|
|
def test_upload_ticket_is_single_use_and_preserves_defaults(self):
|
|
ticket = self.manager.create_upload_ticket(
|
|
expires_seconds=60,
|
|
max_files=1,
|
|
project="bjd",
|
|
doc_type_hint="drawing",
|
|
note="customer uploaded drawing",
|
|
start_job=False,
|
|
)
|
|
|
|
self.assertIn("upload_token", ticket)
|
|
self.assertEqual(ticket["max_files"], 1)
|
|
self.assertEqual(ticket["project"], "bjd")
|
|
self.assertEqual(ticket["doc_type_hint"], "drawing")
|
|
self.assertEqual(ticket["note"], "customer uploaded drawing")
|
|
self.assertFalse(ticket["start_job"])
|
|
self.assertEqual(ticket["used_at"], None)
|
|
|
|
used = self.manager.consume_upload_ticket(ticket["upload_token"])
|
|
|
|
self.assertEqual(used["project"], "bjd")
|
|
self.assertIsNotNone(used["used_at"])
|
|
with self.assertRaises(IngestError):
|
|
self.manager.consume_upload_ticket(ticket["upload_token"])
|
|
|
|
def test_upload_rejects_unsafe_names_oversized_files_and_large_chunks(self):
|
|
with self.assertRaises(IngestError):
|
|
self.manager.create_upload(file_name="../secret.pdf", size_bytes=10, sha256="a" * 64)
|
|
|
|
with self.assertRaises(IngestError):
|
|
self.manager.create_upload(file_name="large.pdf", size_bytes=101, sha256="a" * 64)
|
|
|
|
upload = self.manager.create_upload(file_name="ok.pdf", size_bytes=20, sha256=self._sha256(b"x" * 20))
|
|
with self.assertRaises(IngestError):
|
|
self.manager.append_upload_chunk(
|
|
upload["upload_id"],
|
|
offset=0,
|
|
data_base64=base64.b64encode(b"x" * 9).decode("ascii"),
|
|
)
|
|
|
|
with self.assertRaises(IngestError):
|
|
self.manager.create_completed_upload(file_name="too-large.pdf", payload=b"x" * 101)
|
|
|
|
def test_start_ingest_job_constructs_restricted_opencode_command(self):
|
|
payload = b"source data"
|
|
upload = self._complete_upload("source.pdf", payload)
|
|
popen = FakePopen()
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen) as popen_mock,
|
|
):
|
|
job = self.manager.start_ingest_job([upload["upload_id"]], note="customer asked for drawings")
|
|
|
|
args = popen_mock.call_args.args[0]
|
|
self.assertEqual(args[:5], ["/usr/bin/opencode", "run", "--format", "json", "--auto"])
|
|
self.assertIn("--dir", args)
|
|
self.assertEqual(args[args.index("--dir") + 1], str(self.project))
|
|
self.assertIn("--title", args)
|
|
self.assertIn(f"MCP ingest {job['job_id']}", args)
|
|
prompt = args[-1]
|
|
self.assertIn("/wiki-ingest", prompt)
|
|
self.assertIn(str(Path(upload["path"])), prompt)
|
|
self.assertIn("temporary runtime staging files", prompt)
|
|
self.assertIn("Do not ingest any file outside this list.", prompt)
|
|
self.assertIn("obsidian-wiki file-index rebuild", prompt)
|
|
self.assertNotIn("../secret", prompt)
|
|
|
|
popen.returncode = 0
|
|
self.manager.refresh_jobs()
|
|
status = self.manager.get_job_status(job["job_id"])
|
|
self.assertEqual(status["state"], "succeeded")
|
|
self.assertEqual(status["exit_code"], 0)
|
|
|
|
def test_jobs_queue_one_at_a_time_and_cancel_queued_job(self):
|
|
first = self._complete_upload("first.pdf", b"first")
|
|
second = self._complete_upload("second.pdf", b"second")
|
|
popen = FakePopen(returncode=None)
|
|
|
|
with (
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.run", side_effect=self._fake_static_readiness_run),
|
|
patch("obsidian_wiki.mcp_ingest.subprocess.Popen", return_value=popen),
|
|
):
|
|
first_job = self.manager.start_ingest_job([first["upload_id"]])
|
|
second_job = self.manager.start_ingest_job([second["upload_id"]])
|
|
|
|
self.assertEqual(self.manager.get_job_status(first_job["job_id"])["state"], "running")
|
|
self.assertEqual(self.manager.get_job_status(second_job["job_id"])["state"], "queued")
|
|
|
|
cancelled = self.manager.cancel_ingest_job(second_job["job_id"])
|
|
|
|
self.assertEqual(cancelled["state"], "cancelled")
|
|
|
|
def _complete_upload(self, name: str, payload: bytes) -> dict:
|
|
upload = self.manager.create_upload(name, len(payload), self._sha256(payload))
|
|
offset = 0
|
|
while offset < len(payload):
|
|
chunk = payload[offset : offset + self.manager.max_chunk_bytes]
|
|
self.manager.append_upload_chunk(
|
|
upload["upload_id"],
|
|
offset=offset,
|
|
data_base64=base64.b64encode(chunk).decode("ascii"),
|
|
)
|
|
offset += len(chunk)
|
|
return self.manager.complete_upload(upload["upload_id"])
|
|
|
|
def _sha256(self, payload: bytes) -> str:
|
|
import hashlib
|
|
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
def _write_opencode_skill_fixture(self, root: Path) -> None:
|
|
(root / "AGENTS.md").write_text("# Agent rules\n", encoding="utf-8")
|
|
wiki_ingest = root / ".agents" / "skills" / "wiki-ingest"
|
|
llm_wiki = root / ".agents" / "skills" / "llm-wiki"
|
|
wiki_ingest_refs = wiki_ingest / "references"
|
|
wiki_ingest_refs.mkdir(parents=True)
|
|
llm_wiki.mkdir(parents=True)
|
|
(wiki_ingest / "SKILL.md").write_text(
|
|
"---\nname: wiki-ingest\ndescription: Ingest documents into the wiki.\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
(llm_wiki / "SKILL.md").write_text(
|
|
"---\nname: llm-wiki\ndescription: Shared wiki framework instructions.\n---\n",
|
|
encoding="utf-8",
|
|
)
|
|
(wiki_ingest_refs / "ingest-prompts.md").write_text("# Prompt templates\n", encoding="utf-8")
|
|
|
|
def _fake_static_readiness_run(self, args, **_kwargs):
|
|
if args == ["/usr/bin/opencode", "--version"]:
|
|
return subprocess.CompletedProcess(args, 0, stdout="1.17.13\n", stderr="")
|
|
if args == ["/usr/bin/opencode", "run", "--help"]:
|
|
return subprocess.CompletedProcess(
|
|
args,
|
|
0,
|
|
stdout="--format\n--auto\n--dir\n--title\n",
|
|
stderr="",
|
|
)
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
def _fake_smoke_success_run(self, args, **kwargs):
|
|
if args in (["/usr/bin/opencode", "--version"], ["/usr/bin/opencode", "run", "--help"]):
|
|
return self._fake_static_readiness_run(args, **kwargs)
|
|
if args[:3] == ["/usr/bin/opencode", "run", "--format"]:
|
|
return subprocess.CompletedProcess(args, 0, stdout='{"message":"WIKI_INGEST_SKILL_LOADED"}\n', stderr="")
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
def _fake_smoke_failure_run(self, args, **kwargs):
|
|
if args in (["/usr/bin/opencode", "--version"], ["/usr/bin/opencode", "run", "--help"]):
|
|
return self._fake_static_readiness_run(args, **kwargs)
|
|
if args[:3] == ["/usr/bin/opencode", "run", "--format"]:
|
|
return subprocess.CompletedProcess(args, 0, stdout="SKILL_LOAD_FAILED: missing\n", stderr="")
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
def _fake_smoke_timeout_run(self, args, **kwargs):
|
|
if args in (["/usr/bin/opencode", "--version"], ["/usr/bin/opencode", "run", "--help"]):
|
|
return self._fake_static_readiness_run(args, **kwargs)
|
|
if args[:3] == ["/usr/bin/opencode", "run", "--format"]:
|
|
raise subprocess.TimeoutExpired(args, kwargs.get("timeout"))
|
|
raise AssertionError(f"unexpected subprocess.run args: {args}")
|
|
|
|
|
|
class FakePopen:
|
|
def __init__(self, returncode=None):
|
|
self.returncode = returncode
|
|
self.pid = 12345
|
|
self.terminated = False
|
|
|
|
def poll(self):
|
|
return self.returncode
|
|
|
|
def wait(self, timeout=None):
|
|
deadline = time.monotonic() + (timeout or 0)
|
|
while self.returncode is None and (timeout is None or time.monotonic() < deadline):
|
|
time.sleep(0.01)
|
|
if self.returncode is None:
|
|
raise subprocess.TimeoutExpired("fake", timeout)
|
|
return self.returncode
|
|
|
|
def terminate(self):
|
|
self.terminated = True
|
|
self.returncode = -15
|
|
|
|
def kill(self):
|
|
self.returncode = -9
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|