music-assistant-server

1"""
2End-to-end tests through the real ASGI bridge loop (C11).
3
4These tests exercise ``mount_into_mass`` against an aiohttp ``TestServer``
5hosting a hand-rolled ASGI app. They cover the bits that pure-helper unit
6tests can't reach: streaming chunk pass-through, DELETE / non-GET methods,
7the well-known endpoint living next to the MCP mount.
8"""
9# mypy: disable-error-code="arg-type, no-untyped-def, type-arg, assignment, operator, misc"
10
11from __future__ import annotations
12
13from types import SimpleNamespace
14from typing import Any
15
16import pytest
17from aiohttp.test_utils import TestClient, TestServer
18
19from music_assistant.providers.fastmcp_server.http_bridge import mount_into_mass, mount_well_known
20
21from .conftest import FakeWebserver, build_aiohttp_app
22
23
24async def _lifespan_loop(receive: Any, send: Any) -> None:
25    """Bare-minimum ASGI lifespan handler used by the test ASGI doubles."""
26    while True:
27        msg = await receive()
28        if msg["type"] == "lifespan.startup":
29            await send({"type": "lifespan.startup.complete"})
30        elif msg["type"] == "lifespan.shutdown":
31            await send({"type": "lifespan.shutdown.complete"})
32            return
33
34
35async def _streaming_asgi(scope: dict, receive: Any, send: Any) -> None:
36    """ASGI app that emits three SSE-style chunks before closing the body."""
37    if scope.get("type") == "lifespan":
38        await _lifespan_loop(receive, send)
39        return
40    # Drain the request body so the client write side can complete.
41    while True:
42        msg = await receive()
43        if msg.get("type") == "http.request" and not msg.get("more_body"):
44            break
45    await send(
46        {
47            "type": "http.response.start",
48            "status": 200,
49            "headers": [(b"content-type", b"text/event-stream")],
50        }
51    )
52    for i in range(3):
53        await send(
54            {"type": "http.response.body", "body": f"event{i}\n".encode(), "more_body": True}
55        )
56    await send({"type": "http.response.body", "body": b"", "more_body": False})
57
58
59async def _method_echo_asgi(scope: dict, receive: Any, send: Any) -> None:
60    """ASGI app that echoes the HTTP method in the body."""
61    if scope.get("type") == "lifespan":
62        await _lifespan_loop(receive, send)
63        return
64    while True:
65        msg = await receive()
66        if msg.get("type") == "http.request" and not msg.get("more_body"):
67            break
68    method = scope["method"].encode()
69    await send({"type": "http.response.start", "status": 200, "headers": []})
70    await send({"type": "http.response.body", "body": method})
71
72
73class _Mcp:
74    def __init__(self, asgi: Any) -> None:
75        self._asgi = asgi
76
77    def http_app(self, transport: str = "streamable-http", path: str = "/mcp") -> Any:
78        return self._asgi
79
80
81@pytest.fixture
82async def streaming_client() -> Any:
83    """Bridge an SSE-streaming ASGI app through mount_into_mass."""
84    ws = FakeWebserver()
85    mass = SimpleNamespace(webserver=ws)
86    await mount_into_mass(mass, _Mcp(_streaming_asgi), mount_path="/mcp/v1")
87    async with TestClient(TestServer(build_aiohttp_app(ws))) as client:
88        yield client
89
90
91@pytest.fixture
92async def method_echo_client() -> Any:
93    """Bridge a method-echo ASGI app to verify DELETE / arbitrary verbs work."""
94    ws = FakeWebserver()
95    mass = SimpleNamespace(webserver=ws)
96    await mount_into_mass(mass, _Mcp(_method_echo_asgi), mount_path="/mcp/v1")
97    async with TestClient(TestServer(build_aiohttp_app(ws))) as client:
98        yield client
99
100
101async def test_streaming_chunks_passed_through(streaming_client: TestClient) -> None:
102    """Three ASGI body chunks reach the aiohttp client unbuffered."""
103    resp = await streaming_client.post("/mcp/v1/", headers={"Origin": "http://localhost:8095"})
104    assert resp.status == 200
105    body = await resp.read()
106    assert body == b"event0\nevent1\nevent2\n"
107
108
109async def test_delete_method_reaches_asgi(method_echo_client: TestClient) -> None:
110    """Streamable-HTTP DELETE (session terminate) is forwarded — bridge does not 405."""
111    resp = await method_echo_client.delete("/mcp/v1/", headers={"Origin": "http://localhost:8095"})
112    assert resp.status == 200
113    assert (await resp.read()) == b"DELETE"
114
115
116async def test_get_method_reaches_asgi(method_echo_client: TestClient) -> None:
117    """
118    GET is forwarded to the ASGI app — not rejected at the bridge with a 405.
119
120    This pins the *bridge-level* guarantee only: the verb reaches the mounted
121    app instead of being short-circuited. It does not exercise the real FastMCP
122    app or auth (the fixture is a method-echo double), so it cannot observe the
123    server's actual GET status.
124
125    Why the bridge must forward GET: OpenClaw's bundle-mcp client opens the
126    optional ``GET`` SSE stream *before* ``POST initialize`` and bails if that
127    GET is non-2xx (OpenClaw issue #72757 — it 405s against POST-only servers).
128    The live server answering GET with 401 (no token) / SSE (valid token)
129    rather than 405 was verified manually against a running instance; it is not
130    asserted here.
131    """
132    resp = await method_echo_client.get("/mcp/v1/", headers={"Origin": "http://localhost:8095"})
133    assert resp.status == 200
134    assert (await resp.read()) == b"GET"
135
136
137async def test_bare_mount_path_without_trailing_slash_reaches_asgi(
138    method_echo_client: TestClient,
139) -> None:
140    """
141    ``/mcp/v1`` (no trailing slash) must hit the ASGI bridge too.
142
143    This is the URL the wizard advertises and that MCP clients connect to.
144    MA's real ``_handle_catch_all`` (``helpers/webserver.py``) matches a
145    ``"/mcp/v1/*"`` registration against both the bare stem and any
146    descendant; ``build_aiohttp_app`` must mirror that.
147    """
148    resp = await method_echo_client.post("/mcp/v1", headers={"Origin": "http://localhost:8095"})
149    assert resp.status == 200
150    assert (await resp.read()) == b"POST"
151
152
153async def test_well_known_alongside_mcp_mount() -> None:
154    """Both /mcp/v1/* and /.well-known/oauth-protected-resource are reachable."""
155    ws = FakeWebserver()
156    mass = SimpleNamespace(webserver=ws)
157    await mount_into_mass(mass, _Mcp(_method_echo_asgi), mount_path="/mcp/v1")
158    await mount_well_known(
159        mass,
160        mount_path="/mcp/v1",
161        resource_uri="http://localhost:8095/mcp/v1",
162        authorization_servers=["http://localhost:8095"],
163        scopes_supported=["query:library"],
164        resource_name="Music Assistant MCP",
165    )
166    async with TestClient(TestServer(build_aiohttp_app(ws))) as client:
167        # MCP endpoint reachable
168        resp = await client.post("/mcp/v1/", headers={"Origin": "http://localhost:8095"})
169        assert resp.status == 200
170        # well-known sub-path returns RFC 9728 metadata
171        meta = await client.get("/.well-known/oauth-protected-resource/mcp/v1")
172        assert meta.status == 200
173        doc = await meta.json()
174        assert doc["resource"] == "http://localhost:8095/mcp/v1"
175        assert doc["authorization_servers"] == ["http://localhost:8095"]
176        # Well-known root form also works (RFC 9728 §3.1 fallback).
177        meta_root = await client.get("/.well-known/oauth-protected-resource")
178        assert meta_root.status == 200
179