gabriel / musehub public
test_mcp_elicitation.py python
648 lines 25.1 KB
Raw
b7205dda feat: complete 100% coverage — elicitation bypass paths + ingest_push s… Gabriel Cardona <cgcardona@gmail.com> 2d 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_create_with_preferences: no session → elicitation_unavailable
19 - musehub_review_pr_interactive: no session → elicitation_unavailable
20 - musehub_connect_streaming_platform: no session → elicitation_unavailable
21 - musehub_connect_daw_cloud: no session → elicitation_unavailable
22 - musehub_create_release_interactive: no session → elicitation_unavailable
23
24 New prompts:
25 - musehub/onboard assembles correctly
26 - musehub/release_to_world assembles correctly with repo_id interpolation
27
28 SSE formatting:
29 - sse_event produces correct format
30 - sse_notification produces correct JSON-RPC notification
31 - sse_request produces correct JSON-RPC request with id
32 - sse_response produces correct JSON-RPC response
33 """
34 from __future__ import annotations
35
36 import asyncio
37 import json
38 from unittest.mock import AsyncMock, MagicMock, patch
39
40 import pytest
41
42 from musehub.mcp.context import ToolCallContext
43 from musehub.mcp.elicitation import (
44 SCHEMAS,
45 build_form_elicitation,
46 build_url_elicitation,
47 oauth_connect_url,
48 daw_cloud_connect_url,
49 )
50 from musehub.mcp.prompts import PROMPT_CATALOGUE, get_prompt
51 from musehub.mcp.session import (
52 MCPSession,
53 create_session,
54 create_pending_elicitation,
55 delete_session,
56 resolve_elicitation,
57 cancel_elicitation,
58 push_to_session,
59 )
60 from musehub.mcp.sse import (
61 sse_event,
62 sse_notification,
63 sse_request,
64 sse_response,
65 )
66
67
68 # ── SSE formatting ────────────────────────────────────────────────────────────
69
70
71 def test_sse_event_basic_format() -> None:
72 """sse_event should produce 'data: <json>\\n\\n'."""
73 result = sse_event({"jsonrpc": "2.0", "method": "ping"})
74 assert result.startswith("data:")
75 assert result.endswith("\n\n")
76 # Extract data line and parse JSON
77 data_line = [l for l in result.split("\n") if l.startswith("data:")][0]
78 payload = json.loads(data_line[len("data: "):])
79 assert payload["method"] == "ping"
80
81
82 def test_sse_event_with_id_and_type() -> None:
83 """sse_event with event_id and event_type should include id: and event: lines."""
84 result = sse_event({"a": 1}, event_id="42", event_type="notification")
85 assert "id: 42\n" in result
86 assert "event: notification\n" in result
87
88
89 def test_sse_notification_format() -> None:
90 """sse_notification should produce a valid JSON-RPC 2.0 notification."""
91 result = sse_notification("notifications/progress", {"progress": 50})
92 data_line = [l for l in result.split("\n") if l.startswith("data:")][0]
93 payload = json.loads(data_line[len("data: "):])
94 assert payload["jsonrpc"] == "2.0"
95 assert payload["method"] == "notifications/progress"
96 assert payload["params"]["progress"] == 50
97 assert "id" not in payload # notifications have no id
98
99
100 def test_sse_request_format() -> None:
101 """sse_request should produce a valid JSON-RPC 2.0 request with id."""
102 result = sse_request("elicit-1", "elicitation/create", {"mode": "form"})
103 data_line = [l for l in result.split("\n") if l.startswith("data:")][0]
104 payload = json.loads(data_line[len("data: "):])
105 assert payload["jsonrpc"] == "2.0"
106 assert payload["id"] == "elicit-1"
107 assert payload["method"] == "elicitation/create"
108 assert payload["params"]["mode"] == "form"
109
110
111 def test_sse_response_format() -> None:
112 """sse_response should produce a valid JSON-RPC 2.0 success response."""
113 result = sse_response(42, {"content": [{"type": "text", "text": "ok"}]})
114 data_line = [l for l in result.split("\n") if l.startswith("data:")][0]
115 payload = json.loads(data_line[len("data: "):])
116 assert payload["jsonrpc"] == "2.0"
117 assert payload["id"] == 42
118 assert "result" in payload
119 assert "error" not in payload
120
121
122 # ── Elicitation schemas ───────────────────────────────────────────────────────
123
124
125 def test_schemas_has_all_expected_keys() -> None:
126 """SCHEMAS must contain all 5 musical elicitation schemas."""
127 expected = {
128 "compose_preferences",
129 "repo_creation",
130 "pr_review_focus",
131 "release_metadata",
132 "platform_connect_confirm",
133 }
134 assert expected == set(SCHEMAS.keys())
135
136
137 def test_compose_preferences_schema_required_fields() -> None:
138 """compose_preferences schema must declare correct required fields."""
139 schema = SCHEMAS["compose_preferences"]
140 assert schema["type"] == "object"
141 required = schema["required"]
142 assert "key" in required
143 assert "tempo_bpm" in required
144 assert "mood" in required
145 assert "genre" in required
146
147
148 def test_build_form_elicitation() -> None:
149 """build_form_elicitation should return correct mode and requestedSchema."""
150 params = build_form_elicitation("compose_preferences", "Pick your vibe")
151 assert params["mode"] == "form"
152 assert params["message"] == "Pick your vibe"
153 assert "requestedSchema" in params
154 assert params["requestedSchema"] is SCHEMAS["compose_preferences"]
155
156
157 def test_build_form_elicitation_unknown_key_raises() -> None:
158 """build_form_elicitation with unknown key should raise KeyError."""
159 with pytest.raises(KeyError):
160 build_form_elicitation("nonexistent_schema", "message")
161
162
163 def test_build_url_elicitation() -> None:
164 """build_url_elicitation should return correct mode, url, and elicitationId."""
165 params, eid = build_url_elicitation("https://example.com/oauth", "Connect Spotify")
166 assert params["mode"] == "url"
167 assert params["url"] == "https://example.com/oauth"
168 assert params["message"] == "Connect Spotify"
169 assert params["elicitationId"] == eid
170 assert len(eid) > 10
171
172
173 def test_build_url_elicitation_stable_id() -> None:
174 """build_url_elicitation should use provided elicitation_id."""
175 params, eid = build_url_elicitation(
176 "https://example.com/oauth", "msg", elicitation_id="my-stable-id"
177 )
178 assert eid == "my-stable-id"
179 assert params["elicitationId"] == "my-stable-id"
180
181
182 def test_oauth_connect_url_format() -> None:
183 """oauth_connect_url should produce correct platform-specific MuseHub URL."""
184 url = oauth_connect_url("Spotify", "abc123", base_url="https://musehub.app")
185 assert "spotify" in url
186 assert "elicitation_id=abc123" in url
187 assert url.startswith("https://musehub.app")
188
189
190 def test_daw_cloud_connect_url_format() -> None:
191 """daw_cloud_connect_url should produce correct service-specific URL."""
192 url = daw_cloud_connect_url("LANDR", "xyz789", base_url="https://musehub.app")
193 assert "landr" in url
194 assert "elicitation_id=xyz789" in url
195
196
197 # ── ToolCallContext — elicit_form ─────────────────────────────────────────────
198
199
200 @pytest.mark.anyio
201 async def test_elicit_form_no_session_returns_none() -> None:
202 """elicit_form without an active session should return None."""
203 ctx = ToolCallContext(user_id=None, session=None)
204 result = await ctx.elicit_form(SCHEMAS["compose_preferences"], "msg")
205 assert result is None
206
207
208 @pytest.mark.anyio
209 async def test_elicit_form_accepted_returns_content() -> None:
210 """elicit_form should return content dict when user accepts."""
211 session = create_session("user-1", {"elicitation": {"form": {}}})
212 ctx = ToolCallContext(user_id="user-1", session=session)
213
214 # Pre-resolve the Future before the elicit_form call awaits it.
215 content = {"key": "C major", "tempo_bpm": 120, "mood": "peaceful", "genre": "ambient"}
216
217 async def _resolve_after_push() -> None:
218 await asyncio.sleep(0) # yield to let push happen
219 for req_id, fut in list(session.pending.items()):
220 resolve_elicitation(session, req_id, {"action": "accept", "content": content})
221
222 task = asyncio.create_task(_resolve_after_push())
223 result = await ctx.elicit_form(SCHEMAS["compose_preferences"], "Pick your vibe")
224 await task
225
226 assert result == content
227 delete_session(session.session_id)
228
229
230 @pytest.mark.anyio
231 async def test_elicit_form_declined_returns_none() -> None:
232 """elicit_form should return None when user declines."""
233 session = create_session("user-1", {"elicitation": {"form": {}}})
234 ctx = ToolCallContext(user_id="user-1", session=session)
235
236 async def _decline_after_push() -> None:
237 await asyncio.sleep(0)
238 for req_id in list(session.pending.keys()):
239 resolve_elicitation(session, req_id, {"action": "decline"})
240
241 task = asyncio.create_task(_decline_after_push())
242 result = await ctx.elicit_form(SCHEMAS["compose_preferences"], "Pick your vibe")
243 await task
244
245 assert result is None
246 delete_session(session.session_id)
247
248
249 @pytest.mark.anyio
250 async def test_elicit_form_no_form_capability_returns_none() -> None:
251 """elicit_form should return None if client didn't declare form support."""
252 session = create_session("user-1", {"elicitation": {"url": {}}}) # url only, no form
253 ctx = ToolCallContext(user_id="user-1", session=session)
254 result = await ctx.elicit_form(SCHEMAS["compose_preferences"], "msg")
255 assert result is None
256 delete_session(session.session_id)
257
258
259 # ── ToolCallContext — elicit_url ──────────────────────────────────────────────
260
261
262 @pytest.mark.anyio
263 async def test_elicit_url_no_session_returns_false() -> None:
264 """elicit_url without an active session should return False."""
265 ctx = ToolCallContext(user_id=None, session=None)
266 result = await ctx.elicit_url("https://example.com/oauth", "msg")
267 assert result is False
268
269
270 @pytest.mark.anyio
271 async def test_elicit_url_accepted_returns_true() -> None:
272 """elicit_url should return True when user accepts the URL flow."""
273 session = create_session("user-1", {"elicitation": {"form": {}, "url": {}}})
274 ctx = ToolCallContext(user_id="user-1", session=session)
275
276 async def _accept_after_push() -> None:
277 await asyncio.sleep(0)
278 for req_id in list(session.pending.keys()):
279 resolve_elicitation(session, req_id, {"action": "accept"})
280
281 task = asyncio.create_task(_accept_after_push())
282 result = await ctx.elicit_url("https://example.com/oauth", "msg")
283 await task
284
285 assert result is True
286 delete_session(session.session_id)
287
288
289 # ── ToolCallContext — progress ────────────────────────────────────────────────
290
291
292 @pytest.mark.anyio
293 async def test_progress_no_session_is_noop() -> None:
294 """progress without an active session should not raise."""
295 ctx = ToolCallContext(user_id=None, session=None)
296 await ctx.progress("token", 1, 10, "working…") # must not raise
297
298
299 @pytest.mark.anyio
300 async def test_progress_with_session_pushes_sse_event() -> None:
301 """progress with an active session should push a notifications/progress SSE event."""
302 session = create_session("user-1", {})
303 queue: asyncio.Queue[str | None] = asyncio.Queue()
304 session.sse_queues.append(queue)
305
306 ctx = ToolCallContext(user_id="user-1", session=session)
307 await ctx.progress("compose-token", 2, 5, "generating…")
308
309 assert not queue.empty()
310 event_text = queue.get_nowait()
311 assert event_text is not None
312 data_line = [l for l in event_text.split("\n") if l.startswith("data:")][0]
313 payload = json.loads(data_line[len("data: "):])
314 assert payload["method"] == "notifications/progress"
315 assert payload["params"]["progress"] == 2
316 assert payload["params"]["total"] == 5
317
318 delete_session(session.session_id)
319
320
321 # ── Elicitation tool executors — no session graceful degradation ──────────────
322
323
324 @pytest.mark.anyio
325 async def test_compose_with_preferences_no_session_no_prefs() -> None:
326 """musehub_create_with_preferences with no prefs + no session returns schema guide."""
327 from musehub.mcp.write_tools.elicitation_tools import execute_compose_with_preferences
328
329 ctx = ToolCallContext(user_id=None, session=None)
330 result = await execute_compose_with_preferences(repo_id=None, ctx=ctx)
331 assert result.ok is True
332 assert result.data is not None
333 assert result.data.get("mode") == "schema_guide"
334 assert "fields" in result.data
335
336
337 @pytest.mark.anyio
338 async def test_review_pr_interactive_no_session() -> None:
339 """musehub_review_pr_interactive without session must return error."""
340 from musehub.mcp.write_tools.elicitation_tools import execute_review_pr_interactive
341
342 ctx = ToolCallContext(user_id=None, session=None)
343 result = await execute_review_pr_interactive("repo-1", "pr-1", ctx=ctx)
344 # No bypass params + no session → schema guide (ok=True, not an error)
345 assert result.ok is True
346 assert result.data is not None
347 assert result.data.get("mode") == "schema_guide"
348
349
350 @pytest.mark.anyio
351 async def test_connect_streaming_platform_no_session_with_platform() -> None:
352 """musehub_connect_streaming_platform with platform + no session returns OAuth URL."""
353 from musehub.mcp.write_tools.elicitation_tools import execute_connect_streaming_platform
354
355 ctx = ToolCallContext(user_id=None, session=None)
356 result = await execute_connect_streaming_platform("Spotify", None, ctx=ctx)
357 assert result.ok is True
358 assert result.data is not None
359 assert result.data.get("status") == "pending_oauth"
360 assert "oauth_url" in result.data
361
362
363 @pytest.mark.anyio
364 async def test_connect_streaming_platform_no_session_no_platform() -> None:
365 """musehub_connect_streaming_platform with no platform + no session returns schema guide."""
366 from musehub.mcp.write_tools.elicitation_tools import execute_connect_streaming_platform
367
368 ctx = ToolCallContext(user_id=None, session=None)
369 result = await execute_connect_streaming_platform(None, None, ctx=ctx)
370 assert result.ok is True
371 assert result.data is not None
372 assert result.data.get("mode") == "schema_guide"
373 assert "platform_options" in result.data
374
375
376 @pytest.mark.anyio
377 async def test_connect_daw_cloud_no_session_with_service() -> None:
378 """musehub_connect_daw_cloud with service + no session returns OAuth URL."""
379 from musehub.mcp.write_tools.elicitation_tools import execute_connect_daw_cloud
380
381 ctx = ToolCallContext(user_id=None, session=None)
382 result = await execute_connect_daw_cloud("LANDR", ctx=ctx)
383 assert result.ok is True
384 assert result.data is not None
385 assert result.data.get("status") == "pending_oauth"
386 assert "oauth_url" in result.data
387
388
389 @pytest.mark.anyio
390 async def test_connect_daw_cloud_no_session_no_service() -> None:
391 """musehub_connect_daw_cloud with no service + no session returns schema guide."""
392 from musehub.mcp.write_tools.elicitation_tools import execute_connect_daw_cloud
393
394 ctx = ToolCallContext(user_id=None, session=None)
395 result = await execute_connect_daw_cloud(None, ctx=ctx)
396 assert result.ok is True
397 assert result.data is not None
398 assert result.data.get("mode") == "schema_guide"
399 assert "service_options" in result.data
400
401
402 @pytest.mark.anyio
403 async def test_create_release_interactive_no_session_no_params() -> None:
404 """musehub_create_release_interactive with no params + no session returns schema guide."""
405 from musehub.mcp.write_tools.elicitation_tools import execute_create_release_interactive
406
407 ctx = ToolCallContext(user_id=None, session=None)
408 result = await execute_create_release_interactive("repo-1", ctx=ctx)
409 assert result.ok is True
410 assert result.data is not None
411 assert result.data.get("mode") == "schema_guide"
412 assert "fields" in result.data
413
414
415 # ── New prompts ───────────────────────────────────────────────────────────────
416
417
418 def test_onboard_prompt_assembles() -> None:
419 """musehub/onboard should assemble with 2 messages."""
420 result = get_prompt("musehub/onboard", {"username": "alice"})
421 assert result is not None
422 assert "messages" in result
423 assert len(result["messages"]) == 2
424 assert result["messages"][0]["role"] == "user"
425 text = result["messages"][1]["content"]["text"]
426 assert "alice" in text
427 assert "elicitation" in text.lower() or "elicit" in text.lower() or "compose" in text.lower()
428
429
430 def test_release_to_world_prompt_assembles() -> None:
431 """musehub/release_to_world should assemble with 2 messages and interpolate repo_id."""
432 result = get_prompt("musehub/release_to_world", {"repo_id": "abc-123"})
433 assert result is not None
434 assert "messages" in result
435 assert len(result["messages"]) == 2
436 text = result["messages"][1]["content"]["text"]
437 assert "abc-123" in text
438
439
440 def test_onboard_prompt_in_catalogue() -> None:
441 """musehub/onboard must be in the prompt catalogue."""
442 names = {p["name"] for p in PROMPT_CATALOGUE}
443 assert "musehub/onboard" in names
444
445
446 def test_release_to_world_in_catalogue() -> None:
447 """musehub/release_to_world must be in the prompt catalogue."""
448 names = {p["name"] for p in PROMPT_CATALOGUE}
449 assert "musehub/release_to_world" in names
450
451
452 # ── Dispatcher routing — new notifications (2025-11-25) ──────────────────────
453
454
455 @pytest.mark.anyio
456 async def test_notifications_cancelled_handled() -> None:
457 """notifications/cancelled should be handled as a notification (return None)."""
458 from musehub.mcp.dispatcher import handle_request
459
460 resp = await handle_request({
461 "jsonrpc": "2.0",
462 "method": "notifications/cancelled",
463 "params": {"requestId": "elicit-1", "reason": "user navigated away"},
464 })
465 assert resp is None # notifications return None
466
467
468 @pytest.mark.anyio
469 async def test_notifications_elicitation_complete_handled() -> None:
470 """notifications/elicitation/complete should be handled as a notification."""
471 from musehub.mcp.dispatcher import handle_request
472
473 resp = await handle_request({
474 "jsonrpc": "2.0",
475 "method": "notifications/elicitation/complete",
476 "params": {"elicitationId": "abc-xyz"},
477 })
478 assert resp is None
479
480
481 @pytest.mark.anyio
482 async def test_notifications_cancelled_resolves_future() -> None:
483 """notifications/cancelled with a session should cancel the pending Future."""
484 from musehub.mcp.dispatcher import handle_request
485
486 session = create_session(None, {"elicitation": {"form": {}}})
487 fut = create_pending_elicitation(session, "elicit-99")
488
489 await handle_request(
490 {
491 "jsonrpc": "2.0",
492 "method": "notifications/cancelled",
493 "params": {"requestId": "elicit-99"},
494 },
495 session=session,
496 )
497
498 assert fut.cancelled()
499 delete_session(session.session_id)
500
501
502 # ── Bypass path tests (no session, params provided directly) ──────────────────
503
504
505 @pytest.mark.anyio
506 async def test_compose_with_preferences_bypass_minimal() -> None:
507 """Bypass: preferences dict produces a composition plan without session."""
508 from musehub.mcp.write_tools.elicitation_tools import execute_compose_with_preferences
509
510 ctx = ToolCallContext(user_id=None, session=None)
511 result = await execute_compose_with_preferences(
512 repo_id=None,
513 preferences={"key_signature": "G major", "tempo_bpm": 140, "mood": "joyful", "genre": "jazz"},
514 ctx=ctx,
515 )
516 assert result.ok is True
517 data = result.data
518 assert data is not None
519 # Plan is nested under "composition_plan"
520 plan = data.get("composition_plan") or data
521 assert plan.get("key") == "G major"
522 assert plan.get("tempo_bpm") == 140
523
524
525 @pytest.mark.anyio
526 async def test_compose_with_preferences_bypass_with_repo() -> None:
527 """Bypass: repo_id is injected into the plan and scaffold_hint is set."""
528 from musehub.mcp.write_tools.elicitation_tools import execute_compose_with_preferences
529
530 ctx = ToolCallContext(user_id=None, session=None)
531 result = await execute_compose_with_preferences(
532 repo_id="repo-xyz",
533 preferences={"mood": "ethereal", "genre": "ambient"},
534 ctx=ctx,
535 )
536 assert result.ok is True
537 assert result.data is not None
538 assert result.data.get("repo_id") == "repo-xyz"
539 assert "scaffold_hint" in result.data
540
541
542 @pytest.mark.anyio
543 async def test_compose_with_preferences_bypass_defaults() -> None:
544 """Bypass: empty preferences dict uses sensible defaults."""
545 from musehub.mcp.write_tools.elicitation_tools import execute_compose_with_preferences
546
547 ctx = ToolCallContext(user_id=None, session=None)
548 result = await execute_compose_with_preferences(repo_id=None, preferences={}, ctx=ctx)
549 assert result.ok is True
550 data = result.data
551 assert data is not None
552 plan = data.get("composition_plan") or data
553 assert plan.get("tempo_bpm") == 120
554 assert plan.get("key") == "C major"
555
556
557 @pytest.mark.anyio
558 async def test_review_pr_interactive_bypass_dimension_and_depth() -> None:
559 """Bypass: dimension+depth skip elicitation, run divergence analysis path."""
560 from musehub.mcp.write_tools.elicitation_tools import execute_review_pr_interactive
561 from musehub.services.musehub_mcp_executor import MusehubToolResult
562
563 ctx = ToolCallContext(user_id=None, session=None)
564
565 # DB is unavailable in unit tests — expect a graceful db_unavailable result,
566 # which proves the bypass path was taken (no elicitation_unavailable error).
567 result = await execute_review_pr_interactive(
568 "repo-1", "pr-1",
569 dimension="harmonic",
570 depth="thorough",
571 ctx=ctx,
572 )
573 # Accepted bypass — should proceed to DB lookup, not hit schema_guide
574 assert result.data is None or result.data.get("mode") != "schema_guide"
575 # If DB is down the error_code is db_unavailable, not elicitation_unavailable
576 if not result.ok:
577 assert result.error_code != "elicitation_unavailable"
578
579
580 @pytest.mark.anyio
581 async def test_review_pr_interactive_bypass_dimension_only() -> None:
582 """Bypass: dimension alone (depth defaults to standard) skips elicitation."""
583 from musehub.mcp.write_tools.elicitation_tools import execute_review_pr_interactive
584
585 ctx = ToolCallContext(user_id=None, session=None)
586 result = await execute_review_pr_interactive(
587 "repo-1", "pr-2",
588 dimension="melodic",
589 ctx=ctx,
590 )
591 # Must not be schema_guide — dimension was supplied so bypass took effect.
592 if result.data:
593 assert result.data.get("mode") != "schema_guide"
594 if not result.ok:
595 assert result.error_code != "elicitation_unavailable"
596
597
598 @pytest.mark.anyio
599 async def test_connect_streaming_platform_bypass_returns_oauth_url() -> None:
600 """Bypass: known platform + no session returns a usable OAuth URL."""
601 from musehub.mcp.write_tools.elicitation_tools import execute_connect_streaming_platform
602
603 ctx = ToolCallContext(user_id=None, session=None)
604 for platform in ["Spotify", "SoundCloud", "Bandcamp"]:
605 result = await execute_connect_streaming_platform(platform, None, ctx=ctx)
606 assert result.ok is True
607 assert result.data is not None
608 assert result.data.get("status") == "pending_oauth"
609 url = str(result.data.get("oauth_url", ""))
610 assert "http" in url
611 assert result.data.get("platform") == platform
612
613
614 @pytest.mark.anyio
615 async def test_connect_daw_cloud_bypass_returns_oauth_url() -> None:
616 """Bypass: known service + no session returns a usable OAuth URL."""
617 from musehub.mcp.write_tools.elicitation_tools import execute_connect_daw_cloud
618
619 ctx = ToolCallContext(user_id=None, session=None)
620 for service in ["LANDR", "Splice", "BandLab"]:
621 result = await execute_connect_daw_cloud(service, ctx=ctx)
622 assert result.ok is True
623 assert result.data is not None
624 assert result.data.get("status") == "pending_oauth"
625 url = str(result.data.get("oauth_url", ""))
626 assert "http" in url
627 assert "capabilities" in result.data
628
629
630 @pytest.mark.anyio
631 async def test_create_release_interactive_bypass_tag_only() -> None:
632 """Bypass: tag supplied directly — proceeds to execute_create_release (DB may fail)."""
633 from musehub.mcp.write_tools.elicitation_tools import execute_create_release_interactive
634
635 ctx = ToolCallContext(user_id=None, session=None)
636 result = await execute_create_release_interactive(
637 "repo-1",
638 tag="v0.9.0",
639 title="Beta",
640 notes="First beta.",
641 ctx=ctx,
642 )
643 # If DB is unavailable the create_release call fails, but it must NOT
644 # return elicitation_unavailable — that proves bypass was taken.
645 if not result.ok:
646 assert result.error_code != "elicitation_unavailable"
647 if result.data:
648 assert result.data.get("mode") != "schema_guide"