gabriel / musehub public
test_mcp_elicitation_section16.py python
534 lines 21.3 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago
1 """Section 16 — MCP Elicitation: 7-layer test suite.
2
3 Covers gaps not addressed by the 43 tests in test_mcp_elicitation.py and the
4 14 tests in test_stress_elicitation_bypass.py.
5
6 New coverage:
7
8 Layer 1 Unit:
9 - build_form_elicitation: all available schema keys produce correct shape
10 - build_url_elicitation: caller-supplied elicitation_id preserved
11 - ElicitationRequest TypedDict has required fields (mode, message)
12 - ElicitationAction TypedDict has required action field
13 - AVAILABLE_PLATFORMS and AVAILABLE_DAW_CLOUDS are non-empty
14 - compose_preferences schema has all expected properties
15
16 Layer 2 Integration (session elicitation primitives):
17 - create_pending_elicitation stores Future in session.pending
18 - resolve_elicitation matching ID → Future set, returns True
19 - resolve_elicitation non-matching ID → returns False
20 - resolve_elicitation already-done Future → returns False
21 - _signal_elicitation_complete resolves matching Future across sessions
22 - _signal_elicitation_complete pushes SSE notification to queue
23 - _signal_elicitation_complete with no matching session → returns 0
24 - delete_session cancels all pending elicitation Futures
25
26 Layer 3 E2E (HTTP UI elicitation routes):
27 - GET /mcp/elicitation/{id}/callback?status=accepted → 200 HTML
28 - GET /mcp/elicitation/{id}/callback?status=declined → 200 HTML
29 - GET /mcp/connect/{invalid_slug} → 404
30 - GET /mcp/connect/daw/{invalid_slug} → 404
31 - GET /mcp/connect/{valid_slug} unauthenticated → 302 redirect
32 - GET /mcp/connect/daw/{valid_slug} unauthenticated → 302 redirect
33
34 Layer 4 Stress:
35 - 50 pending elicitations in one session, all created without collision
36 - _signal_elicitation_complete resolves across 5 simultaneous sessions
37
38 Layer 5 Data Integrity:
39 - Accepted content preserved in Future result
40 - Declined action stored (action="decline", no content key in result)
41 - resolve_elicitation stores exact result dict
42 - build_form_elicitation requestedSchema matches SCHEMAS entry exactly
43
44 Layer 6 Security:
45 - elicitation_callback with unknown elicitation_id → 200 (no crash)
46 - platform_connect_start unknown platform → 404 (not 500)
47 - Double resolve → second call returns False (can't resolve twice)
48 - execute_review_proposal_interactive with malformed args → graceful
49
50 Layer 7 Performance:
51 - 1000× build_form_elicitation under 50 ms
52 - 1000× build_url_elicitation under 50 ms
53 - 100 pending elicitations create+resolve under 100 ms
54 """
55 from __future__ import annotations
56
57 import asyncio
58 import time
59 from typing import get_type_hints
60
61 import pytest
62 import pytest_asyncio
63 from httpx import AsyncClient, ASGITransport
64 from unittest.mock import patch
65 from sqlalchemy.ext.asyncio import AsyncSession
66
67 from musehub.muse_contracts.mcp_types import ElicitationAction, ElicitationRequest
68 from musehub.main import app
69 from musehub.mcp.elicitation import (
70 AVAILABLE_DAW_CLOUDS,
71 AVAILABLE_PLATFORMS,
72 SCHEMAS,
73 build_form_elicitation,
74 build_url_elicitation,
75 daw_cloud_connect_url,
76 oauth_connect_url,
77 )
78 from musehub.mcp.session import (
79 MCPSession,
80 create_pending_elicitation,
81 create_session,
82 delete_session,
83 push_to_session,
84 resolve_elicitation,
85 )
86
87
88 # ── Fixtures ──────────────────────────────────────────────────────────────────
89
90
91 @pytest.fixture
92 def anyio_backend() -> str:
93 return "asyncio"
94
95
96 @pytest_asyncio.fixture
97 async def http_client(db_session: AsyncSession) -> AsyncClient:
98 async with AsyncClient(
99 transport=ASGITransport(app=app),
100 base_url="http://localhost",
101 ) as c:
102 yield c
103
104
105 # ── Layer 1 — Unit ────────────────────────────────────────────────────────────
106
107
108 class TestUnitSchemaBuilders:
109 def test_build_form_all_schema_keys(self) -> None:
110 """build_form_elicitation must succeed for every key in SCHEMAS."""
111 for key in SCHEMAS:
112 result = build_form_elicitation(key, message=f"Test: {key}")
113 assert result["mode"] == "form", f"mode wrong for key={key}"
114 assert result["message"] == f"Test: {key}"
115 assert "requestedSchema" in result, f"missing requestedSchema for key={key}"
116
117 def test_build_form_schema_matches_schemas_dict(self) -> None:
118 """requestedSchema must be the exact SCHEMAS[key] dict."""
119 for key in SCHEMAS:
120 result = build_form_elicitation(key, message="m")
121 assert result["requestedSchema"] is SCHEMAS[key]
122
123 def test_build_url_caller_supplied_id_preserved(self) -> None:
124 params, eid = build_url_elicitation("https://example.com", "Connect", elicitation_id="stable-id")
125 assert eid == "stable-id"
126 assert params["elicitationId"] == "stable-id"
127
128 def test_build_url_auto_generates_id_when_omitted(self) -> None:
129 params, eid1 = build_url_elicitation("https://a.com", "m1")
130 params2, eid2 = build_url_elicitation("https://a.com", "m2")
131 assert eid1 != eid2 # unique IDs
132 assert len(eid1) > 8
133
134 def test_build_url_mode_and_url_set(self) -> None:
135 params, _ = build_url_elicitation("https://musehub.ai/connect", "msg")
136 assert params["mode"] == "url"
137 assert params["url"] == "https://musehub.ai/connect"
138 assert params["message"] == "msg"
139
140
141 class TestUnitElicitationTypesDicts:
142 def test_elicitation_action_has_action_field(self) -> None:
143 action: ElicitationAction = {"action": "accept"}
144 assert action["action"] == "accept"
145
146 def test_elicitation_action_with_content(self) -> None:
147 action: ElicitationAction = {"action": "accept", "content": {"key": "C major"}}
148 assert action["content"]["key"] == "C major"
149
150 def test_elicitation_request_form_shape(self) -> None:
151 req: ElicitationRequest = {
152 "mode": "form",
153 "message": "Pick preferences",
154 "requestedSchema": {"type": "object"},
155 }
156 assert req["mode"] == "form"
157 assert "requestedSchema" in req
158
159 def test_elicitation_request_url_shape(self) -> None:
160 req: ElicitationRequest = {
161 "mode": "url",
162 "message": "Connect platform",
163 "url": "https://musehub.ai/connect/spotify",
164 "elicitationId": "eid-123",
165 }
166 assert req["url"] == "https://musehub.ai/connect/spotify"
167
168
169 class TestUnitAvailableLists:
170 def test_available_platforms_non_empty(self) -> None:
171 assert len(AVAILABLE_PLATFORMS) > 0
172
173 def test_available_daw_clouds_non_empty(self) -> None:
174 assert len(AVAILABLE_DAW_CLOUDS) > 0
175
176 def test_platforms_are_strings(self) -> None:
177 assert all(isinstance(p, str) for p in AVAILABLE_PLATFORMS)
178
179 def test_daw_clouds_are_strings(self) -> None:
180 assert all(isinstance(d, str) for d in AVAILABLE_DAW_CLOUDS)
181
182
183 class TestUnitComposePreferencesSchema:
184 def test_schema_is_object_type(self) -> None:
185 schema = SCHEMAS["compose_preferences"]
186 assert schema["type"] == "object"
187
188 def test_schema_has_key_property(self) -> None:
189 props = SCHEMAS["compose_preferences"]["properties"]
190 assert "key" in props
191
192 def test_schema_has_tempo_bpm_property(self) -> None:
193 props = SCHEMAS["compose_preferences"]["properties"]
194 assert "tempo_bpm" in props
195
196 def test_tempo_bpm_has_min_max_constraints(self) -> None:
197 tempo = SCHEMAS["compose_preferences"]["properties"]["tempo_bpm"]
198 assert tempo["minimum"] >= 1
199 assert tempo["maximum"] <= 500
200
201
202 # ── Layer 2 — Integration ─────────────────────────────────────────────────────
203
204
205 class TestIntegrationPendingElicitation:
206 @pytest.mark.anyio
207 async def test_create_pending_stores_future(self) -> None:
208 s = create_session(None, {"elicitation": {"form": {}}})
209 try:
210 fut = create_pending_elicitation(s, "req-1")
211 assert "req-1" in s.pending
212 assert s.pending["req-1"] is fut
213 assert not fut.done()
214 finally:
215 delete_session(s.session_id)
216
217 @pytest.mark.anyio
218 async def test_resolve_matching_id_returns_true(self) -> None:
219 s = create_session(None, {"elicitation": {"form": {}}})
220 try:
221 fut = create_pending_elicitation(s, "req-resolve")
222 result = resolve_elicitation(s, "req-resolve", {"action": "accept", "content": {"key": "D minor"}})
223 assert result is True
224 assert fut.done()
225 assert fut.result() == {"action": "accept", "content": {"key": "D minor"}}
226 finally:
227 delete_session(s.session_id)
228
229 @pytest.mark.anyio
230 async def test_resolve_non_matching_id_returns_false(self) -> None:
231 s = create_session(None, {"elicitation": {"form": {}}})
232 try:
233 create_pending_elicitation(s, "req-real")
234 result = resolve_elicitation(s, "req-wrong", {"action": "accept"})
235 assert result is False
236 finally:
237 delete_session(s.session_id)
238
239 @pytest.mark.anyio
240 async def test_resolve_already_done_future_returns_false(self) -> None:
241 s = create_session(None, {"elicitation": {"form": {}}})
242 try:
243 create_pending_elicitation(s, "req-done")
244 resolve_elicitation(s, "req-done", {"action": "accept"})
245 # Second resolve should fail.
246 result = resolve_elicitation(s, "req-done", {"action": "accept"})
247 assert result is False
248 finally:
249 delete_session(s.session_id)
250
251
252 class TestIntegrationSignalElicitation:
253 @pytest.mark.anyio
254 async def test_signal_resolves_matching_future(self) -> None:
255 from musehub.api.routes.musehub.ui_mcp_elicitation import _signal_elicitation_complete
256
257 s = create_session(None, {"elicitation": {"form": {}}})
258 try:
259 fut = create_pending_elicitation(s, "sig-id-1")
260 resolved = _signal_elicitation_complete("sig-id-1", action="accept")
261 assert resolved >= 1
262 assert fut.done()
263 assert fut.result()["action"] == "accept"
264 finally:
265 delete_session(s.session_id)
266
267 @pytest.mark.anyio
268 async def test_signal_pushes_sse_notification_to_queue(self) -> None:
269 from musehub.api.routes.musehub.ui_mcp_elicitation import _signal_elicitation_complete
270
271 s = create_session(None, {"elicitation": {"form": {}}})
272 try:
273 queue: asyncio.Queue[str | None] = asyncio.Queue()
274 s.sse_queues.append(queue)
275 create_pending_elicitation(s, "sig-sse-1")
276
277 _signal_elicitation_complete("sig-sse-1", action="accept")
278
279 item = queue.get_nowait()
280 assert item is not None
281 assert "notifications/elicitation/complete" in item
282 finally:
283 delete_session(s.session_id)
284
285 @pytest.mark.anyio
286 async def test_signal_no_matching_session_returns_zero(self) -> None:
287 from musehub.api.routes.musehub.ui_mcp_elicitation import _signal_elicitation_complete
288
289 # Signal an ID that no session has pending.
290 resolved = _signal_elicitation_complete("completely-unknown-elicitation-id")
291 assert resolved == 0
292
293 @pytest.mark.anyio
294 async def test_delete_session_cancels_pending_futures(self) -> None:
295 s = create_session(None, {"elicitation": {"form": {}}})
296 sid = s.session_id
297 fut1 = create_pending_elicitation(s, "cancel-1")
298 fut2 = create_pending_elicitation(s, "cancel-2")
299
300 delete_session(sid)
301
302 assert fut1.cancelled()
303 assert fut2.cancelled()
304
305
306 # ── Layer 3 — End-to-End ──────────────────────────────────────────────────────
307
308
309 class TestE2EElicitationCallbackRoute:
310 @pytest.mark.anyio
311 async def test_callback_accepted_returns_200(
312 self, http_client: AsyncClient, db_session: AsyncSession
313 ) -> None:
314 resp = await http_client.get(
315 "/mcp/elicitation/test-eid-accepted/callback?status=accepted"
316 )
317 assert resp.status_code == 200
318
319 @pytest.mark.anyio
320 async def test_callback_declined_returns_200(
321 self, http_client: AsyncClient, db_session: AsyncSession
322 ) -> None:
323 resp = await http_client.get(
324 "/mcp/elicitation/test-eid-declined/callback?status=declined"
325 )
326 assert resp.status_code == 200
327
328
329 class TestE2EPlatformConnectRoute:
330 @pytest.mark.anyio
331 async def test_invalid_platform_slug_returns_404(
332 self, http_client: AsyncClient, db_session: AsyncSession
333 ) -> None:
334 resp = await http_client.get(
335 "/mcp/connect/totally-made-up-platform?elicitation_id=eid-xyz"
336 )
337 assert resp.status_code == 404
338
339 @pytest.mark.anyio
340 async def test_unauthenticated_valid_platform_redirects(
341 self, http_client: AsyncClient, db_session: AsyncSession
342 ) -> None:
343 # Use first available platform slug.
344 from musehub.api.routes.musehub.ui_mcp_elicitation import _PLATFORM_BY_SLUG
345 if not _PLATFORM_BY_SLUG:
346 pytest.skip("No platforms configured")
347 slug = next(iter(_PLATFORM_BY_SLUG))
348 # Patch _get_musehub_user_id to return None (unauthenticated) since the
349 # test client has no SessionMiddleware installed.
350 with patch(
351 "musehub.api.routes.musehub.ui_mcp_elicitation._get_musehub_user_id",
352 return_value=None,
353 ):
354 resp = await http_client.get(
355 f"/mcp/connect/{slug}?elicitation_id=eid-auth-test",
356 follow_redirects=False,
357 )
358 # Unauthenticated → redirect to login page.
359 assert resp.status_code in (302, 303)
360 assert "login" in resp.headers.get("location", "").lower()
361
362
363 class TestE2EDawConnectRoute:
364 @pytest.mark.anyio
365 async def test_invalid_daw_slug_returns_404(
366 self, http_client: AsyncClient, db_session: AsyncSession
367 ) -> None:
368 resp = await http_client.get(
369 "/mcp/connect/daw/totally-unknown-daw?elicitation_id=eid-daw"
370 )
371 assert resp.status_code == 404
372
373 @pytest.mark.anyio
374 async def test_unauthenticated_valid_daw_redirects(
375 self, http_client: AsyncClient, db_session: AsyncSession
376 ) -> None:
377 from musehub.api.routes.musehub.ui_mcp_elicitation import _DAW_BY_SLUG
378 if not _DAW_BY_SLUG:
379 pytest.skip("No DAW services configured")
380 slug = next(iter(_DAW_BY_SLUG))
381 # Patch _get_musehub_user_id to return None (unauthenticated) since the
382 # test client has no SessionMiddleware installed.
383 with patch(
384 "musehub.api.routes.musehub.ui_mcp_elicitation._get_musehub_user_id",
385 return_value=None,
386 ):
387 resp = await http_client.get(
388 f"/mcp/connect/daw/{slug}?elicitation_id=eid-daw-test",
389 follow_redirects=False,
390 )
391 assert resp.status_code in (302, 303)
392 assert "login" in resp.headers.get("location", "").lower()
393
394
395 # ── Layer 4 — Stress ──────────────────────────────────────────────────────────
396
397
398 class TestStressElicitation:
399 @pytest.mark.anyio
400 async def test_50_pending_elicitations_no_collision(self) -> None:
401 s = create_session(None, {"elicitation": {"form": {}}})
402 try:
403 futs = []
404 for i in range(50):
405 fut = create_pending_elicitation(s, f"elicit-{i}")
406 futs.append((i, fut))
407 # All 50 should be distinct, non-done futures.
408 assert len(s.pending) == 50
409 assert all(not fut.done() for _, fut in futs)
410 finally:
411 delete_session(s.session_id)
412
413 @pytest.mark.anyio
414 async def test_signal_resolves_across_5_sessions(self) -> None:
415 from musehub.api.routes.musehub.ui_mcp_elicitation import _signal_elicitation_complete
416
417 sessions = [create_session(None, {"elicitation": {"form": {}}}) for _ in range(5)]
418 futs = []
419 try:
420 for s in sessions:
421 fut = create_pending_elicitation(s, "shared-eid")
422 futs.append(fut)
423
424 resolved = _signal_elicitation_complete("shared-eid", action="accept")
425 assert resolved == 5
426 assert all(fut.done() for fut in futs)
427 finally:
428 for s in sessions:
429 delete_session(s.session_id)
430
431
432 # ── Layer 5 — Data Integrity ──────────────────────────────────────────────────
433
434
435 class TestDataIntegrityElicitation:
436 @pytest.mark.anyio
437 async def test_accepted_content_preserved_in_result(self) -> None:
438 s = create_session(None, {"elicitation": {"form": {}}})
439 try:
440 fut = create_pending_elicitation(s, "di-accept")
441 payload = {"action": "accept", "content": {"key": "G major", "tempo_bpm": 140}}
442 resolve_elicitation(s, "di-accept", payload)
443 assert fut.result() == payload
444 finally:
445 delete_session(s.session_id)
446
447 @pytest.mark.anyio
448 async def test_declined_action_in_result(self) -> None:
449 s = create_session(None, {"elicitation": {"form": {}}})
450 try:
451 fut = create_pending_elicitation(s, "di-decline")
452 resolve_elicitation(s, "di-decline", {"action": "decline"})
453 assert fut.result()["action"] == "decline"
454 assert "content" not in fut.result()
455 finally:
456 delete_session(s.session_id)
457
458 def test_build_form_requested_schema_is_schemas_entry(self) -> None:
459 for key in list(SCHEMAS.keys())[:5]: # check first 5 to keep test fast
460 params = build_form_elicitation(key, message="m")
461 assert params["requestedSchema"] == SCHEMAS[key]
462
463 def test_build_url_elicitation_id_in_params(self) -> None:
464 params, eid = build_url_elicitation("https://example.com", "msg")
465 assert params["elicitationId"] == eid
466
467
468 # ── Layer 6 — Security ────────────────────────────────────────────────────────
469
470
471 class TestSecurityElicitation:
472 @pytest.mark.anyio
473 async def test_callback_unknown_id_does_not_crash(
474 self, http_client: AsyncClient, db_session: AsyncSession
475 ) -> None:
476 """elicitation_callback with unknown ID must return 200, not 500."""
477 resp = await http_client.get(
478 "/mcp/elicitation/completely-unknown-id/callback?status=accepted"
479 )
480 assert resp.status_code == 200
481
482 @pytest.mark.anyio
483 async def test_platform_unknown_slug_returns_404_not_500(
484 self, http_client: AsyncClient, db_session: AsyncSession
485 ) -> None:
486 resp = await http_client.get("/mcp/connect/injected<script>?elicitation_id=x")
487 # Must not be a 500; either 404 or 422 (validation).
488 assert resp.status_code in (400, 404, 422)
489
490 @pytest.mark.anyio
491 async def test_double_resolve_returns_false(self) -> None:
492 """Resolving the same elicitation twice must fail on the second call."""
493 s = create_session(None, {"elicitation": {"form": {}}})
494 try:
495 create_pending_elicitation(s, "double-resolve")
496 r1 = resolve_elicitation(s, "double-resolve", {"action": "accept"})
497 r2 = resolve_elicitation(s, "double-resolve", {"action": "accept"})
498 assert r1 is True
499 assert r2 is False
500 finally:
501 delete_session(s.session_id)
502
503 # ── Layer 7 — Performance ─────────────────────────────────────────────────────
504
505
506 class TestPerformanceElicitation:
507 def test_1000_build_form_under_50ms(self) -> None:
508 keys = list(SCHEMAS.keys())
509 start = time.perf_counter()
510 for i in range(1000):
511 build_form_elicitation(keys[i % len(keys)], message="perf test")
512 elapsed_ms = (time.perf_counter() - start) * 1000
513 assert elapsed_ms < 50, f"1000× build_form_elicitation took {elapsed_ms:.1f} ms"
514
515 def test_1000_build_url_under_50ms(self) -> None:
516 start = time.perf_counter()
517 for i in range(1000):
518 build_url_elicitation(f"https://example.com/flow/{i}", "Connect")
519 elapsed_ms = (time.perf_counter() - start) * 1000
520 assert elapsed_ms < 50, f"1000× build_url_elicitation took {elapsed_ms:.1f} ms"
521
522 @pytest.mark.anyio
523 async def test_100_pending_create_and_resolve_under_100ms(self) -> None:
524 s = create_session(None, {"elicitation": {"form": {}}})
525 try:
526 start = time.perf_counter()
527 for i in range(100):
528 create_pending_elicitation(s, f"perf-{i}")
529 for i in range(100):
530 resolve_elicitation(s, f"perf-{i}", {"action": "accept"})
531 elapsed_ms = (time.perf_counter() - start) * 1000
532 assert elapsed_ms < 100, f"100 create+resolve took {elapsed_ms:.1f} ms"
533 finally:
534 delete_session(s.session_id)
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago