test_mcp_elicitation.py
python
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a
init: musehub initial commit
Human
171 days ago
| 1 | """Tests for MCP 2025-11-25 Elicitation: ToolCallContext, session, and tools. |
| 2 | |
| 3 | Covers: |
| 4 | ToolCallContext: |
| 5 | - elicit_form: accept, decline, cancel, timeout, no-session fallback |
| 6 | - elicit_url: accept, decline, no-session fallback |
| 7 | - progress: session push, no-session no-op |
| 8 | |
| 9 | Session elicitation helpers: |
| 10 | - create_pending_elicitation, resolve_elicitation, cancel_elicitation |
| 11 | |
| 12 | Elicitation schemas: |
| 13 | - SCHEMAS contains all expected keys |
| 14 | - build_form_elicitation returns correct mode/requestedSchema |
| 15 | - build_url_elicitation returns correct mode/url/elicitationId |
| 16 | |
| 17 | Tool routing (unit): |
| 18 | - musehub_review_proposal_interactive: no session → schema_guide |
| 19 | - musehub_create_release_interactive: no session → schema_guide |
| 20 | |
| 21 | New prompts: |
| 22 | - musehub/onboard assembles correctly |
| 23 | |
| 24 | SSE formatting: |
| 25 | - sse_event produces correct format |
| 26 | - sse_notification produces correct JSON-RPC notification |
| 27 | - sse_request produces correct JSON-RPC request with id |
| 28 | - sse_response produces correct JSON-RPC response |
| 29 | """ |
| 30 | from __future__ import annotations |
| 31 | |
| 32 | import asyncio |
| 33 | import json |
| 34 | from unittest.mock import AsyncMock, MagicMock, patch |
| 35 | |
| 36 | import pytest |
| 37 | |
| 38 | from musehub.mcp.context import ToolCallContext |
| 39 | from musehub.mcp.elicitation import ( |
| 40 | SCHEMAS, |
| 41 | build_form_elicitation, |
| 42 | build_url_elicitation, |
| 43 | oauth_connect_url, |
| 44 | daw_cloud_connect_url, |
| 45 | ) |
| 46 | from musehub.mcp.prompts import PROMPT_CATALOGUE, get_prompt |
| 47 | from musehub.mcp.session import ( |
| 48 | MCPSession, |
| 49 | create_session, |
| 50 | create_pending_elicitation, |
| 51 | delete_session, |
| 52 | resolve_elicitation, |
| 53 | cancel_elicitation, |
| 54 | push_to_session, |
| 55 | ) |
| 56 | from musehub.mcp.sse import ( |
| 57 | sse_event, |
| 58 | sse_notification, |
| 59 | sse_request, |
| 60 | sse_response, |
| 61 | ) |
| 62 | |
| 63 | |
| 64 | # ── SSE formatting ──────────────────────────────────────────────────────────── |
| 65 | |
| 66 | |
| 67 | def test_sse_event_basic_format() -> None: |
| 68 | """sse_event should produce 'data: <json>\\n\\n'.""" |
| 69 | result = sse_event({"jsonrpc": "2.0", "method": "ping"}) |
| 70 | assert result.startswith("data:") |
| 71 | assert result.endswith("\n\n") |
| 72 | # Extract data line and parse JSON |
| 73 | data_line = [l for l in result.split("\n") if l.startswith("data:")][0] |
| 74 | payload = json.loads(data_line[len("data: "):]) |
| 75 | assert payload["method"] == "ping" |
| 76 | |
| 77 | |
| 78 | def test_sse_event_with_id_and_type() -> None: |
| 79 | """sse_event with event_id and event_type should include id: and event: lines.""" |
| 80 | result = sse_event({"a": 1}, event_id="42", event_type="notification") |
| 81 | assert "id: 42\n" in result |
| 82 | assert "event: notification\n" in result |
| 83 | |
| 84 | |
| 85 | def test_sse_notification_format() -> None: |
| 86 | """sse_notification should produce a valid JSON-RPC 2.0 notification.""" |
| 87 | result = sse_notification("notifications/progress", {"progress": 50}) |
| 88 | data_line = [l for l in result.split("\n") if l.startswith("data:")][0] |
| 89 | payload = json.loads(data_line[len("data: "):]) |
| 90 | assert payload["jsonrpc"] == "2.0" |
| 91 | assert payload["method"] == "notifications/progress" |
| 92 | assert payload["params"]["progress"] == 50 |
| 93 | assert "id" not in payload # notifications have no id |
| 94 | |
| 95 | |
| 96 | def test_sse_request_format() -> None: |
| 97 | """sse_request should produce a valid JSON-RPC 2.0 request with id.""" |
| 98 | result = sse_request("elicit-1", "elicitation/create", {"mode": "form"}) |
| 99 | data_line = [l for l in result.split("\n") if l.startswith("data:")][0] |
| 100 | payload = json.loads(data_line[len("data: "):]) |
| 101 | assert payload["jsonrpc"] == "2.0" |
| 102 | assert payload["id"] == "elicit-1" |
| 103 | assert payload["method"] == "elicitation/create" |
| 104 | assert payload["params"]["mode"] == "form" |
| 105 | |
| 106 | |
| 107 | def test_sse_response_format() -> None: |
| 108 | """sse_response should produce a valid JSON-RPC 2.0 success response.""" |
| 109 | result = sse_response(42, {"content": [{"type": "text", "text": "ok"}]}) |
| 110 | data_line = [l for l in result.split("\n") if l.startswith("data:")][0] |
| 111 | payload = json.loads(data_line[len("data: "):]) |
| 112 | assert payload["jsonrpc"] == "2.0" |
| 113 | assert payload["id"] == 42 |
| 114 | assert "result" in payload |
| 115 | assert "error" not in payload |
| 116 | |
| 117 | |
| 118 | # ── Elicitation schemas ─────────────────────────────────────────────────────── |
| 119 | |
| 120 | |
| 121 | def test_schemas_has_all_expected_keys() -> None: |
| 122 | """SCHEMAS must contain all 5 musical elicitation schemas.""" |
| 123 | expected = { |
| 124 | "compose_preferences", |
| 125 | "repo_creation", |
| 126 | "proposal_review_focus", |
| 127 | "release_metadata", |
| 128 | "platform_connect_confirm", |
| 129 | } |
| 130 | assert expected == set(SCHEMAS.keys()) |
| 131 | |
| 132 | |
| 133 | def test_compose_preferences_schema_required_fields() -> None: |
| 134 | """compose_preferences schema must declare correct required fields.""" |
| 135 | schema = SCHEMAS["compose_preferences"] |
| 136 | assert schema["type"] == "object" |
| 137 | required = schema["required"] |
| 138 | assert "key" in required |
| 139 | assert "tempo_bpm" in required |
| 140 | assert "mood" in required |
| 141 | assert "genre" in required |
| 142 | |
| 143 | |
| 144 | def test_build_form_elicitation() -> None: |
| 145 | """build_form_elicitation should return correct mode and requestedSchema.""" |
| 146 | params = build_form_elicitation("compose_preferences", "Pick your vibe") |
| 147 | assert params["mode"] == "form" |
| 148 | assert params["message"] == "Pick your vibe" |
| 149 | assert "requestedSchema" in params |
| 150 | assert params["requestedSchema"] is SCHEMAS["compose_preferences"] |
| 151 | |
| 152 | |
| 153 | def test_build_form_elicitation_unknown_key_raises() -> None: |
| 154 | """build_form_elicitation with unknown key should raise KeyError.""" |
| 155 | with pytest.raises(KeyError): |
| 156 | build_form_elicitation("nonexistent_schema", "message") |
| 157 | |
| 158 | |
| 159 | def test_build_url_elicitation() -> None: |
| 160 | """build_url_elicitation should return correct mode, url, and elicitationId.""" |
| 161 | params, eid = build_url_elicitation("https://example.com/oauth", "Connect Spotify") |
| 162 | assert params["mode"] == "url" |
| 163 | assert params["url"] == "https://example.com/oauth" |
| 164 | assert params["message"] == "Connect Spotify" |
| 165 | assert params["elicitationId"] == eid |
| 166 | assert len(eid) > 10 |
| 167 | |
| 168 | |
| 169 | def test_build_url_elicitation_stable_id() -> None: |
| 170 | """build_url_elicitation should use provided elicitation_id.""" |
| 171 | params, eid = build_url_elicitation( |
| 172 | "https://example.com/oauth", "msg", elicitation_id="my-stable-id" |
| 173 | ) |
| 174 | assert eid == "my-stable-id" |
| 175 | assert params["elicitationId"] == "my-stable-id" |
| 176 | |
| 177 | |
| 178 | def test_oauth_connect_url_format() -> None: |
| 179 | """oauth_connect_url should produce correct platform-specific MuseHub URL.""" |
| 180 | url = oauth_connect_url("Spotify", "abc123", base_url="https://musehub.ai") |
| 181 | assert "spotify" in url |
| 182 | assert "elicitation_id=abc123" in url |
| 183 | assert url.startswith("https://musehub.ai") |
| 184 | |
| 185 | |
| 186 | def test_daw_cloud_connect_url_format() -> None: |
| 187 | """daw_cloud_connect_url should produce correct service-specific URL.""" |
| 188 | url = daw_cloud_connect_url("LANDR", "xyz789", base_url="https://musehub.ai") |
| 189 | assert "landr" in url |
| 190 | assert "elicitation_id=xyz789" in url |
| 191 | |
| 192 | |
| 193 | # ── ToolCallContext — elicit_form ───────────────────────────────────────────── |
| 194 | |
| 195 | |
| 196 | @pytest.mark.anyio |
| 197 | async def test_elicit_form_no_session_returns_none() -> None: |
| 198 | """elicit_form without an active session should return None.""" |
| 199 | ctx = ToolCallContext(user_id=None, session=None) |
| 200 | result = await ctx.elicit_form(SCHEMAS["compose_preferences"], "msg") |
| 201 | assert result is None |
| 202 | |
| 203 | |
| 204 | @pytest.mark.anyio |
| 205 | async def test_elicit_form_accepted_returns_content() -> None: |
| 206 | """elicit_form should return content dict when user accepts.""" |
| 207 | session = create_session("user-1", {"elicitation": {"form": {}}}) |
| 208 | ctx = ToolCallContext(user_id="user-1", session=session) |
| 209 | |
| 210 | # Pre-resolve the Future before the elicit_form call awaits it. |
| 211 | content = {"key": "C major", "tempo_bpm": 120, "mood": "peaceful", "genre": "ambient"} |
| 212 | |
| 213 | async def _resolve_after_push() -> None: |
| 214 | await asyncio.sleep(0) # yield to let push happen |
| 215 | for req_id, fut in list(session.pending.items()): |
| 216 | resolve_elicitation(session, req_id, {"action": "accept", "content": content}) |
| 217 | |
| 218 | task = asyncio.create_task(_resolve_after_push()) |
| 219 | result = await ctx.elicit_form(SCHEMAS["compose_preferences"], "Pick your vibe") |
| 220 | await task |
| 221 | |
| 222 | assert result == content |
| 223 | delete_session(session.session_id) |
| 224 | |
| 225 | |
| 226 | @pytest.mark.anyio |
| 227 | async def test_elicit_form_declined_returns_none() -> None: |
| 228 | """elicit_form should return None when user declines.""" |
| 229 | session = create_session("user-1", {"elicitation": {"form": {}}}) |
| 230 | ctx = ToolCallContext(user_id="user-1", session=session) |
| 231 | |
| 232 | async def _decline_after_push() -> None: |
| 233 | await asyncio.sleep(0) |
| 234 | for req_id in list(session.pending.keys()): |
| 235 | resolve_elicitation(session, req_id, {"action": "decline"}) |
| 236 | |
| 237 | task = asyncio.create_task(_decline_after_push()) |
| 238 | result = await ctx.elicit_form(SCHEMAS["compose_preferences"], "Pick your vibe") |
| 239 | await task |
| 240 | |
| 241 | assert result is None |
| 242 | delete_session(session.session_id) |
| 243 | |
| 244 | |
| 245 | @pytest.mark.anyio |
| 246 | async def test_elicit_form_no_form_capability_returns_none() -> None: |
| 247 | """elicit_form should return None if client didn't declare form support.""" |
| 248 | session = create_session("user-1", {"elicitation": {"url": {}}}) # url only, no form |
| 249 | ctx = ToolCallContext(user_id="user-1", session=session) |
| 250 | result = await ctx.elicit_form(SCHEMAS["compose_preferences"], "msg") |
| 251 | assert result is None |
| 252 | delete_session(session.session_id) |
| 253 | |
| 254 | |
| 255 | # ── ToolCallContext — elicit_url ────────────────────────────────────────────── |
| 256 | |
| 257 | |
| 258 | @pytest.mark.anyio |
| 259 | async def test_elicit_url_no_session_returns_false() -> None: |
| 260 | """elicit_url without an active session should return False.""" |
| 261 | ctx = ToolCallContext(user_id=None, session=None) |
| 262 | result = await ctx.elicit_url("https://example.com/oauth", "msg") |
| 263 | assert result is False |
| 264 | |
| 265 | |
| 266 | @pytest.mark.anyio |
| 267 | async def test_elicit_url_accepted_returns_true() -> None: |
| 268 | """elicit_url should return True when user accepts the URL flow.""" |
| 269 | session = create_session("user-1", {"elicitation": {"form": {}, "url": {}}}) |
| 270 | ctx = ToolCallContext(user_id="user-1", session=session) |
| 271 | |
| 272 | async def _accept_after_push() -> None: |
| 273 | await asyncio.sleep(0) |
| 274 | for req_id in list(session.pending.keys()): |
| 275 | resolve_elicitation(session, req_id, {"action": "accept"}) |
| 276 | |
| 277 | task = asyncio.create_task(_accept_after_push()) |
| 278 | result = await ctx.elicit_url("https://example.com/oauth", "msg") |
| 279 | await task |
| 280 | |
| 281 | assert result is True |
| 282 | delete_session(session.session_id) |
| 283 | |
| 284 | |
| 285 | # ── ToolCallContext — progress ──────────────────────────────────────────────── |
| 286 | |
| 287 | |
| 288 | @pytest.mark.anyio |
| 289 | async def test_progress_no_session_is_noop() -> None: |
| 290 | """progress without an active session should not raise.""" |
| 291 | ctx = ToolCallContext(user_id=None, session=None) |
| 292 | await ctx.progress("token", 1, 10, "working…") # must not raise |
| 293 | |
| 294 | |
| 295 | @pytest.mark.anyio |
| 296 | async def test_progress_with_session_pushes_sse_event() -> None: |
| 297 | """progress with an active session should push a notifications/progress SSE event.""" |
| 298 | session = create_session("user-1", {}) |
| 299 | queue: asyncio.Queue[str | None] = asyncio.Queue() |
| 300 | session.sse_queues.append(queue) |
| 301 | |
| 302 | ctx = ToolCallContext(user_id="user-1", session=session) |
| 303 | await ctx.progress("compose-token", 2, 5, "generating…") |
| 304 | |
| 305 | assert not queue.empty() |
| 306 | event_text = queue.get_nowait() |
| 307 | assert event_text is not None |
| 308 | data_line = [l for l in event_text.split("\n") if l.startswith("data:")][0] |
| 309 | payload = json.loads(data_line[len("data: "):]) |
| 310 | assert payload["method"] == "notifications/progress" |
| 311 | assert payload["params"]["progress"] == 2 |
| 312 | assert payload["params"]["total"] == 5 |
| 313 | |
| 314 | delete_session(session.session_id) |
| 315 | |
| 316 | |
| 317 | # ── Elicitation tool executors — no session graceful degradation ────────────── |
| 318 | |
| 319 | |
| 320 | @pytest.mark.anyio |
| 321 | async def test_review_proposal_interactive_no_session() -> None: |
| 322 | """musehub_review_proposal_interactive without session must return error.""" |
| 323 | from musehub.mcp.write_tools.elicitation_tools import execute_review_proposal_interactive |
| 324 | |
| 325 | ctx = ToolCallContext(user_id=None, session=None) |
| 326 | result = await execute_review_proposal_interactive("repo-1", "proposal-1", ctx=ctx) |
| 327 | # No bypass params + no session → schema guide (ok=True, not an error) |
| 328 | assert result.ok is True |
| 329 | assert result.data is not None |
| 330 | assert result.data.get("mode") == "schema_guide" |
| 331 | |
| 332 | |
| 333 | @pytest.mark.anyio |
| 334 | async def test_create_release_interactive_no_session_no_params() -> None: |
| 335 | """musehub_create_release_interactive with no params + no session returns schema guide.""" |
| 336 | from musehub.mcp.write_tools.elicitation_tools import execute_create_release_interactive |
| 337 | |
| 338 | ctx = ToolCallContext(user_id=None, session=None) |
| 339 | result = await execute_create_release_interactive("repo-1", ctx=ctx) |
| 340 | assert result.ok is True |
| 341 | assert result.data is not None |
| 342 | assert result.data.get("mode") == "schema_guide" |
| 343 | assert "fields" in result.data |
| 344 | |
| 345 | |
| 346 | # ── New prompts ─────────────────────────────────────────────────────────────── |
| 347 | |
| 348 | |
| 349 | def test_onboard_prompt_assembles() -> None: |
| 350 | """musehub/onboard should assemble with 2 messages.""" |
| 351 | result = get_prompt("musehub/onboard", {"username": "alice"}) |
| 352 | assert result is not None |
| 353 | assert "messages" in result |
| 354 | assert len(result["messages"]) == 2 |
| 355 | assert result["messages"][0]["role"] == "user" |
| 356 | text = result["messages"][1]["content"]["text"] |
| 357 | assert "alice" in text |
| 358 | assert "elicitation" in text.lower() or "elicit" in text.lower() or "compose" in text.lower() |
| 359 | |
| 360 | |
| 361 | def test_onboard_prompt_in_catalogue() -> None: |
| 362 | """musehub/onboard must be in the prompt catalogue.""" |
| 363 | names = {p["name"] for p in PROMPT_CATALOGUE} |
| 364 | assert "musehub/onboard" in names |
| 365 | |
| 366 | |
| 367 | # ── Dispatcher routing — new notifications (2025-11-25) ────────────────────── |
| 368 | |
| 369 | |
| 370 | @pytest.mark.anyio |
| 371 | async def test_notifications_cancelled_handled() -> None: |
| 372 | """notifications/cancelled should be handled as a notification (return None).""" |
| 373 | from musehub.mcp.dispatcher import handle_request |
| 374 | |
| 375 | resp = await handle_request({ |
| 376 | "jsonrpc": "2.0", |
| 377 | "method": "notifications/cancelled", |
| 378 | "params": {"requestId": "elicit-1", "reason": "user navigated away"}, |
| 379 | }) |
| 380 | assert resp is None # notifications return None |
| 381 | |
| 382 | |
| 383 | @pytest.mark.anyio |
| 384 | async def test_notifications_elicitation_complete_handled() -> None: |
| 385 | """notifications/elicitation/complete should be handled as a notification.""" |
| 386 | from musehub.mcp.dispatcher import handle_request |
| 387 | |
| 388 | resp = await handle_request({ |
| 389 | "jsonrpc": "2.0", |
| 390 | "method": "notifications/elicitation/complete", |
| 391 | "params": {"elicitationId": "abc-xyz"}, |
| 392 | }) |
| 393 | assert resp is None |
| 394 | |
| 395 | |
| 396 | @pytest.mark.anyio |
| 397 | async def test_notifications_cancelled_resolves_future() -> None: |
| 398 | """notifications/cancelled with a session should cancel the pending Future.""" |
| 399 | from musehub.mcp.dispatcher import handle_request |
| 400 | |
| 401 | session = create_session(None, {"elicitation": {"form": {}}}) |
| 402 | fut = create_pending_elicitation(session, "elicit-99") |
| 403 | |
| 404 | await handle_request( |
| 405 | { |
| 406 | "jsonrpc": "2.0", |
| 407 | "method": "notifications/cancelled", |
| 408 | "params": {"requestId": "elicit-99"}, |
| 409 | }, |
| 410 | session=session, |
| 411 | ) |
| 412 | |
| 413 | assert fut.cancelled() |
| 414 | delete_session(session.session_id) |
| 415 | |
| 416 | |
| 417 | # ── Bypass path tests (no session, params provided directly) ────────────────── |
| 418 | |
| 419 | |
| 420 | @pytest.mark.anyio |
| 421 | async def test_review_proposal_interactive_bypass_dimension_and_depth() -> None: |
| 422 | """Bypass: dimension+depth skip elicitation, run divergence analysis path.""" |
| 423 | from musehub.mcp.write_tools.elicitation_tools import execute_review_proposal_interactive |
| 424 | from musehub.services.musehub_mcp_executor import MusehubToolResult |
| 425 | |
| 426 | ctx = ToolCallContext(user_id=None, session=None) |
| 427 | |
| 428 | # DB is unavailable in unit tests — expect a graceful db_unavailable result, |
| 429 | # which proves the bypass path was taken (no elicitation_unavailable error). |
| 430 | result = await execute_review_proposal_interactive( |
| 431 | "repo-1", "proposal-1", |
| 432 | dimension="harmonic", |
| 433 | depth="thorough", |
| 434 | ctx=ctx, |
| 435 | ) |
| 436 | # Accepted bypass — should proceed to DB lookup, not hit schema_guide |
| 437 | assert result.data is None or result.data.get("mode") != "schema_guide" |
| 438 | # If DB is down the error_code is db_unavailable, not elicitation_unavailable |
| 439 | if not result.ok: |
| 440 | assert result.error_code != "elicitation_unavailable" |
| 441 | |
| 442 | |
| 443 | @pytest.mark.anyio |
| 444 | async def test_review_proposal_interactive_bypass_dimension_only() -> None: |
| 445 | """Bypass: dimension alone (depth defaults to standard) skips elicitation.""" |
| 446 | from musehub.mcp.write_tools.elicitation_tools import execute_review_proposal_interactive |
| 447 | |
| 448 | ctx = ToolCallContext(user_id=None, session=None) |
| 449 | result = await execute_review_proposal_interactive( |
| 450 | "repo-1", "proposal-2", |
| 451 | dimension="melodic", |
| 452 | ctx=ctx, |
| 453 | ) |
| 454 | # Must not be schema_guide — dimension was supplied so bypass took effect. |
| 455 | if result.data: |
| 456 | assert result.data.get("mode") != "schema_guide" |
| 457 | if not result.ok: |
| 458 | assert result.error_code != "elicitation_unavailable" |
| 459 | |
| 460 | |
| 461 | @pytest.mark.anyio |
| 462 | async def test_create_release_interactive_bypass_tag_only() -> None: |
| 463 | """Bypass: tag supplied directly — proceeds to execute_create_release (DB may fail).""" |
| 464 | from musehub.mcp.write_tools.elicitation_tools import execute_create_release_interactive |
| 465 | |
| 466 | ctx = ToolCallContext(user_id=None, session=None) |
| 467 | result = await execute_create_release_interactive( |
| 468 | "repo-1", |
| 469 | tag="v0.9.0", |
| 470 | title="Beta", |
| 471 | notes="First beta.", |
| 472 | ctx=ctx, |
| 473 | ) |
| 474 | # If DB is unavailable the create_release call fails, but it must NOT |
| 475 | # return elicitation_unavailable — that proves bypass was taken. |
| 476 | if not result.ok: |
| 477 | assert result.error_code != "elicitation_unavailable" |
| 478 | if result.data: |
| 479 | assert result.data.get("mode") != "schema_guide" |
File History
1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a
init: musehub initial commit
Human
171 days ago