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_stages_without_starting_job_by_default(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( {"project": "bjd"}, [("files", "drawing.pdf", "application/pdf", b"drawing")], ) with patch.object( manager, "start_ingest_job", side_effect=AssertionError("default upload staging must not start a job"), ) as start_job: 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"]) start_job.assert_not_called() 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", ) 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 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.assertFalse(instructions["http_upload"]["start_job_default"]) self.assertFalse(instructions["direct_text_tool"]["start_job_default"]) self.assertIn("start_job=false", instructions["http_upload"]["curl_example"]) 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, ) with patch.object( manager, "start_ingest_job", side_effect=AssertionError("default text staging must not start a job"), ) as start_job: 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.assert_not_called() 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()