music-assistant-server

1"""Tests for TagFilterMiddleware enforcement on direct invocation (C3)."""
2# mypy: disable-error-code="arg-type, no-untyped-def, type-arg, assignment, operator, misc"
3
4from __future__ import annotations
5
6import pytest
7from fastmcp import Client, FastMCP
8from fastmcp.exceptions import ToolError
9from mcp.shared.exceptions import McpError
10
11from music_assistant.providers.fastmcp_server.middleware import TagFilterMiddleware
12from music_assistant.providers.fastmcp_server.server import build_tag_lookup
13
14
15def _build_server(allowed: set[str]) -> FastMCP:
16    """Construct a FastMCP root with one tagged tool and the tag-filter middleware."""
17    mcp: FastMCP = FastMCP(name="test-server")
18
19    @mcp.tool(tags={"query"})  # type: ignore[untyped-decorator, unused-ignore]
20    async def reads() -> str:
21        """Return a read-only result."""
22        return "ok"
23
24    @mcp.tool(tags={"delete"})  # type: ignore[untyped-decorator, unused-ignore]
25    async def deletes() -> str:
26        """Pretend to perform a destructive action."""
27        return "deleted"
28
29    @mcp.tool  # type: ignore[untyped-decorator, unused-ignore]
30    async def untagged() -> str:
31        """Return a value from an untagged tool — always exposed."""
32        return "untagged"
33
34    @mcp.resource("data://thing/{thing_id}", tags={"query"})  # type: ignore[untyped-decorator, unused-ignore]
35    async def thing(thing_id: str) -> str:
36        """Return a read-only resource value for the given id."""
37        return f"thing:{thing_id}"
38
39    @mcp.prompt(name="suggest", tags={"query"})  # type: ignore[untyped-decorator, unused-ignore]
40    def suggest() -> str:
41        """Return a sample prompt template."""
42        return "Pick something."
43
44    mcp.add_middleware(TagFilterMiddleware(lambda: allowed, build_tag_lookup(mcp)))
45    return mcp
46
47
48async def test_listing_filters_disabled_tools() -> None:
49    """A tool whose tags are all disabled doesn't appear in tools/list."""
50    mcp = _build_server(allowed={"query"})
51    async with Client(mcp) as client:
52        names = {t.name for t in await client.list_tools()}
53    assert "reads" in names
54    assert "untagged" in names
55    assert "deletes" not in names
56
57
58async def test_call_disabled_tool_blocked() -> None:
59    """
60    A client cannot bypass the listing filter by calling the disabled tool by name.
61
62    FastMCP's ``call_tool`` raises ``ToolError`` for any server-side rejection
63    (the middleware re-raises ``NotFoundError`` as a tool-call failure).
64    Pinning that type (rather than ``Exception``) keeps a future bug that
65    raises e.g. ``TypeError`` from a wrong call signature from being
66    silently masked.
67    """
68    mcp = _build_server(allowed={"query"})
69    async with Client(mcp) as client:
70        with pytest.raises(ToolError):
71            await client.call_tool("deletes", {})
72
73
74async def test_call_enabled_tool_works() -> None:
75    """An enabled tool runs normally with the middleware in place."""
76    mcp = _build_server(allowed={"query"})
77    async with Client(mcp) as client:
78        result = await client.call_tool("reads", {})
79    text_blocks = [c for c in result.content if hasattr(c, "text")]
80    assert any("ok" in c.text for c in text_blocks)
81
82
83async def test_untagged_tool_always_callable() -> None:
84    """Tools without tags are infrastructure and remain callable regardless of permissions."""
85    mcp = _build_server(allowed=set())
86    async with Client(mcp) as client:
87        result = await client.call_tool("untagged", {})
88    text_blocks = [c for c in result.content if hasattr(c, "text")]
89    assert any("untagged" in c.text for c in text_blocks)
90
91
92async def test_disabled_resource_blocked_on_read() -> None:
93    """
94    Reading a disabled resource by URI raises rather than silently succeeding.
95
96    ``read_resource`` lifts server errors to ``McpError`` (the MCP SDK's own
97    JSON-RPC error envelope class), not to ``ToolError`` — different transport
98    path from ``call_tool``.
99    """
100    mcp = _build_server(allowed=set())
101    async with Client(mcp) as client:
102        with pytest.raises(McpError):
103            await client.read_resource("data://thing/42")
104
105
106async def test_template_resource_read_via_concrete_uri() -> None:
107    """
108    A concrete URI matched by a template resource is readable when its tag is enabled.
109
110    The middleware lookup must fall back from ``get_resource`` (statically
111    registered URIs only) to ``get_resource_template`` (URI-template matching);
112    otherwise every ``@mcp.resource("scheme://{var}")``-backed URI gets blocked
113    as not-found even though the tag is enabled.
114    """
115    mcp = _build_server(allowed={"query"})
116    async with Client(mcp) as client:
117        contents = await client.read_resource("data://thing/42")
118    text_blocks = [c for c in contents if hasattr(c, "text")]
119    assert any("thing:42" in c.text for c in text_blocks)
120
121
122async def test_disabled_prompt_blocked_on_get() -> None:
123    """Getting a disabled prompt by name raises ``McpError`` (RPC envelope)."""
124    mcp = _build_server(allowed=set())
125    async with Client(mcp) as client:
126        with pytest.raises(McpError):
127            await client.get_prompt("suggest", {})
128