gabriel / musehub public
test_mcp_dispatcher.py python
445 lines 16.9 KB
Raw
8b8f8144 fix: MEDIUM security patch (M1–M7) — dev → main (#27) Gabriel Cardona <cgcardona@gmail.com> 2d ago
1 """Tests for the MuseHub MCP dispatcher, resources, and prompts.
2
3 Covers:
4 - JSON-RPC 2.0 protocol correctness (initialize, tools/list, resources/list,
5 resources/templates/list, prompts/list, prompts/get, ping, unknown method)
6 - tools/call routing: known read tools, unknown tool, write tool auth gate
7 - resources/read: musehub:// URI dispatch and unknown URI handling
8 - prompts/get: known prompt assembly and unknown prompt error
9 - Batch request handling
10 - Notification handling (no id → returns None)
11 - Tool catalogue completeness (41 tools)
12 - Resource catalogue completeness (12 static, 17 templated)
13 - Prompt catalogue completeness (10 prompts)
14 - MCP 2025-11-25: elicitation capability in initialize, new notifications
15 """
16 from __future__ import annotations
17
18 import json
19 from unittest.mock import AsyncMock, MagicMock, patch
20
21 import pytest
22
23 from musehub.mcp.dispatcher import handle_batch, handle_request
24 from musehub.mcp.prompts import PROMPT_CATALOGUE, get_prompt
25 from musehub.mcp.resources import RESOURCE_TEMPLATES, STATIC_RESOURCES, read_resource
26 from musehub.mcp.tools import MCP_TOOLS, MUSEHUB_WRITE_TOOL_NAMES
27
28
29 # ── Helpers ───────────────────────────────────────────────────────────────────
30
31
32 def _req(method: str, params: dict | None = None, req_id: int = 1) -> dict:
33 """Build a minimal JSON-RPC 2.0 request dict."""
34 msg: dict = {"jsonrpc": "2.0", "id": req_id, "method": method}
35 if params is not None:
36 msg["params"] = params
37 return msg
38
39
40 def _notification(method: str, params: dict | None = None) -> dict:
41 """Build a JSON-RPC 2.0 notification (no id)."""
42 msg: dict = {"jsonrpc": "2.0", "method": method}
43 if params is not None:
44 msg["params"] = params
45 return msg
46
47
48 # ── Protocol correctness ──────────────────────────────────────────────────────
49
50
51 @pytest.mark.asyncio
52 async def test_initialize_returns_capabilities() -> None:
53 """initialize should return protocolVersion 2025-11-25 and capabilities."""
54 resp = await handle_request(_req("initialize", {"protocolVersion": "2025-11-25"}))
55 assert resp is not None
56 assert resp["jsonrpc"] == "2.0"
57 assert resp["id"] == 1
58 result = resp["result"]
59 assert isinstance(result, dict)
60 assert result["protocolVersion"] == "2025-11-25"
61 assert "capabilities" in result
62 assert "tools" in result["capabilities"]
63 assert "resources" in result["capabilities"]
64 assert "prompts" in result["capabilities"]
65 assert "elicitation" in result["capabilities"]
66 assert "form" in result["capabilities"]["elicitation"]
67 assert "url" in result["capabilities"]["elicitation"]
68 # serverInfo must only contain name and version (not capabilities)
69 assert "name" in result["serverInfo"]
70 assert "version" in result["serverInfo"]
71 assert "capabilities" not in result["serverInfo"]
72
73
74 @pytest.mark.asyncio
75 async def test_ping_returns_empty_result() -> None:
76 """ping should return an empty result dict."""
77 resp = await handle_request(_req("ping"))
78 assert resp is not None
79 assert resp["result"] == {}
80
81
82 @pytest.mark.asyncio
83 async def test_unknown_method_returns_error() -> None:
84 """Unknown methods should return a JSON-RPC method-not-found error."""
85 resp = await handle_request(_req("musehub/does-not-exist"))
86 assert resp is not None
87 assert "error" in resp
88 assert resp["error"]["code"] == -32601
89
90
91 @pytest.mark.asyncio
92 async def test_completions_complete_returns_empty() -> None:
93 """completions/complete stub returns empty values list (MCP 2025-11-25)."""
94 resp = await handle_request(_req("completions/complete", {"ref": {}, "argument": {"name": "x", "value": "y"}}))
95 assert resp is not None
96 assert "result" in resp
97 assert resp["result"]["completion"]["values"] == []
98
99
100 @pytest.mark.asyncio
101 async def test_logging_set_level_requires_auth_anonymous_rejected() -> None:
102 """logging/setLevel must return an error for unauthenticated callers (M5)."""
103 resp = await handle_request(_req("logging/setLevel", {"level": "info"}), user_id=None)
104 assert resp is not None
105 assert "error" in resp
106
107
108 @pytest.mark.asyncio
109 async def test_logging_set_level_authenticated_returns_empty() -> None:
110 """logging/setLevel returns empty result for authenticated callers (MCP 2025-11-25)."""
111 resp = await handle_request(_req("logging/setLevel", {"level": "info"}), user_id="test-user")
112 assert resp is not None
113 assert resp["result"] == {}
114
115
116 @pytest.mark.asyncio
117 async def test_notification_returns_none() -> None:
118 """Notifications (no id) should return None from handle_request."""
119 result = await handle_request(_notification("ping"))
120 assert result is None
121
122
123 # ── Tool catalogue ────────────────────────────────────────────────────────────
124
125
126 @pytest.mark.asyncio
127 async def test_tools_list_returns_40_tools() -> None:
128 """tools/list should return all 40 registered tools."""
129 resp = await handle_request(_req("tools/list"))
130 assert resp is not None
131 result = resp["result"]
132 assert isinstance(result, dict)
133 tools = result["tools"]
134 assert isinstance(tools, list)
135 assert len(tools) == 41 # 20 read + 16 write + 5 elicitation
136
137
138 @pytest.mark.asyncio
139 async def test_tools_list_no_server_side_field() -> None:
140 """tools/list should strip the internal server_side field."""
141 resp = await handle_request(_req("tools/list"))
142 assert resp is not None
143 for tool in resp["result"]["tools"]:
144 assert "server_side" not in tool, f"Tool {tool['name']} exposes server_side"
145
146
147 @pytest.mark.asyncio
148 async def test_tools_list_all_have_required_fields() -> None:
149 """Every tool in tools/list must have name, description, inputSchema, and annotations."""
150 resp = await handle_request(_req("tools/list"))
151 assert resp is not None
152 for tool in resp["result"]["tools"]:
153 assert "name" in tool, f"Missing name: {tool}"
154 assert "description" in tool, f"Missing description for {tool.get('name')}"
155 assert "inputSchema" in tool, f"Missing inputSchema for {tool.get('name')}"
156 assert "annotations" in tool, f"Missing MCP 2025-11-25 annotations for {tool.get('name')}"
157
158
159 def test_tool_catalogue_has_40_tools() -> None:
160 """The MCP_TOOLS list must contain exactly 40 tools."""
161 assert len(MCP_TOOLS) == 41
162
163
164 def test_write_tool_names_all_in_catalogue() -> None:
165 """Every write tool name must appear in the full catalogue."""
166 all_names = {t["name"] for t in MCP_TOOLS}
167 for name in MUSEHUB_WRITE_TOOL_NAMES:
168 assert name in all_names, f"Write tool {name!r} not in MCP_TOOLS"
169
170
171 # ── tools/call routing ────────────────────────────────────────────────────────
172
173
174 @pytest.mark.asyncio
175 async def test_tools_call_unknown_tool_returns_iserror() -> None:
176 """Calling an unknown tool should return isError=true (not a JSON-RPC error)."""
177 resp = await handle_request(
178 _req("tools/call", {"name": "nonexistent_tool", "arguments": {}})
179 )
180 assert resp is not None
181 # Envelope is success (has "result", not "error")
182 assert "result" in resp
183 result = resp["result"]
184 assert result.get("isError") is True
185
186
187 @pytest.mark.asyncio
188 async def test_tools_call_write_tool_requires_auth() -> None:
189 """Calling a write tool without a user_id should return isError=true."""
190 resp = await handle_request(
191 _req("tools/call", {"name": "musehub_create_repo", "arguments": {"name": "test"}}),
192 user_id=None,
193 )
194 assert resp is not None
195 assert "result" in resp
196 assert resp["result"].get("isError") is True
197
198
199 @pytest.mark.asyncio
200 async def test_tools_call_write_tool_passes_with_auth() -> None:
201 """Calling a write tool with user_id should reach the executor (not auth-gate)."""
202 mock_result = MagicMock()
203 mock_result.ok = True
204 mock_result.data = {"repo_id": "test-123", "name": "Test", "slug": "test",
205 "owner": "alice", "visibility": "public", "clone_url": "musehub://alice/test",
206 "created_at": None}
207
208 with patch(
209 "musehub.mcp.write_tools.repos.execute_create_repo",
210 new_callable=AsyncMock,
211 return_value=mock_result,
212 ):
213 resp = await handle_request(
214 _req("tools/call", {"name": "musehub_create_repo", "arguments": {"name": "Test"}}),
215 user_id="alice",
216 )
217 assert resp is not None
218 assert "result" in resp
219 assert resp["result"].get("isError") is False
220
221
222 @pytest.mark.asyncio
223 async def test_tools_call_read_tool_with_mock_executor() -> None:
224 """Read tools should delegate to the executor and return text content."""
225 mock_result = MagicMock()
226 mock_result.ok = True
227 mock_result.data = {"repo_id": "r123", "branches": []}
228
229 with patch(
230 "musehub.services.musehub_mcp_executor.execute_list_branches",
231 new_callable=AsyncMock,
232 return_value=mock_result,
233 ):
234 resp = await handle_request(
235 _req("tools/call", {"name": "musehub_list_branches", "arguments": {"repo_id": "r123"}})
236 )
237
238 assert resp is not None
239 assert "result" in resp
240 result = resp["result"]
241 assert result.get("isError") is False
242 content = result["content"]
243 assert isinstance(content, list)
244 assert len(content) == 1
245 assert content[0]["type"] == "text"
246 # Text should be valid JSON
247 data = json.loads(content[0]["text"])
248 assert data["repo_id"] == "r123"
249
250
251 # ── Resource catalogue ────────────────────────────────────────────────────────
252
253
254 @pytest.mark.asyncio
255 async def test_resources_list_returns_12_static() -> None:
256 """resources/list should return all 12 static resources (musehub:// + muse:// docs/domains)."""
257 resp = await handle_request(_req("resources/list"))
258 assert resp is not None
259 resources = resp["result"]["resources"]
260 assert len(resources) == 12
261
262
263 @pytest.mark.asyncio
264 async def test_resources_templates_list_returns_17_templates() -> None:
265 """resources/templates/list should return the 17 URI templates."""
266 resp = await handle_request(_req("resources/templates/list"))
267 assert resp is not None
268 templates = resp["result"]["resourceTemplates"]
269 assert len(templates) == 17
270
271
272 def test_static_resources_have_required_fields() -> None:
273 """Each static resource must have uri, name, and mimeType."""
274 _VALID_PREFIXES = ("musehub://", "muse://")
275 for r in STATIC_RESOURCES:
276 assert "uri" in r
277 assert "name" in r
278 assert r["uri"].startswith(_VALID_PREFIXES), f"Unexpected URI scheme: {r['uri']}"
279
280
281 def test_resource_templates_have_required_fields() -> None:
282 """Each resource template must have uriTemplate, name, and mimeType."""
283 _VALID_PREFIXES = ("musehub://", "muse://")
284 for t in RESOURCE_TEMPLATES:
285 assert "uriTemplate" in t
286 assert "name" in t
287 assert t["uriTemplate"].startswith(_VALID_PREFIXES), f"Unexpected URI scheme: {t['uriTemplate']}"
288
289
290 @pytest.mark.asyncio
291 async def test_resources_read_unknown_uri_returns_error_content() -> None:
292 """resources/read with an unknown URI should return an error in the text content."""
293 resp = await handle_request(
294 _req("resources/read", {"uri": "musehub://nonexistent/path/that/does/not/exist"})
295 )
296 assert resp is not None
297 assert "result" in resp
298 contents = resp["result"]["contents"]
299 assert isinstance(contents, list)
300 assert len(contents) == 1
301 data = json.loads(contents[0]["text"])
302 assert "error" in data
303
304
305 @pytest.mark.asyncio
306 async def test_resources_read_missing_uri_returns_error() -> None:
307 """resources/read without a uri parameter should return an InvalidParams error."""
308 resp = await handle_request(_req("resources/read", {}))
309 assert resp is not None
310 assert "error" in resp
311 assert resp["error"]["code"] == -32602
312
313
314 @pytest.mark.asyncio
315 async def test_resources_read_unsupported_scheme() -> None:
316 """resources/read with a non-musehub:// URI should return an error in content."""
317 result = await read_resource("https://example.com/foo")
318 assert "error" in result
319
320
321 @pytest.mark.asyncio
322 async def test_resources_read_me_requires_auth() -> None:
323 """musehub://me should return an error when user_id is None."""
324 from musehub.mcp.resources import _read_me
325 result = await _read_me(None)
326 assert "error" in result
327
328
329 # ── Prompt catalogue ──────────────────────────────────────────────────────────
330
331
332 @pytest.mark.asyncio
333 async def test_prompts_list_returns_10_prompts() -> None:
334 """prompts/list should return all 10 workflow prompts."""
335 resp = await handle_request(_req("prompts/list"))
336 assert resp is not None
337 prompts = resp["result"]["prompts"]
338 assert len(prompts) == 10
339
340
341 def test_prompt_catalogue_completeness() -> None:
342 """PROMPT_CATALOGUE must have exactly 10 entries."""
343 assert len(PROMPT_CATALOGUE) == 10
344
345
346 def test_prompt_names_are_correct() -> None:
347 """All 10 expected prompt names must be present."""
348 names = {p["name"] for p in PROMPT_CATALOGUE}
349 assert "musehub/orientation" in names
350 assert "musehub/contribute" in names
351 assert "musehub/create" in names
352 assert "musehub/review_pr" in names
353 assert "musehub/issue_triage" in names
354 assert "musehub/release_prep" in names
355 assert "musehub/onboard" in names
356 assert "musehub/release_to_world" in names
357 assert "musehub/domain-discovery" in names
358 assert "musehub/domain-authoring" in names
359
360
361 @pytest.mark.asyncio
362 async def test_prompts_get_orientation_returns_messages() -> None:
363 """prompts/get for musehub/orientation should return messages."""
364 resp = await handle_request(
365 _req("prompts/get", {"name": "musehub/orientation", "arguments": {}})
366 )
367 assert resp is not None
368 assert "result" in resp
369 result = resp["result"]
370 assert "messages" in result
371 messages = result["messages"]
372 assert len(messages) == 2
373 assert messages[0]["role"] == "user"
374 assert messages[1]["role"] == "assistant"
375
376
377 @pytest.mark.asyncio
378 async def test_prompts_get_contribute_interpolates_args() -> None:
379 """prompts/get for musehub/contribute should accept repo_id, owner, slug args."""
380 resp = await handle_request(
381 _req("prompts/get", {
382 "name": "musehub/contribute",
383 "arguments": {"repo_id": "abc-123", "owner": "alice", "slug": "jazz-session"},
384 })
385 )
386 assert resp is not None
387 assert "result" in resp
388 text = resp["result"]["messages"][1]["content"]["text"]
389 assert "jazz-session" in text
390
391
392 @pytest.mark.asyncio
393 async def test_prompts_get_unknown_returns_method_not_found() -> None:
394 """prompts/get for an unknown name should return a -32601 JSON-RPC error."""
395 resp = await handle_request(
396 _req("prompts/get", {"name": "musehub/nonexistent"})
397 )
398 assert resp is not None
399 assert "error" in resp
400 assert resp["error"]["code"] == -32601
401
402
403 def test_get_prompt_all_prompts_assemble() -> None:
404 """All 10 prompts should assemble without raising exceptions."""
405 for prompt_def in PROMPT_CATALOGUE:
406 name = prompt_def["name"]
407 result = get_prompt(name, {"repo_id": "test-id", "pr_id": "pr-id", "owner": "user", "slug": "repo"})
408 assert result is not None, f"get_prompt({name!r}) returned None"
409 assert "messages" in result
410 assert len(result["messages"]) >= 2
411
412
413 def test_get_prompt_unknown_returns_none() -> None:
414 """get_prompt for an unknown name should return None."""
415 result = get_prompt("musehub/unknown")
416 assert result is None
417
418
419 # ── Batch handling ────────────────────────────────────────────────────────────
420
421
422 @pytest.mark.asyncio
423 async def test_batch_handles_multiple_requests() -> None:
424 """handle_batch should return responses for all non-notifications."""
425 batch = [
426 _req("initialize", {"protocolVersion": "2025-03-26"}, req_id=1),
427 _req("tools/list", req_id=2),
428 _req("prompts/list", req_id=3),
429 ]
430 responses = await handle_batch(batch)
431 assert len(responses) == 3
432 ids = {r["id"] for r in responses}
433 assert ids == {1, 2, 3}
434
435
436 @pytest.mark.asyncio
437 async def test_batch_excludes_notifications() -> None:
438 """handle_batch should not include responses for notifications."""
439 batch = [
440 _req("ping", req_id=1),
441 _notification("ping"), # no id → no response
442 ]
443 responses = await handle_batch(batch)
444 assert len(responses) == 1
445 assert responses[0]["id"] == 1