gitea-admin/obsidian-wiki
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2183404966ea5d59f56fba5eda27226f4eaa39a5
obsidian-wiki/tests/test_mcp_server_remote.py
T
Codex 2183404966 feat(mcp): add verified ingest job phases 
Expose authoritative ingest phases and supervisor-backed process liveness without semantic activity inference. Require deterministic registry, manifest, archive, file-index, and declared-output verification before an agent-successful job can finish succeeded.\n\nPreserve legacy state values, derive defaults for older job records, keep verification outside the manager lock, and retain partial outputs on verification failure.\n\nValidated with the full unittest suite (205 tests).
2026-07-16 14:26:58 +00:00

1791 lines
71 KiB
Python
Raw Blame History

import asyncio
import hashlib
import json
import os
import tempfile
import time
import unittest
from pathlib import Path
from unittest.mock import patch
from obsidian_wiki import mcp_server
from obsidian_wiki.mcp_ingest import IngestError, IngestManager
class FakeApp:
def __init__(self):
self.called = False
async def __call__(self, scope, receive, send):
self.called = True
await send({"type": "http.response.start", "status": 200, "headers": []})
await send({"type": "http.response.body", "body": b"ok"})
def run_asgi_app(app, *, headers=None, scope_type="http"):
messages = []
scope = {"type": scope_type, "headers": headers or []}
async def receive():
return {"type": "http.request", "body": b"", "more_body": False}
async def send(message):
messages.append(message)
asyncio.run(app(scope, receive, send))
return messages
def run_asgi_http_request(app, *, method="GET", path="/", headers=None, body=b"", query_string=b""):
messages = []
sent = False
query = query_string.encode("utf-8") if isinstance(query_string, str) else query_string
scope = {
"type": "http",
"method": method,
"path": path,
"raw_path": path.encode("ascii"),
"query_string": query,
"headers": headers or [],
}
async def receive():
nonlocal sent
if sent:
return {"type": "http.disconnect"}
sent = True
return {"type": "http.request", "body": body, "more_body": False}
async def send(message):
messages.append(message)
asyncio.run(app(scope, receive, send))
return messages
def multipart_body(fields, files, boundary="test-boundary"):
chunks = []
for name, value in fields.items():
chunks.append(f"--{boundary}\r\n".encode("ascii"))
chunks.append(f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode("ascii"))
chunks.append(str(value).encode("utf-8"))
chunks.append(b"\r\n")
for field_name, file_name, content_type, payload in files:
chunks.append(f"--{boundary}\r\n".encode("ascii"))
chunks.append(
(
f'Content-Disposition: form-data; name="{field_name}"; filename="{file_name}"\r\n'
f"Content-Type: {content_type}\r\n\r\n"
).encode("ascii")
)
chunks.append(payload)
chunks.append(b"\r\n")
chunks.append(f"--{boundary}--\r\n".encode("ascii"))
return b"".join(chunks), boundary
class RemoteMcpServerTests(unittest.TestCase):
def test_bearer_token_middleware_accepts_matching_token(self):
inner = FakeApp()
middleware = mcp_server.BearerTokenAuthMiddleware(inner, "secret-token")
messages = run_asgi_app(
middleware,
headers=[(b"authorization", b"Bearer secret-token")],
)
self.assertTrue(inner.called)
self.assertEqual(messages[0]["status"], 200)
def test_bearer_token_middleware_rejects_missing_or_wrong_token(self):
inner = FakeApp()
middleware = mcp_server.BearerTokenAuthMiddleware(inner, "secret-token")
messages = run_asgi_app(
middleware,
headers=[(b"authorization", b"Bearer wrong-token")],
)
self.assertFalse(inner.called)
self.assertEqual(messages[0]["status"], 401)
self.assertIn((b"www-authenticate", b"Bearer"), messages[0]["headers"])
def test_bearer_token_middleware_requires_non_empty_token(self):
with self.assertRaises(ValueError):
mcp_server.BearerTokenAuthMiddleware(FakeApp(), "")
def test_read_token_from_env_trims_and_requires_value(self):
with patch.dict(os.environ, {"TEST_MCP_TOKEN": " secret-token "}, clear=False):
self.assertEqual(mcp_server.read_token_from_env("TEST_MCP_TOKEN"), "secret-token")
with patch.dict(os.environ, {}, clear=True):
with self.assertRaises(RuntimeError):
mcp_server.read_token_from_env("TEST_MCP_TOKEN")
def test_runtime_config_reads_environment_with_defaults(self):
with patch.dict(os.environ, {"OBSIDIAN_MCP_HOST": " 0.0.0.0 ", "OBSIDIAN_MCP_PORT": "9876"}, clear=True):
self.assertEqual(mcp_server.runtime_config_value("OBSIDIAN_MCP_HOST", "127.0.0.1"), "0.0.0.0")
self.assertEqual(mcp_server.runtime_config_int("OBSIDIAN_MCP_PORT", 8765), 9876)
with patch.dict(os.environ, {"OBSIDIAN_MCP_PORT": "not-an-int"}, clear=True):
self.assertEqual(mcp_server.runtime_config_int("OBSIDIAN_MCP_PORT", 8765), 8765)
with patch.dict(os.environ, {"OBSIDIAN_MCP_ALLOWED_HOSTS": "wiki.example:*, 10.0.0.2:*"}, clear=True):
self.assertEqual(
mcp_server.runtime_config_list("OBSIDIAN_MCP_ALLOWED_HOSTS"),
["wiki.example:*", "10.0.0.2:*"],
)
with patch.dict(os.environ, {}, clear=True):
self.assertIsNone(mcp_server.runtime_config_list("OBSIDIAN_MCP_ALLOWED_HOSTS"))
def test_enabled_tools_config_prefers_tool_list_over_legacy_ingest_flag(self):
with patch.dict(
os.environ,
{
"OBSIDIAN_MCP_ENABLED_TOOLS": "file-assets",
"OBSIDIAN_MCP_ENABLE_INGEST": "1",
},
clear=True,
):
self.assertEqual(mcp_server.enabled_tool_names(), {"file-assets"})
with patch.dict(os.environ, {"OBSIDIAN_MCP_ENABLE_INGEST": "1"}, clear=True):
self.assertEqual(mcp_server.enabled_tool_names(), {"file-assets", "ingest"})
with patch.dict(os.environ, {"OBSIDIAN_MCP_ENABLED_TOOLS": "file-assets,query"}, clear=True):
self.assertEqual(mcp_server.enabled_tool_names(), {"file-assets", "query"})
def test_create_http_app_wraps_streamable_http_app_with_auth(self):
class FakeServer:
def __init__(self):
self.inner = FakeApp()
def streamable_http_app(self):
return self.inner
fake_server = FakeServer()
app = mcp_server.create_http_app(fake_server, "secret-token")
messages = run_asgi_app(
app,
headers=[(b"authorization", b"Bearer secret-token")],
)
self.assertTrue(fake_server.inner.called)
self.assertEqual(messages[0]["status"], 200)
def test_http_upload_endpoint_accepts_multipart_when_ingest_enabled(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
payload = b"uploaded drawing"
body, boundary = multipart_body(
{"project": "bjd", "doc_type_hint": "drawing", "start_job": "false"},
[("files", "drawing.pdf", "application/pdf", payload)],
)
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"authorization", b"Bearer secret-token"),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
body=body,
)
self.assertEqual(messages[0]["status"], 200)
response = b"".join(message.get("body", b"") for message in messages if message["type"] == "http.response.body")
data = json.loads(response.decode("utf-8"))
self.assertEqual(data["job"], None)
self.assertEqual(data["uploads"][0]["state"], "complete")
self.assertEqual(data["uploads"][0]["file_name"], "drawing.pdf")
self.assertEqual(data["uploads"][0]["sha256"], "sha256:" + hashlib.sha256(payload).hexdigest())
self.assertNotIn("path", data["uploads"][0])
def test_http_upload_endpoint_creates_bundle_from_body_and_attachments(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
app = mcp_server.create_http_app(
FakeApp(),
"secret-token",
ingest_manager=manager,
)
body, boundary = multipart_body(
{
"body_text": "First message\n\nSecond message",
"body_title": "Customer thread",
"bundle_id": "thread-123",
"source_type": "email_thread",
"label": "Customer quality thread",
"start_job": "false",
},
[
("files", "report.pdf", "application/pdf", b"report"),
("files", "photo.jpg", "image/jpeg", b"photo"),
],
)
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"authorization", b"Bearer secret-token"),
(
b"content-type",
f"multipart/form-data; boundary={boundary}".encode("ascii"),
),
],
body=body,
)
self.assertEqual(messages[0]["status"], 200)
response = b"".join(
message.get("body", b"")
for message in messages
if message["type"] == "http.response.body"
)
data = json.loads(response.decode("utf-8"))
self.assertIsNone(data["job"])
self.assertEqual(data["bundle"]["bundle_id"], "thread-123")
self.assertEqual(data["bundle"]["source_type"], "email_thread")
self.assertEqual(data["bundle"]["label"], "Customer quality thread")
self.assertEqual(len(data["uploads"]), 3)
self.assertEqual(data["uploads"][0]["bundle_role"], "primary")
self.assertEqual(data["uploads"][1]["bundle_role"], "attachment")
self.assertEqual(data["uploads"][2]["bundle_role"], "attachment")
body_meta = manager._load_upload(data["bundle"]["body_upload_id"])
body_payload = Path(body_meta["path"]).read_text(encoding="utf-8")
self.assertIn("# Customer thread", body_payload)
self.assertIn("First message", body_payload)
def test_http_upload_endpoint_streams_multipart_without_buffering_whole_body(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
payload = b"stream this upload"
body, boundary = multipart_body(
{"start_job": "false"},
[("files", "stream.txt", "text/plain", payload)],
)
with (
patch(
"obsidian_wiki.mcp_server._read_http_body",
side_effect=AssertionError("buffered body"),
create=True,
),
patch.object(manager, "create_completed_upload", side_effect=AssertionError("bytes upload")),
):
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"authorization", b"Bearer secret-token"),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
body=body,
)
self.assertEqual(messages[0]["status"], 200)
response = b"".join(message.get("body", b"") for message in messages if message["type"] == "http.response.body")
data = json.loads(response.decode("utf-8"))
self.assertEqual(data["uploads"][0]["sha256"], "sha256:" + hashlib.sha256(payload).hexdigest())
def test_http_upload_file_copy_does_not_block_event_loop(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
body, boundary = multipart_body(
{"start_job": "false"},
[("files", "stream.txt", "text/plain", b"stream this upload")],
)
original = manager.create_completed_upload_from_stream
def slow_copy(*args, **kwargs):
time.sleep(0.1)
return original(*args, **kwargs)
async def exercise():
messages = []
sent = False
scope = {
"type": "http",
"method": "POST",
"path": "/mcp-ingest/uploads",
"raw_path": b"/mcp-ingest/uploads",
"query_string": b"",
"headers": [
(b"authorization", b"Bearer secret-token"),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
}
async def receive():
nonlocal sent
if sent:
return {"type": "http.disconnect"}
sent = True
return {"type": "http.request", "body": body, "more_body": False}
async def send(message):
messages.append(message)
started = time.monotonic()
request = asyncio.create_task(app(scope, receive, send))
await asyncio.sleep(0.01)
responsive_after = time.monotonic() - started
await request
return responsive_after, messages
with patch.object(manager, "create_completed_upload_from_stream", side_effect=slow_copy):
responsive_after, messages = asyncio.run(exercise())
self.assertLess(responsive_after, 0.08)
self.assertEqual(messages[0]["status"], 200)
def test_http_upload_endpoint_rejects_oversized_content_length_before_reading(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
max_file_bytes=1000,
max_files_per_job=1,
max_request_bytes=10,
)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
with patch("starlette.formparsers.MultiPartParser.parse", side_effect=AssertionError("body parsed")):
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"authorization", b"Bearer secret-token"),
(b"content-type", b"multipart/form-data; boundary=test"),
(b"content-length", b"999999"),
],
body=b"",
)
self.assertEqual(messages[0]["status"], 413)
def test_http_upload_endpoint_accepts_upload_ticket_without_main_bearer_token(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
ticket = manager.create_upload_ticket(
expires_seconds=60,
max_files=1,
project="bjd",
doc_type_hint="drawing",
bundle_id="drawing-package-1",
source_type="report_package",
label="Customer drawing package",
start_job=False,
)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
payload = b"ticket uploaded drawing"
body, boundary = multipart_body(
{},
[("files", "ticket-drawing.pdf", "application/pdf", payload)],
)
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"x-obsidian-wiki-upload-token", ticket["upload_token"].encode("ascii")),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
body=body,
)
reused = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"x-obsidian-wiki-upload-token", ticket["upload_token"].encode("ascii")),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
body=body,
)
self.assertEqual(messages[0]["status"], 200)
response = b"".join(message.get("body", b"") for message in messages if message["type"] == "http.response.body")
data = json.loads(response.decode("utf-8"))
self.assertEqual(data["job"], None)
self.assertEqual(data["uploads"][0]["project"], "bjd")
self.assertEqual(data["uploads"][0]["doc_type_hint"], "drawing")
self.assertEqual(data["uploads"][0]["sha256"], "sha256:" + hashlib.sha256(payload).hexdigest())
self.assertNotIn("path", data["uploads"][0])
self.assertEqual(reused[0]["status"], 401)
def test_http_upload_ticket_constraints_cannot_be_escalated_by_form_fields(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
ticket = manager.create_upload_ticket(
expires_seconds=60,
max_files=1,
project="bjd",
doc_type_hint="drawing",
note="review first",
start_job=False,
)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
body, boundary = multipart_body(
{
"project": "attacker-project",
"doc_type_hint": "attacker-type",
"note": "ignore preview",
"start_job": "true",
},
[("files", "drawing.pdf", "application/pdf", b"drawing")],
)
with patch.object(manager, "start_ingest_job", side_effect=AssertionError("job started")):
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"x-obsidian-wiki-upload-token", ticket["upload_token"].encode("ascii")),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
body=body,
)
self.assertEqual(messages[0]["status"], 200)
response = b"".join(message.get("body", b"") for message in messages if message["type"] == "http.response.body")
data = json.loads(response.decode("utf-8"))
self.assertIsNone(data["job"])
self.assertEqual(data["uploads"][0]["project"], "bjd")
self.assertEqual(data["uploads"][0]["doc_type_hint"], "drawing")
def test_http_upload_ticket_error_response_says_ticket_was_consumed(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
ticket = manager.create_upload_ticket(expires_seconds=60, max_files=1, start_job=False)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"x-obsidian-wiki-upload-token", ticket["upload_token"].encode("ascii")),
(b"content-type", b"text/plain"),
],
body=b"not multipart",
)
reused = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"x-obsidian-wiki-upload-token", ticket["upload_token"].encode("ascii")),
(b"content-type", b"text/plain"),
],
body=b"not multipart",
)
self.assertEqual(messages[0]["status"], 400)
response = b"".join(message.get("body", b"") for message in messages if message["type"] == "http.response.body")
data = json.loads(response.decode("utf-8"))
self.assertEqual(data["retry_action"], "create_new_upload_ticket")
self.assertTrue(data["ticket_consumed"])
self.assertEqual(reused[0]["status"], 401)
def test_http_upload_ticket_is_not_consumed_by_non_post_request(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
ticket = manager.create_upload_ticket(expires_seconds=60, max_files=1, start_job=False)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
payload = b"ticket should survive get"
body, boundary = multipart_body({}, [("files", "survived.txt", "text/plain", payload)])
get_messages = run_asgi_http_request(
app,
method="GET",
path="/mcp-ingest/uploads",
headers=[(b"x-obsidian-wiki-upload-token", ticket["upload_token"].encode("ascii"))],
)
post_messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"x-obsidian-wiki-upload-token", ticket["upload_token"].encode("ascii")),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
body=body,
)
self.assertEqual(get_messages[0]["status"], 405)
self.assertEqual(post_messages[0]["status"], 200)
def test_http_upload_ticket_is_not_accepted_in_query_string(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
ticket = manager.create_upload_ticket(expires_seconds=60, max_files=1, start_job=False)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
body, boundary = multipart_body({}, [("files", "safe.txt", "text/plain", b"safe")])
query_auth = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
query_string="upload_token=" + ticket["upload_token"],
headers=[(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii"))],
body=body,
)
header_auth = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"x-obsidian-wiki-upload-token", ticket["upload_token"].encode("ascii")),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
body=body,
)
self.assertEqual(query_auth[0]["status"], 401)
self.assertEqual(header_auth[0]["status"], 200)
def test_http_upload_endpoint_rejects_oversized_body_with_413(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
max_file_bytes=1,
max_files_per_job=1,
)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
body, boundary = multipart_body({}, [("files", "too-large.bin", "application/octet-stream", b"x" * 25000)])
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"authorization", b"Bearer secret-token"),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
body=body,
)
self.assertEqual(messages[0]["status"], 413)
def test_http_upload_endpoint_rejects_oversized_file_part_during_parsing(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
max_file_bytes=5,
max_files_per_job=2,
)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
body, boundary = multipart_body(
{"start_job": "false"},
[("files", "too-large.txt", "text/plain", b"123456")],
)
with patch.object(
manager,
"create_completed_upload_from_stream",
side_effect=AssertionError("file was fully parsed"),
):
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"authorization", b"Bearer secret-token"),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
body=body,
)
self.assertEqual(messages[0]["status"], 413)
def test_http_upload_endpoint_rolls_back_files_when_one_file_is_invalid(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
max_file_bytes=5,
max_files_per_job=2,
)
app = mcp_server.create_http_app(FakeApp(), "secret-token", ingest_manager=manager)
body, boundary = multipart_body(
{"start_job": "false"},
[
("files", "valid.txt", "text/plain", b"12345"),
("files", "../invalid.txt", "text/plain", b"123"),
],
)
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"authorization", b"Bearer secret-token"),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
],
body=body,
)
remaining_uploads = list(manager.upload_root.rglob("upload.json"))
self.assertEqual(messages[0]["status"], 400)
self.assertEqual(remaining_uploads, [])
def test_http_upload_endpoint_enforces_allowed_host_and_origin(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
app = mcp_server.create_http_app(
FakeApp(),
"secret-token",
ingest_manager=manager,
allowed_hosts=["good.example:*"],
allowed_origins=["https://app.example"],
)
body, boundary = multipart_body(
{"start_job": "false"},
[("files", "allowed.txt", "text/plain", b"ok")],
)
common_headers = [
(b"authorization", b"Bearer secret-token"),
(b"content-type", f"multipart/form-data; boundary={boundary}".encode("ascii")),
]
bad_host = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[*common_headers, (b"host", b"bad.example:8765"), (b"origin", b"https://app.example")],
body=body,
)
bad_origin = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[*common_headers, (b"host", b"good.example:8765"), (b"origin", b"https://evil.example")],
body=body,
)
allowed = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[*common_headers, (b"host", b"good.example:8765"), (b"origin", b"https://app.example")],
body=body,
)
self.assertEqual(bad_host[0]["status"], 403)
self.assertEqual(bad_origin[0]["status"], 403)
self.assertEqual(allowed[0]["status"], 200)
def test_http_upload_endpoint_rejects_present_origin_when_origin_allowlist_is_empty(self):
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
)
app = mcp_server.create_http_app(
FakeApp(),
"secret-token",
ingest_manager=manager,
allowed_hosts=["good.example"],
allowed_origins=[],
)
messages = run_asgi_http_request(
app,
method="POST",
path="/mcp-ingest/uploads",
headers=[
(b"authorization", b"Bearer secret-token"),
(b"host", b"good.example"),
(b"origin", b"https://unexpected.example"),
],
)
self.assertEqual(messages[0]["status"], 403)
def test_http_upload_endpoint_requires_ingest_and_bearer_token(self):
no_ingest_app = mcp_server.create_http_app(FakeApp(), "secret-token")
missing_token = run_asgi_http_request(no_ingest_app, method="POST", path="/mcp-ingest/uploads")
self.assertEqual(missing_token[0]["status"], 401)
disabled = run_asgi_http_request(
no_ingest_app,
method="POST",
path="/mcp-ingest/uploads",
headers=[(b"authorization", b"Bearer secret-token")],
)
self.assertEqual(disabled[0]["status"], 404)
def test_create_server_uses_file_asset_service_name(self):
class FakeMCP:
names = []
tool_names = []
prompt_names = []
def __init__(self, name, **kwargs):
self.names.append(name)
def tool(self):
def decorator(func):
self.tool_names.append(func.__name__)
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, name=None, **_kwargs):
def decorator(func):
self.prompt_names.append(name or func.__name__)
return func
return decorator
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
server = mcp_server.create_server(vault_path=Path(tmp))
self.assertIsInstance(server, FakeMCP)
self.assertEqual(FakeMCP.names[-1], "wiki-file-assets")
self.assertIn("search_file_assets", FakeMCP.tool_names)
self.assertIn("list_file_assets", FakeMCP.tool_names)
self.assertIn("get_file_asset_metadata", FakeMCP.tool_names)
self.assertIn("get_mcp_capabilities", FakeMCP.tool_names)
self.assertIn("how_to_use_wiki_file_assets", FakeMCP.prompt_names)
self.assertNotIn("create_ingest_upload", FakeMCP.tool_names)
def test_create_server_registers_ingest_tools_when_enabled(self):
class FakeMCP:
tool_names = []
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tool_names.append(func.__name__)
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
server = mcp_server.create_server(
vault_path=root / "vault",
enable_ingest=True,
project_root=root / "project",
ingest_runtime_root=root / "runtime",
)
self.assertIsInstance(server, FakeMCP)
self.assertNotIn("create_ingest_upload", FakeMCP.tool_names)
self.assertNotIn("append_ingest_upload_chunk", FakeMCP.tool_names)
self.assertNotIn("complete_ingest_upload", FakeMCP.tool_names)
self.assertIn("start_ingest_job", FakeMCP.tool_names)
self.assertIn("get_ingest_job_status", FakeMCP.tool_names)
self.assertIn("get_ingest_job_result", FakeMCP.tool_names)
self.assertIn("cancel_ingest_job", FakeMCP.tool_names)
self.assertIn("check_ingest_opencode_readiness", FakeMCP.tool_names)
self.assertIn("get_ingest_upload_instructions", FakeMCP.tool_names)
self.assertIn("create_ingest_upload_ticket", FakeMCP.tool_names)
self.assertIn("ingest_text_note", FakeMCP.tool_names)
self.assertNotIn("ingest_file_base64", FakeMCP.tool_names)
def test_create_server_registers_query_tools_only_when_enabled(self):
class FakeMCP:
tool_names = []
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tool_names.append(func.__name__)
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
class FakeQueryManager:
def ask_wiki(self, query, mode="normal", public_only=False, note=None, timeout_seconds=90, poll_interval_seconds=2):
return {
"job_id": "query-job",
"state": "succeeded",
"answer": "answer",
"answer_markdown": "answer",
"citations": [],
"warnings": [],
}
def start_query_job(self, query, mode="normal", public_only=False, note=None):
return {"job_id": "query-job", "state": "queued"}
def get_query_job_status(self, job_id):
return {"job_id": job_id, "state": "running"}
def get_query_job_result(self, job_id):
return {"job_id": job_id, "state": "succeeded", "answer": "answer"}
def cancel_query_job(self, job_id):
return {"job_id": job_id, "state": "cancelled"}
def check_opencode_readiness(self, *, run_smoke_test=False):
return {"ok": True, "runtime_smoke_test": "skipped"}
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
disabled_server = mcp_server.create_server(vault_path=root / "vault")
disabled_tool_names = list(FakeMCP.tool_names)
FakeMCP.tool_names.clear()
enabled_server = mcp_server.create_server(
vault_path=root / "vault",
enable_query=True,
project_root=root / "project",
query_runtime_root=root / "runtime",
query_manager=FakeQueryManager(),
)
self.assertIsInstance(disabled_server, FakeMCP)
self.assertIsInstance(enabled_server, FakeMCP)
self.assertNotIn("start_wiki_query", disabled_tool_names)
self.assertNotIn("ask_wiki", disabled_tool_names)
self.assertIn("ask_wiki", FakeMCP.tool_names)
self.assertIn("start_wiki_query", FakeMCP.tool_names)
self.assertIn("get_wiki_query_status", FakeMCP.tool_names)
self.assertIn("get_wiki_query_result", FakeMCP.tool_names)
self.assertIn("cancel_wiki_query", FakeMCP.tool_names)
self.assertIn("check_query_opencode_readiness", FakeMCP.tool_names)
self.assertEqual(
FakeMCP.tools["ask_wiki"]("what do I know?", timeout_seconds=3, poll_interval_seconds=1)["answer_markdown"],
"answer",
)
def test_get_mcp_capabilities_reports_agent_facing_tools_without_local_paths(self):
class FakeMCP:
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
mcp_server.create_server(
vault_path=root / "vault",
enable_ingest=True,
enable_query=True,
project_root=root / "project",
ingest_runtime_root=root / "ingest-runtime",
query_runtime_root=root / "query-runtime",
public_base_url="https://wiki.example.test/mcp",
allow_source_files=True,
ingest_manager=IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "fake-ingest",
),
query_manager=type(
"FakeQueryManager",
(),
{
"ask_wiki": lambda self, query, mode="normal", public_only=False, note=None, timeout_seconds=90, poll_interval_seconds=2: {},
"start_query_job": lambda self, query, mode="normal", public_only=False, note=None: {},
"get_query_job_status": lambda self, job_id: {},
"get_query_job_result": lambda self, job_id: {},
"cancel_query_job": lambda self, job_id: {},
"check_opencode_readiness": lambda self, run_smoke_test=False: {},
},
)(),
)
capabilities = FakeMCP.tools["get_mcp_capabilities"]()
self.assertEqual(capabilities["server_name"], "wiki-file-assets")
self.assertIn("server_version", capabilities)
self.assertEqual(capabilities["schema_version"], 3)
self.assertEqual(capabilities["capabilities_schema_version"], 3)
self.assertEqual(capabilities["tools_version"], 3)
self.assertEqual(capabilities["enabled_tools"], ["file-assets", "ingest", "query"])
self.assertIn("file_index", capabilities)
self.assertIn("record_count", capabilities["file_index"])
self.assertIn("last_rebuild_at", capabilities["file_index"])
self.assertIn("last_index_rebuild_at", capabilities["file_index"])
self.assertTrue(capabilities["query"]["enabled"])
self.assertEqual(capabilities["query"]["preferred_tool"], "ask_wiki")
self.assertIn("ask_wiki", capabilities["query"]["tools"])
self.assertTrue(capabilities["query"]["supports_structured_result"])
self.assertTrue(capabilities["query"]["known_limitations"])
self.assertTrue(capabilities["ingest"]["enabled"])
self.assertEqual(capabilities["ingest"]["preferred_upload_auth"], "upload_ticket")
self.assertEqual(capabilities["ingest"]["http_upload_path"], "/mcp-ingest/uploads")
self.assertEqual(capabilities["ingest"]["http_upload_url"], "https://wiki.example.test/mcp-ingest/uploads")
self.assertEqual(capabilities["ingest"]["preferred_text_tool"], "ingest_text_note")
self.assertTrue(capabilities["ingest"]["supports_bundle_upload"])
self.assertTrue(capabilities["ingest"]["supports_body_text"])
self.assertEqual(
capabilities["ingest"]["authoritative_job_phases"],
["queued", "preflight", "agent_running", "verifying", "complete"],
)
self.assertEqual(
capabilities["ingest"]["process_alive_source"],
"process_supervisor",
)
self.assertTrue(capabilities["ingest"]["supports_post_run_verification"])
self.assertFalse(capabilities["ingest"]["semantic_activity_inference"])
self.assertIn("body_text", capabilities["ingest"]["multipart_optional_fields"])
self.assertIn("source_type", capabilities["ingest"]["multipart_optional_fields"])
self.assertIn("validate_upload", capabilities["ingest"]["tools"])
self.assertIn("preview_upload_metadata", capabilities["ingest"]["tools"])
self.assertIn("queue", capabilities["ingest"])
self.assertIn("last_ingest_at", capabilities["ingest"])
self.assertTrue(capabilities["ingest"]["known_limitations"])
self.assertIn("search_file_assets", capabilities["file_assets"]["tools"])
self.assertTrue(capabilities["file_assets"]["source_file_download_enabled"])
self.assertTrue(capabilities["safety"]["arbitrary_shell"])
self.assertFalse(capabilities["safety"]["shell_sandbox_enforced"])
self.assertTrue(capabilities["safety"]["opencode_auto_approve"])
self.assertTrue(capabilities["safety"]["trusted_clients_required"])
self.assertNotIn("vault_path", json.dumps(capabilities))
self.assertNotIn("runtime", json.dumps(capabilities).lower())
self.assertNotIn("Bearer ", json.dumps(capabilities))
self.assertNotIn("upload_token", json.dumps(capabilities))
def test_server_resource_read_honors_source_file_gate(self):
class FakeMCP:
resources = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
return lambda func: func
def resource(self, template, **_kwargs):
def decorator(func):
self.resources[template] = func
return func
return decorator
def prompt(self, *_args, **_kwargs):
return lambda func: func
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
patch.object(
mcp_server,
"read_file_asset_resource",
return_value={"bytes": b"content"},
) as read_resource,
):
root = Path(tmp)
mcp_server.create_server(vault_path=root, allow_source_files=False)
resource = FakeMCP.resources["asset://source/sha256/{digest}"]
self.assertEqual(resource("a" * 64), b"content")
self.assertFalse(read_resource.call_args.kwargs["allow_source_files"])
mcp_server.create_server(vault_path=root, allow_source_files=True)
resource = FakeMCP.resources["asset://source/sha256/{digest}"]
self.assertEqual(resource("b" * 64), b"content")
self.assertTrue(read_resource.call_args.kwargs["allow_source_files"])
def test_readiness_tools_hide_local_paths(self):
class FakeMCP:
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
return lambda func: func
def prompt(self, *_args, **_kwargs):
return lambda func: func
class FakeManager:
def check_opencode_readiness(self, *, run_smoke_test=False):
return {
"ok": False,
"checked_paths": ["/secret/project/.agents/skills/wiki-ingest/SKILL.md"],
"errors": ["missing /secret/project/.agents/skills/wiki-ingest/SKILL.md"],
}
def _public_preflight(self, result):
return {"ok": result["ok"], "errors": ["missing <project> skill"]}
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
mcp_server.create_server(
vault_path=Path(tmp) / "vault",
enable_ingest=True,
ingest_manager=FakeManager(),
)
readiness = FakeMCP.tools["check_ingest_opencode_readiness"]()
self.assertNotIn("checked_paths", readiness)
self.assertNotIn("/secret/project", json.dumps(readiness))
def test_create_server_passes_job_timeouts_to_managers(self):
class FakeMCP:
def __init__(self, _name, **_kwargs):
pass
def tool(self):
return lambda func: func
def resource(self, *_args, **_kwargs):
return lambda func: func
def prompt(self, *_args, **_kwargs):
return lambda func: func
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
patch.object(mcp_server, "IngestManager") as ingest_manager,
patch.object(mcp_server, "QueryManager") as query_manager,
):
root = Path(tmp)
mcp_server.create_server(
vault_path=root / "vault",
enable_ingest=True,
enable_query=True,
project_root=root / "project",
ingest_runtime_root=root / "ingest",
query_runtime_root=root / "query",
ingest_job_timeout_seconds=17,
query_job_timeout_seconds=23,
)
self.assertEqual(ingest_manager.call_args.kwargs["job_timeout_seconds"], 17)
self.assertEqual(query_manager.call_args.kwargs["job_timeout_seconds"], 23)
def test_get_mcp_capabilities_recommends_only_enabled_tool_groups(self):
class FakeMCP:
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
class FakeQueryManager:
def ask_wiki(self, query, mode="normal", public_only=False, note=None, timeout_seconds=90, poll_interval_seconds=2):
return {}
def start_query_job(self, query, mode="normal", public_only=False, note=None):
return {}
def get_query_job_status(self, job_id):
return {}
def get_query_job_result(self, job_id):
return {}
def cancel_query_job(self, job_id):
return {}
def check_opencode_readiness(self, *, run_smoke_test=False):
return {}
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
mcp_server.create_server(
vault_path=root / "vault",
enable_file_assets=False,
enable_ingest=False,
enable_query=True,
project_root=root / "project",
query_runtime_root=root / "query-runtime",
query_manager=FakeQueryManager(),
)
capabilities = FakeMCP.tools["get_mcp_capabilities"]()
first_steps = "\n".join(capabilities["recommended_first_steps"])
self.assertEqual(capabilities["enabled_tools"], ["query"])
self.assertIn("ask_wiki", first_steps)
self.assertNotIn("search_file_assets", first_steps)
self.assertNotIn("ingest_text_note", first_steps)
self.assertNotIn("create_ingest_upload_ticket", first_steps)
def test_file_index_status_reports_opencode_model_settings(self):
class FakeMCP:
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
mcp_server.create_server(
vault_path=root / "vault",
enable_ingest=True,
enable_query=True,
project_root=root / "project",
ingest_runtime_root=root / "ingest-runtime",
query_runtime_root=root / "query-runtime",
ingest_opencode_model="openai/gpt-5.5",
ingest_opencode_variant="high",
query_opencode_model="openai/gpt-5.5",
query_opencode_variant="minimal",
ingest_manager=IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "fake-ingest",
),
query_manager=type(
"FakeQueryManager",
(),
{
"ask_wiki": lambda self, query, mode="normal", public_only=False, note=None, timeout_seconds=90, poll_interval_seconds=2: {},
"start_query_job": lambda self, query, mode="normal", public_only=False, note=None: {},
"get_query_job_status": lambda self, job_id: {},
"get_query_job_result": lambda self, job_id: {},
"cancel_query_job": lambda self, job_id: {},
"check_opencode_readiness": lambda self, run_smoke_test=False: {},
},
)(),
)
status = FakeMCP.tools["file_index_status"]()
self.assertEqual(status["enabled_tools"], ["file-assets", "ingest", "query"])
self.assertEqual(status["ingest_opencode_model"], "openai/gpt-5.5")
self.assertEqual(status["ingest_opencode_variant"], "high")
self.assertEqual(status["query_opencode_model"], "openai/gpt-5.5")
self.assertEqual(status["query_opencode_variant"], "minimal")
self.assertNotIn("vault_path", status)
self.assertNotIn("db_path", status)
self.assertNotIn("registry_path", status)
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
mcp_server.create_server(
vault_path=root / "vault",
expose_paths=True,
)
path_status = FakeMCP.tools["file_index_status"]()
self.assertIn("vault_path", path_status)
self.assertIn("db_path", path_status)
self.assertIn("registry_path", path_status)
def test_create_ingest_upload_ticket_tool_returns_http_only_token(self):
class FakeMCP:
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
mcp_server.create_server(
vault_path=root / "vault",
enable_ingest=True,
project_root=root / "project",
ingest_runtime_root=root / "runtime",
public_base_url="https://wiki.example.test/mcp",
)
ticket = FakeMCP.tools["create_ingest_upload_ticket"](
expires_seconds=60,
max_files=1,
project="bjd",
doc_type_hint="drawing",
bundle_id="drawing-package-1",
source_type="report_package",
label="Customer drawing package",
start_job=False,
)
instructions = FakeMCP.tools["get_ingest_upload_instructions"]()
self.assertIn("upload_token", ticket)
self.assertEqual(ticket["token_header"], "X-Obsidian-Wiki-Upload-Token")
self.assertEqual(ticket["upload_path"], "/mcp-ingest/uploads")
self.assertEqual(ticket["upload_url"], "https://wiki.example.test/mcp-ingest/uploads")
self.assertIn("X-Obsidian-Wiki-Upload-Token", ticket["curl_example"])
self.assertNotIn("Authorization: Bearer", ticket["curl_example"])
self.assertEqual(ticket["max_files"], 1)
self.assertEqual(ticket["bundle_id"], "drawing-package-1")
self.assertEqual(ticket["source_type"], "report_package")
self.assertEqual(ticket["label"], "Customer drawing package")
self.assertFalse(ticket["start_job"])
self.assertEqual(instructions["preferred_upload_auth"], "upload_ticket")
self.assertEqual(instructions["http_upload"]["url"], "https://wiki.example.test/mcp-ingest/uploads")
self.assertIn("body_text", instructions["http_upload"]["optional_fields"])
self.assertIn("bundle_id", instructions["http_upload"]["optional_fields"])
self.assertIn("create_ingest_upload_ticket", instructions["preferred_method"])
self.assertEqual(instructions["http_upload"]["bearer_auth"], "admin_or_local_only")
self.assertNotIn("Authorization: Bearer", instructions["http_upload"]["curl_example"])
self.assertEqual(instructions["direct_text_tool"]["name"], "ingest_text_note")
def test_create_ingest_upload_ticket_prefers_explicit_upload_url(self):
class FakeMCP:
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict(
os.environ,
{"OBSIDIAN_MCP_INGEST_PUBLIC_UPLOAD_URL": "https://uploads.example.test/custom/upload"},
clear=False,
),
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
mcp_server.create_server(
vault_path=root / "vault",
enable_ingest=True,
project_root=root / "project",
ingest_runtime_root=root / "runtime",
public_base_url="https://wiki.example.test/proxy/mcp",
)
capabilities = FakeMCP.tools["get_mcp_capabilities"]()
ticket = FakeMCP.tools["create_ingest_upload_ticket"](
expires_seconds=60,
max_files=1,
start_job=False,
)
instructions = FakeMCP.tools["get_ingest_upload_instructions"]()
self.assertEqual(capabilities["ingest"]["http_upload_url"], "https://uploads.example.test/custom/upload")
self.assertEqual(ticket["upload_url"], "https://uploads.example.test/custom/upload")
self.assertEqual(instructions["http_upload"]["url"], "https://uploads.example.test/custom/upload")
def test_ingest_tools_validate_upload_and_preview_metadata(self):
class FakeMCP:
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
max_file_bytes=20,
)
mcp_server.create_server(
vault_path=root / "vault",
enable_ingest=True,
project_root=root / "project",
ingest_runtime_root=root / "runtime",
ingest_manager=manager,
)
validate = FakeMCP.tools["validate_upload"](
files=[
{"file_name": "ok.pdf", "size_bytes": 2, "sha256": "a" * 64},
{"file_name": "large.pdf", "size_bytes": 21, "sha256": "b" * 64},
],
project="bjd",
doc_type_hint="drawing",
)
upload = manager.create_completed_upload("ok.pdf", b"ok", project="bjd", doc_type_hint="drawing")
preview = FakeMCP.tools["preview_upload_metadata"]([upload["upload_id"]])
self.assertFalse(validate["ok"])
self.assertEqual(validate["files"][0]["status"], "ok")
self.assertEqual(validate["files"][1]["status"], "error")
self.assertTrue(preview["ok"])
self.assertEqual(preview["uploads"][0]["file_name"], "ok.pdf")
self.assertEqual(preview["next_action"], "start_ingest_job")
def test_ingest_text_note_tool_creates_completed_markdown_upload(self):
class FakeMCP:
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
max_file_bytes=200,
)
mcp_server.create_server(
vault_path=root / "vault",
enable_ingest=True,
project_root=root / "project",
ingest_runtime_root=root / "runtime",
ingest_manager=manager,
)
result = FakeMCP.tools["ingest_text_note"](
text="Remote agent thought\n\nNeed to remember this.",
title="Customer thought: ACME / risk",
project="bjd",
doc_type_hint="thought",
note="captured from remote agent",
start_job=False,
)
payload_path = Path(manager._load_upload(result["uploads"][0]["upload_id"])["path"])
payload = payload_path.read_text(encoding="utf-8")
self.assertEqual(result["job"], None)
self.assertEqual(len(result["uploads"]), 1)
self.assertEqual(result["bundle"]["source_type"], "text_note")
self.assertEqual(
result["bundle"]["body_upload_id"],
result["uploads"][0]["upload_id"],
)
upload = result["uploads"][0]
self.assertEqual(upload["state"], "complete")
self.assertEqual(upload["project"], "bjd")
self.assertEqual(upload["doc_type_hint"], "thought")
self.assertEqual(upload["bundle_role"], "primary")
self.assertNotIn("path", upload)
self.assertTrue(upload["file_name"].endswith(".md"))
self.assertNotIn("/", upload["file_name"])
self.assertIn("# Customer thought: ACME / risk", payload)
self.assertIn("Remote agent thought", payload)
def test_ingest_text_note_tool_rejects_empty_and_oversized_text(self):
class FakeMCP:
tools = {}
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tools[func.__name__] = func
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
root = Path(tmp)
manager = IngestManager(
vault_path=root / "vault",
project_root=root / "project",
runtime_root=root / "runtime",
max_file_bytes=20,
)
mcp_server.create_server(
vault_path=root / "vault",
enable_ingest=True,
project_root=root / "project",
ingest_runtime_root=root / "runtime",
ingest_manager=manager,
)
with self.assertRaises(IngestError):
FakeMCP.tools["ingest_text_note"](text=" ", start_job=False)
with self.assertRaises(IngestError):
FakeMCP.tools["ingest_text_note"](text="x" * 21, start_job=False)
def test_create_server_can_disable_file_asset_tools(self):
class FakeMCP:
tool_names = []
prompt_names = []
def __init__(self, _name, **_kwargs):
pass
def tool(self):
def decorator(func):
self.tool_names.append(func.__name__)
return func
return decorator
def resource(self, *_args, **_kwargs):
def decorator(func):
return func
return decorator
def prompt(self, name=None, **_kwargs):
def decorator(func):
self.prompt_names.append(name or func.__name__)
return func
return decorator
with (
tempfile.TemporaryDirectory() as tmp,
patch.dict("sys.modules", {"mcp.server.fastmcp": type("Mod", (), {"FastMCP": FakeMCP})()}),
):
server = mcp_server.create_server(vault_path=Path(tmp), enable_file_assets=False)
self.assertIsInstance(server, FakeMCP)
self.assertIn("file_index_status", FakeMCP.tool_names)
self.assertNotIn("search_file_assets", FakeMCP.tool_names)
self.assertNotIn("list_file_assets", FakeMCP.tool_names)
self.assertNotIn("get_file_asset_metadata", FakeMCP.tool_names)
self.assertNotIn("how_to_use_wiki_file_assets", FakeMCP.prompt_names)
def test_run_http_server_fails_before_server_creation_without_token(self):
with tempfile.TemporaryDirectory() as tmp:
with patch.dict(os.environ, {}, clear=True):
with patch.object(mcp_server, "create_server") as create_server:
with self.assertRaises(RuntimeError):
mcp_server.run_http_server(
vault_path=Path(tmp),
db_path=None,
profile_name="bjd-general-bot",
max_file_bytes=1024,
expose_paths=False,
host="127.0.0.1",
port=8765,
token_env="TEST_MCP_TOKEN",
streamable_http_path="/mcp",
allowed_hosts=None,
allowed_origins=None,
)
create_server.assert_not_called()
def test_main_forwards_managed_security_and_timeout_environment(self):
with (
patch.dict(
os.environ,
{
"OBSIDIAN_MCP_TRANSPORT": "streamable-http",
"OBSIDIAN_MCP_ALLOW_SOURCE_FILES": "1",
"OBSIDIAN_MCP_ALLOWED_HOSTS": "wiki.example:*,10.0.0.2:*",
"OBSIDIAN_MCP_ALLOWED_ORIGINS": "https://app.example",
"OBSIDIAN_MCP_INGEST_JOB_TIMEOUT_SECONDS": "123",
"OBSIDIAN_MCP_QUERY_JOB_TIMEOUT_SECONDS": "45",
"OBSIDIAN_MCP_INGEST_MAX_REQUEST_BYTES": "999",
},
clear=True,
),
patch.object(mcp_server, "resolve_vault_path", return_value="/tmp/test-vault"),
patch.object(mcp_server, "run_http_server") as run_http_server,
):
result = mcp_server.main([])
self.assertEqual(result, 0)
kwargs = run_http_server.call_args.kwargs
self.assertTrue(kwargs["allow_source_files"])
self.assertEqual(kwargs["allowed_hosts"], ["wiki.example:*", "10.0.0.2:*"])
self.assertEqual(kwargs["allowed_origins"], ["https://app.example"])
self.assertEqual(kwargs["ingest_job_timeout_seconds"], 123)
self.assertEqual(kwargs["query_job_timeout_seconds"], 45)
self.assertEqual(kwargs["ingest_max_request_bytes"], 999)
def test_main_closes_job_managers_after_stdio_server_stops(self):
manager = type("Manager", (), {"close": unittest.mock.Mock()})()
class FakeServer:
_obsidian_job_managers = (manager,)
def run(self, **_kwargs):
return None
with (
patch.dict(os.environ, {"OBSIDIAN_MCP_TRANSPORT": "stdio"}, clear=True),
patch.object(mcp_server, "resolve_vault_path", return_value="/tmp/test-vault"),
patch.object(mcp_server, "create_server", return_value=FakeServer()),
):
result = mcp_server.main([])
self.assertEqual(result, 0)
manager.close.assert_called_once_with()
def test_close_server_managers_attempts_every_manager(self):
first = type("Manager", (), {"close": lambda self: (_ for _ in ()).throw(OSError("disk full"))})()
second = type("Manager", (), {"close": unittest.mock.Mock()})()
server = type("Server", (), {"_obsidian_job_managers": (first, second)})()
errors = mcp_server.close_server_managers(server)
self.assertEqual(len(errors), 1)
second.close.assert_called_once_with()
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink