import asyncio import base64 import hashlib import json import os import tempfile import unittest from pathlib import Path from unittest.mock import patch from obsidian_wiki import mcp_server from obsidian_wiki.mcp_ingest import 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""): messages = [] sent = False scope = { "type": "http", "method": method, "path": path, "raw_path": path.encode("ascii"), "query_string": b"", "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) 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"}) 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()) 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("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 = [] 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, *_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("ingest_file_base64", FakeMCP.tool_names) def test_ingest_file_base64_tool_uploads_without_chunking(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", ) payload = b"one shot upload" result = FakeMCP.tools["ingest_file_base64"]( file_name="source.pdf", data_base64=base64.b64encode(payload).decode("ascii"), project="bjd", doc_type_hint="drawing", start_job=False, ) self.assertEqual(result["job"], None) self.assertEqual(result["upload"]["state"], "complete") self.assertEqual(result["upload"]["file_name"], "source.pdf") self.assertEqual(result["upload"]["sha256"], "sha256:" + hashlib.sha256(payload).hexdigest()) 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() if __name__ == "__main__": unittest.main()